opendaylight/netconf/netconf-client/src/main/java/org/opendaylight/controller/netconf/client/SimpleNetconfClientSessionListener.java

   1 /*
   2  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.controller.netconf.client;
  10
  11 import com.google.common.base.Preconditions;
  12 import io.netty.util.concurrent.Future;
  13 import io.netty.util.concurrent.GlobalEventExecutor;
  14 import io.netty.util.concurrent.Promise;
  15 import java.util.ArrayDeque;
  16 import java.util.Queue;
  17 import javax.annotation.concurrent.GuardedBy;
  18 import org.opendaylight.controller.netconf.api.NetconfMessage;
  19 import org.opendaylight.controller.netconf.api.NetconfTerminationReason;
  20 import org.slf4j.Logger;
  21 import org.slf4j.LoggerFactory;
  22
  23 public class SimpleNetconfClientSessionListener implements NetconfClientSessionListener {
  24     private static final class RequestEntry {
  25         private final Promise<NetconfMessage> promise;
  26         private final NetconfMessage request;
  27
  28         public RequestEntry(Promise<NetconfMessage> future, NetconfMessage request) {
  29             this.promise = Preconditions.checkNotNull(future);
  30             this.request = Preconditions.checkNotNull(request);
  31         }
  32     }
  33
  34     private static final Logger LOG = LoggerFactory.getLogger(SimpleNetconfClientSessionListener.class);
  35
  36     @GuardedBy("this")
  37     private final Queue<RequestEntry> requests = new ArrayDeque<>();
  38
  39     @GuardedBy("this")
  40     private NetconfClientSession clientSession;
  41
  42     @GuardedBy("this")
  43     private void dispatchRequest() {
  44         while (!requests.isEmpty()) {
  45             final RequestEntry e = requests.peek();
  46             if (e.promise.setUncancellable()) {
  47                 LOG.debug("Sending message {}", e.request);
  48                 clientSession.sendMessage(e.request);
  49                 break;
  50             }
  51
  52             LOG.debug("Message {} has been cancelled, skipping it", e.request);
  53             requests.poll();
  54         }
  55     }
  56
  57     @Override
  58     public final synchronized void onSessionUp(NetconfClientSession clientSession) {
  59         this.clientSession = Preconditions.checkNotNull(clientSession);
  60         LOG.debug("Client session {} went up", clientSession);
  61         dispatchRequest();
  62     }
  63
  64     private synchronized void tearDown(final Exception cause) {
  65         final RequestEntry e = requests.poll();
  66         if (e != null) {
  67             e.promise.setFailure(cause);
  68         }
  69
  70         this.clientSession = null;
  71     }
  72
  73     @Override
  74     public final void onSessionDown(NetconfClientSession clientSession, Exception e) {
  75         LOG.debug("Client Session {} went down unexpectedly", clientSession, e);
  76         tearDown(e);
  77     }
  78
  79     @Override
  80     public final void onSessionTerminated(NetconfClientSession clientSession,
  81             NetconfTerminationReason netconfTerminationReason) {
  82         LOG.debug("Client Session {} terminated, reason: {}", clientSession,
  83                 netconfTerminationReason.getErrorMessage());
  84         tearDown(new RuntimeException(netconfTerminationReason.getErrorMessage()));
  85     }
  86
  87     @Override
  88     public synchronized void onMessage(NetconfClientSession session, NetconfMessage message) {
  89         LOG.debug("New message arrived: {}", message);
  90
  91         final RequestEntry e = requests.poll();
  92         if (e != null) {
  93             e.promise.setSuccess(message);
  94             dispatchRequest();
  95         } else {
  96             LOG.info("Ignoring unsolicited message {}", message);
  97         }
  98     }
  99
 100     public final synchronized Future<NetconfMessage> sendRequest(NetconfMessage message) {
 101         final RequestEntry req = new RequestEntry(GlobalEventExecutor.INSTANCE.<NetconfMessage>newPromise(), message);
 102
 103         requests.add(req);
 104         if (clientSession != null) {
 105             dispatchRequest();
 106         }
 107
 108         return req.promise;
 109     }
 110 }