netconf/netconf-client/src/main/java/org/opendaylight/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 package org.opendaylight.netconf.client;
   9
  10 import com.google.common.base.Preconditions;
  11 import io.netty.util.concurrent.Future;
  12 import io.netty.util.concurrent.GlobalEventExecutor;
  13 import io.netty.util.concurrent.Promise;
  14 import java.util.ArrayDeque;
  15 import java.util.Queue;
  16 import org.checkerframework.checker.lock.qual.GuardedBy;
  17 import org.checkerframework.checker.lock.qual.Holding;
  18 import org.opendaylight.netconf.api.NetconfMessage;
  19 import org.opendaylight.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         RequestEntry(final Promise<NetconfMessage> future, final 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     @Holding("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.remove();
  54         }
  55     }
  56
  57     @Override
  58     @SuppressWarnings("checkstyle:hiddenField")
  59     public final synchronized void onSessionUp(final NetconfClientSession clientSession) {
  60         this.clientSession = Preconditions.checkNotNull(clientSession);
  61         LOG.debug("Client session {} went up", clientSession);
  62         dispatchRequest();
  63     }
  64
  65     private synchronized void tearDown(final Exception cause) {
  66         final RequestEntry e = requests.poll();
  67         if (e != null) {
  68             e.promise.setFailure(cause);
  69         }
  70
  71         this.clientSession = null;
  72     }
  73
  74     @Override
  75     @SuppressWarnings("checkstyle:hiddenField")
  76     public final void onSessionDown(final NetconfClientSession clientSession, final Exception exception) {
  77         LOG.debug("Client Session {} went down unexpectedly", clientSession, exception);
  78         tearDown(exception);
  79     }
  80
  81     @Override
  82     @SuppressWarnings("checkstyle:hiddenField")
  83     public final void onSessionTerminated(final NetconfClientSession clientSession,
  84                                           final NetconfTerminationReason netconfTerminationReason) {
  85         LOG.debug("Client Session {} terminated, reason: {}", clientSession,
  86                 netconfTerminationReason.getErrorMessage());
  87         tearDown(new RuntimeException(netconfTerminationReason.getErrorMessage()));
  88     }
  89
  90     @Override
  91     public synchronized void onMessage(final NetconfClientSession session, final NetconfMessage message) {
  92         LOG.debug("New message arrived: {}", message);
  93
  94         final RequestEntry e = requests.poll();
  95         if (e != null) {
  96             e.promise.setSuccess(message);
  97             dispatchRequest();
  98         } else {
  99             LOG.info("Ignoring unsolicited message {}", message);
 100         }
 101     }
 102
 103     public final synchronized Future<NetconfMessage> sendRequest(final NetconfMessage message) {
 104         final RequestEntry req = new RequestEntry(GlobalEventExecutor.INSTANCE.newPromise(), message);
 105
 106         requests.add(req);
 107         if (clientSession != null) {
 108             dispatchRequest();
 109         }
 110
 111         return req.promise;
 112     }
 113 }