Bug 2697: Improvement wrong response handling, missing message

[controller.git] / opendaylight / md-sal / sal-netconf-connector / src / main / java / org / opendaylight / controller / sal / connect / netconf / listener / NetconfDeviceCommunicator.java

diff --git a/opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/listener/NetconfDeviceCommunicator.java b/opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/listener/NetconfDeviceCommunicator.java
index 3871cdfa4f4cb7556313d8419407e1d3326dc802..c62f56ac1ebd29b2c9eabf065a3512ec24ba60b2 100644 (file)
--- a/opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/listener/NetconfDeviceCommunicator.java
+++ b/opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/listener/NetconfDeviceCommunicator.java
@@ -14,6 +14,7 @@ import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import io.netty.util.concurrent.Future;
 import io.netty.util.concurrent.FutureListener;
+import io.netty.util.concurrent.GenericFutureListener;
 import java.util.ArrayDeque;
 import java.util.Iterator;
 import java.util.List;
@@ -46,26 +47,28 @@ public class NetconfDeviceCommunicator implements NetconfClientSessionListener,
 
     private static final Logger logger = LoggerFactory.getLogger(NetconfDeviceCommunicator.class);
 
-    private final RemoteDevice<NetconfSessionCapabilities, NetconfMessage> remoteDevice;
-    private final Optional<NetconfSessionCapabilities> overrideNetconfCapabilities;
+    private final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> remoteDevice;
+    private final Optional<NetconfSessionPreferences> overrideNetconfCapabilities;
     private final RemoteDeviceId id;
     private final Lock sessionLock = new ReentrantLock();
 
+    // TODO implement concurrent message limit
     private final Queue<Request> requests = new ArrayDeque<>();
     private NetconfClientSession session;
+    private Future<?> initFuture;
 
-    public NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice<NetconfSessionCapabilities, NetconfMessage> remoteDevice,
-            final NetconfSessionCapabilities netconfSessionCapabilities) {
-        this(id, remoteDevice, Optional.of(netconfSessionCapabilities));
+    public NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> remoteDevice,
+            final NetconfSessionPreferences NetconfSessionPreferences) {
+        this(id, remoteDevice, Optional.of(NetconfSessionPreferences));
     }
 
     public NetconfDeviceCommunicator(final RemoteDeviceId id,
-                                     final RemoteDevice<NetconfSessionCapabilities, NetconfMessage> remoteDevice) {
-        this(id, remoteDevice, Optional.<NetconfSessionCapabilities>absent());
+                                     final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> remoteDevice) {
+        this(id, remoteDevice, Optional.<NetconfSessionPreferences>absent());
     }
 
-    private NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice<NetconfSessionCapabilities, NetconfMessage> remoteDevice,
-            final Optional<NetconfSessionCapabilities> overrideNetconfCapabilities) {
+    private NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> remoteDevice,
+            final Optional<NetconfSessionPreferences> overrideNetconfCapabilities) {
         this.id = id;
         this.remoteDevice = remoteDevice;
         this.overrideNetconfCapabilities = overrideNetconfCapabilities;
@@ -78,28 +81,47 @@ public class NetconfDeviceCommunicator implements NetconfClientSessionListener,
             logger.debug("{}: Session established", id);
             this.session = session;
 
-            NetconfSessionCapabilities netconfSessionCapabilities =
-                                             NetconfSessionCapabilities.fromNetconfSession(session);
-            logger.trace("{}: Session advertised capabilities: {}", id, netconfSessionCapabilities);
+            NetconfSessionPreferences netconfSessionPreferences =
+                                             NetconfSessionPreferences.fromNetconfSession(session);
+            logger.trace("{}: Session advertised capabilities: {}", id, netconfSessionPreferences);
 
             if(overrideNetconfCapabilities.isPresent()) {
-                netconfSessionCapabilities = netconfSessionCapabilities.replaceModuleCaps(overrideNetconfCapabilities.get());
-                logger.debug("{}: Session capabilities overridden, capabilities that will be used: {}", id, netconfSessionCapabilities);
+                netconfSessionPreferences = netconfSessionPreferences.replaceModuleCaps(overrideNetconfCapabilities.get());
+                logger.debug("{}: Session capabilities overridden, capabilities that will be used: {}", id, netconfSessionPreferences);
             }
 
-            remoteDevice.onRemoteSessionUp(netconfSessionCapabilities, this);
+            remoteDevice.onRemoteSessionUp(netconfSessionPreferences, this);
         }
         finally {
             sessionLock.unlock();
         }
     }
 
