X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fmd-sal%2Fsal-netconf-connector%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fsal%2Fconnect%2Fnetconf%2Flistener%2FNetconfDeviceCommunicator.java;h=4f2f6ab38ea204a45b02473b55130900d9c627c3;hp=3871cdfa4f4cb7556313d8419407e1d3326dc802;hb=1e884647502a8d91f8a57bde8193c60b9bbcce0d;hpb=5a9287bb6ddaaa8805939b3b3301f648c03785f4 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 3871cdfa4f..4f2f6ab38e 100644 --- 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 remoteDevice; - private final Optional overrideNetconfCapabilities; + private final RemoteDevice remoteDevice; + private final Optional overrideNetconfCapabilities; private final RemoteDeviceId id; private final Lock sessionLock = new ReentrantLock(); + // TODO implement concurrent message limit private final Queue requests = new ArrayDeque<>(); private NetconfClientSession session; + private Future initFuture; - public NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice remoteDevice, - final NetconfSessionCapabilities netconfSessionCapabilities) { - this(id, remoteDevice, Optional.of(netconfSessionCapabilities)); + public NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice remoteDevice, + final NetconfSessionPreferences NetconfSessionPreferences) { + this(id, remoteDevice, Optional.of(NetconfSessionPreferences)); } public NetconfDeviceCommunicator(final RemoteDeviceId id, - final RemoteDevice remoteDevice) { - this(id, remoteDevice, Optional.absent()); + final RemoteDevice remoteDevice) { + this(id, remoteDevice, Optional.absent()); } - private NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice remoteDevice, - final Optional overrideNetconfCapabilities) { + private NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice remoteDevice, + final Optional 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.addModuleCaps(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>(){ + + @Override + public void operationComplete(Future future) throws Exception { + if (!future.isSuccess() && !future.isCancelled()) { + 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 createSessionDownRpcResult() - { + private RpcResult createSessionDownRpcResult() { return createErrorRpcResult( RpcError.ErrorType.TRANSPORT, String.format( "The netconf session to %1$s is disconnected", id.getName() ) ); } - private RpcResult createErrorRpcResult( RpcError.ErrorType errorType, String message ) - { + private RpcResult createErrorRpcResult( RpcError.ErrorType errorType, String message ) { return RpcResultBuilder.failed() - .withError( errorType, NetconfDocumentedException.ErrorTag.operation_failed.getTagValue(), - message ) - .build(); + .withError(errorType, NetconfDocumentedException.ErrorTag.operation_failed.getTagValue(), message).build(); } @Override @@ -172,7 +190,15 @@ 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(); + // tear down not necessary, called indirectly by above close + } } @Override @@ -191,12 +217,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 +236,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.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 +274,11 @@ public class NetconfDeviceCommunicator implements NetconfClientSessionListener, } @Override - public ListenableFuture> sendRequest( - final NetconfMessage message, final QName rpc) { + public ListenableFuture> sendRequest(final NetconfMessage message, final QName rpc) { sessionLock.lock(); try { return sendRequestWithLock( message, rpc ); - } - finally { + } finally { sessionLock.unlock(); } }