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 static java.util.Objects.requireNonNull;
  11
  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 org.checkerframework.checker.lock.qual.GuardedBy;
  18 import org.checkerframework.checker.lock.qual.Holding;
  19 import org.opendaylight.netconf.api.NetconfMessage;
  20 import org.opendaylight.netconf.api.NetconfTerminationReason;
  21 import org.slf4j.Logger;
  22 import org.slf4j.LoggerFactory;
  23
  24 public class SimpleNetconfClientSessionListener implements NetconfClientSessionListener {
  25     private static final class RequestEntry {
  26         private final Promise<NetconfMessage> promise;
  27         private final NetconfMessage request;
  28
  29         RequestEntry(final Promise<NetconfMessage> future, final NetconfMessage request) {
  30             this.promise = requireNonNull(future);
  31             this.request = requireNonNull(request);
  32         }
  33     }
  34
  35     private static final Logger LOG = LoggerFactory.getLogger(SimpleNetconfClientSessionListener.class);
  36
  37     @GuardedBy("this")
  38     private final Queue<RequestEntry> requests = new ArrayDeque<>();
  39
  40     @GuardedBy("this")
  41     private NetconfClientSession clientSession;
  42
  43     @Holding("this")
  44     private void dispatchRequest() {
  45         while (!requests.isEmpty()) {
  46             final RequestEntry e = requests.peek();
  47             if (e.promise.setUncancellable()) {
  48                 LOG.debug("Sending message {}", e.request);
  49                 clientSession.sendMessage(e.request);
  50                 break;
  51             }
  52
  53             LOG.debug("Message {} has been cancelled, skipping it", e.request);
  54             requests.remove();
  55         }
  56     }
  57
  58     @Override
  59     @SuppressWarnings("checkstyle:hiddenField")
  60     public final synchronized void onSessionUp(final NetconfClientSession clientSession) {
  61         this.clientSession = requireNonNull(clientSession);
  62         LOG.debug("Client session {} went up", clientSession);
  63         dispatchRequest();
  64     }
  65
  66     private synchronized void tearDown(final Exception cause) {
  67         final RequestEntry e = requests.poll();
  68         if (e != null) {
  69             e.promise.setFailure(cause);
  70         }
  71
  72         this.clientSession = null;
  73     }
  74
  75     @Override
  76     @SuppressWarnings("checkstyle:hiddenField")
  77     public final void onSessionDown(final NetconfClientSession clientSession, final Exception exception) {
  78         LOG.debug("Client Session {} went down unexpectedly", clientSession, exception);
  79         tearDown(exception);
  80     }
  81
  82     @Override
  83     @SuppressWarnings("checkstyle:hiddenField")
  84     public final void onSessionTerminated(final NetconfClientSession clientSession,
  85                                           final NetconfTerminationReason netconfTerminationReason) {
  86         LOG.debug("Client Session {} terminated, reason: {}", clientSession,
  87                 netconfTerminationReason.getErrorMessage());
  88         tearDown(new RuntimeException(netconfTerminationReason.getErrorMessage()));
  89     }
  90
  91     @Override
  92     public synchronized void onMessage(final NetconfClientSession session, final NetconfMessage message) {
  93         LOG.debug("New message arrived: {}", message);
  94
  95         final RequestEntry e = requests.poll();
  96         if (e != null) {
  97             e.promise.setSuccess(message);
  98             dispatchRequest();
  99         } else {
 100             LOG.info("Ignoring unsolicited message {}", message);
 101         }
 102     }
 103
 104     public final synchronized Future<NetconfMessage> sendRequest(final NetconfMessage message) {
 105         final RequestEntry req = new RequestEntry(GlobalEventExecutor.INSTANCE.newPromise(), message);
 106
 107         requests.add(req);
 108         if (clientSession != null) {
 109             dispatchRequest();
 110         }
 111
 112         return req.promise;
 113     }
 114 }