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=8553820b40095755d4ee82361d184cbf5590fc9e;hp=8045f8cb4a9556160b44fb46d4d3bbc12374435a;hb=bd8beb1bfee9f421ad8f2d07b1424b21038234a2;hpb=79202e1fd05d2606b35e163f608fad9cce84b5d4 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 8045f8cb4a..8553820b40 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 @@ -7,13 +7,20 @@ */ package org.opendaylight.controller.sal.connect.netconf.listener; +import com.google.common.base.Optional; +import com.google.common.base.Strings; +import com.google.common.collect.Lists; +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; import java.util.Queue; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; - import org.opendaylight.controller.netconf.api.NetconfDocumentedException; import org.opendaylight.controller.netconf.api.NetconfMessage; import org.opendaylight.controller.netconf.api.NetconfTerminationReason; @@ -36,31 +43,37 @@ import org.opendaylight.yangtools.yang.common.RpcResultBuilder; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import com.google.common.base.Strings; -import com.google.common.collect.Lists; -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; - public class NetconfDeviceCommunicator implements NetconfClientSessionListener, RemoteDeviceCommunicator { private static final Logger logger = LoggerFactory.getLogger(NetconfDeviceCommunicator.class); - private final RemoteDevice remoteDevice; + 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 NetconfSessionPreferences NetconfSessionPreferences) { + this(id, remoteDevice, Optional.of(NetconfSessionPreferences)); + } + public NetconfDeviceCommunicator(final RemoteDeviceId id, - final RemoteDevice remoteDevice) { + final RemoteDevice remoteDevice) { + this(id, remoteDevice, Optional.absent()); + } + + private NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice remoteDevice, + final Optional overrideNetconfCapabilities) { this.id = id; this.remoteDevice = remoteDevice; + this.overrideNetconfCapabilities = overrideNetconfCapabilities; } - private final Queue requests = new ArrayDeque<>(); - private NetconfClientSession session; - @Override public void onSessionUp(final NetconfClientSession session) { sessionLock.lock(); @@ -68,23 +81,47 @@ public class NetconfDeviceCommunicator implements NetconfClientSessionListener, logger.debug("{}: Session established", id); this.session = session; - final NetconfSessionCapabilities netconfSessionCapabilities = - NetconfSessionCapabilities.fromNetconfSession(session); - logger.trace("{}: Session advertised capabilities: {}", id, netconfSessionCapabilities); + NetconfSessionPreferences netconfSessionPreferences = + NetconfSessionPreferences.fromNetconfSession(session); + logger.trace("{}: Session advertised capabilities: {}", id, netconfSessionPreferences); - remoteDevice.onRemoteSessionUp(netconfSessionCapabilities, this); + if(overrideNetconfCapabilities.isPresent()) { + netconfSessionPreferences = netconfSessionPreferences.replaceModuleCaps(overrideNetconfCapabilities.get()); + logger.debug("{}: Session capabilities overridden, capabilities that will be used: {}", id, netconfSessionPreferences); + } + + 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()) { + logger.debug("{}: Connection failed", id, future.cause()); + NetconfDeviceCommunicator.this.remoteDevice.onRemoteSessionFailed(future.cause()); + } + } + }); + + } + + public void disconnect() { + if(session != null) { + session.close(); } } @@ -129,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 @@ -157,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 @@ -176,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)); } @@ -195,14 +237,12 @@ 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 ); @@ -213,8 +253,7 @@ public class NetconfDeviceCommunicator implements NetconfClientSessionListener, 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 ); @@ -223,7 +262,7 @@ public class NetconfDeviceCommunicator implements NetconfClientSessionListener, return; } - request.future.set( RpcResultBuilder.success( message ).build() ); + request.future.set( RpcResultBuilder.success( message ).build() ); } } @@ -232,13 +271,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(); } }