-    public void initializeRemoteConnection(final NetconfClientDispatcher dispatch,
-                                           final NetconfClientConfiguration config) {
+    public void initializeRemoteConnection(final NetconfClientDispatcher dispatcher, final NetconfClientConfiguration config) {
+        // TODO 2313 extract listener from configuration
         if(config instanceof NetconfReconnectingClientConfiguration) {
-            dispatch.createReconnectingClient((NetconfReconnectingClientConfiguration) config);
+            initFuture = dispatcher.createReconnectingClient((NetconfReconnectingClientConfiguration) config);
         } else {
-            dispatch.createClient(config);
+            initFuture = dispatcher.createClient(config);
+        }
+
+
+        initFuture.addListener(new GenericFutureListener<Future<Object>>(){
+
+            @Override
+            public void operationComplete(Future<Object> future) throws Exception {
+                if (!future.isSuccess()) {
+                    logger.debug("{}: Connection failed", id, future.cause());
+                    NetconfDeviceCommunicator.this.remoteDevice.onRemoteSessionFailed(future.cause());
+                }
+            }
+        });
+
+    }
+
+    public void disconnect() {
+        if(session != null) {
+            session.close();
         }
     }
 
@@ -144,18 +166,14 @@ public class NetconfDeviceCommunicator implements NetconfClientSessionListener,
         }
     }
 
-    private RpcResult<NetconfMessage> createSessionDownRpcResult()
-    {
+    private RpcResult<NetconfMessage> createSessionDownRpcResult() {
         return createErrorRpcResult( RpcError.ErrorType.TRANSPORT,
                              String.format( "The netconf session to %1$s is disconnected", id.getName() ) );
     }
 
-    private RpcResult<NetconfMessage> createErrorRpcResult( RpcError.ErrorType errorType, String message )
-    {
+    private RpcResult<NetconfMessage> createErrorRpcResult( RpcError.ErrorType errorType, String message ) {
         return RpcResultBuilder.<NetconfMessage>failed()
-                .withError( errorType, NetconfDocumentedException.ErrorTag.operation_failed.getTagValue(),
-                            message )
-                .build();
+                .withError(errorType, NetconfDocumentedException.ErrorTag.operation_failed.getTagValue(), message).build();
     }
 
     @Override
@@ -172,7 +190,16 @@ public class NetconfDeviceCommunicator implements NetconfClientSessionListener,
 
     @Override
     public void close() {
-        tearDown( String.format( "The netconf session to %1$s has been closed", id.getName() ) );
+        // Cancel reconnect if in progress
+        if(initFuture != null) {
+            initFuture.cancel(false);
+        }
+        // Disconnect from device
+        if(session != null) {
+            session.close();
+        }
+
+        tearDown(id + ": Netconf session closed");
     }
 
     @Override
@@ -191,12 +218,12 @@ public class NetconfDeviceCommunicator implements NetconfClientSessionListener,
     private void processMessage(final NetconfMessage message) {
         Request request = null;
         sessionLock.lock();
+
         try {
             request = requests.peek();
-            if (request.future.isUncancellable()) {
+            if (request != null && request.future.isUncancellable()) {
                 requests.poll();
-            }
-            else {
+            } else {
                 request = null;
                 logger.warn("{}: Ignoring unsolicited message {}", id, msgToS(message));
             }
@@ -210,26 +237,27 @@ public class NetconfDeviceCommunicator implements NetconfClientSessionListener,
             logger.debug("{}: Message received {}", id, message);
 
             if(logger.isTraceEnabled()) {
-                logger.trace( "{}: Matched request: {} to response: {}", id,
-                              msgToS( request.request ), msgToS( message ) );
+                logger.trace( "{}: Matched request: {} to response: {}", id, msgToS( request.request ), msgToS( message ) );
             }
 
             try {
                 NetconfMessageTransformUtil.checkValidReply( request.request, message );
-            }
-            catch (final NetconfDocumentedException e) {
+            } catch (final NetconfDocumentedException e) {
                 logger.warn( "{}: Invalid request-reply match, reply message contains different message-id, request: {}, response: {}",
                              id, msgToS( request.request ), msgToS( message ), e );
 
                 request.future.set( RpcResultBuilder.<NetconfMessage>failed()
                         .withRpcError( NetconfMessageTransformUtil.toRpcError( e ) ).build() );
+
+                //recursively processing message to eventually find matching request
+                processMessage(message);
+
                 return;
             }
 
             try {
                 NetconfMessageTransformUtil.checkSuccessReply(message);
-            }
-            catch( NetconfDocumentedException e ) {
+            } catch(final NetconfDocumentedException e) {
                 logger.warn( "{}: Error reply from remote device, request: {}, response: {}", id,
                              msgToS( request.request ), msgToS( message ), e );
 
@@ -247,13 +275,11 @@ public class NetconfDeviceCommunicator implements NetconfClientSessionListener,
     }
 
     @Override
-    public ListenableFuture<RpcResult<NetconfMessage>> sendRequest(
-                                               final NetconfMessage message, final QName rpc) {
+    public ListenableFuture<RpcResult<NetconfMessage>> sendRequest(final NetconfMessage message, final QName rpc) {
         sessionLock.lock();
         try {
             return sendRequestWithLock( message, rpc );
-        }
-        finally {
+        } finally {
             sessionLock.unlock();
         }
     }