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=c62f56ac1ebd29b2c9eabf065a3512ec24ba60b2;hp=4da727f5c24e56e37c4e11acd6150d9afded1cd8;hb=2e7af5f8377352ecdae3bace51417ee286dc6796;hpb=81bbe76bd26399118d028663d08e464ce6b7d040 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 4da727f5c2..c62f56ac1e 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,14 +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.Collections; 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; @@ -22,47 +28,52 @@ import org.opendaylight.controller.netconf.api.xml.XmlNetconfConstants; import org.opendaylight.controller.netconf.client.NetconfClientDispatcher; import org.opendaylight.controller.netconf.client.NetconfClientSession; import org.opendaylight.controller.netconf.client.NetconfClientSessionListener; +import org.opendaylight.controller.netconf.client.conf.NetconfClientConfiguration; import org.opendaylight.controller.netconf.client.conf.NetconfReconnectingClientConfiguration; import org.opendaylight.controller.netconf.util.xml.XmlElement; import org.opendaylight.controller.netconf.util.xml.XmlUtil; -import org.opendaylight.controller.sal.common.util.RpcErrors; -import org.opendaylight.controller.sal.common.util.Rpcs; import org.opendaylight.controller.sal.connect.api.RemoteDevice; import org.opendaylight.controller.sal.connect.api.RemoteDeviceCommunicator; import org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil; -import org.opendaylight.controller.sal.connect.util.FailedRpcResult; import org.opendaylight.controller.sal.connect.util.RemoteDeviceId; import org.opendaylight.yangtools.yang.common.QName; import org.opendaylight.yangtools.yang.common.RpcError; import org.opendaylight.yangtools.yang.common.RpcResult; +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(); @@ -70,20 +81,48 @@ 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); + + if(overrideNetconfCapabilities.isPresent()) { + 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 NetconfReconnectingClientConfiguration config) { - dispatch.createReconnectingClient(config); + public void initializeRemoteConnection(final NetconfClientDispatcher dispatcher, final NetconfClientConfiguration config) { + // TODO 2313 extract listener from configuration + if(config instanceof NetconfReconnectingClientConfiguration) { + initFuture = dispatcher.createReconnectingClient((NetconfReconnectingClientConfiguration) config); + } else { + 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(); + } } private void tearDown( String reason ) { @@ -127,17 +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 ) - { - return new FailedRpcResult( RpcErrors.getRpcError( null, - NetconfDocumentedException.ErrorTag.operation_failed.getTagValue(), - null, RpcError.ErrorSeverity.ERROR, message, errorType, null ) ); + private RpcResult createErrorRpcResult( RpcError.ErrorType errorType, String message ) { + return RpcResultBuilder.failed() + .withError(errorType, NetconfDocumentedException.ErrorTag.operation_failed.getTagValue(), message).build(); } @Override @@ -154,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 @@ -173,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)); } @@ -192,35 +237,36 @@ 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( new FailedRpcResult( - NetconfMessageTransformUtil.toRpcError( 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 ); - request.future.set( new FailedRpcResult( - NetconfMessageTransformUtil.toRpcError( e ) ) ); + request.future.set( RpcResultBuilder.failed() + .withRpcError( NetconfMessageTransformUtil.toRpcError( e ) ).build() ); return; } - request.future.set(Rpcs.getRpcResult( true, message, Collections.emptySet() ) ); + request.future.set( RpcResultBuilder.success( message ).build() ); } } @@ -229,13 +275,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(); } }