*/
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;
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<NetconfMessage> {
private static final Logger logger = LoggerFactory.getLogger(NetconfDeviceCommunicator.class);
- private final RemoteDevice<NetconfSessionCapabilities, NetconfMessage> remoteDevice;
+ 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<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> remoteDevice,
+ final NetconfSessionPreferences NetconfSessionPreferences) {
+ this(id, remoteDevice, Optional.of(NetconfSessionPreferences));
+ }
+
public NetconfDeviceCommunicator(final RemoteDeviceId id,
- final RemoteDevice<NetconfSessionCapabilities, NetconfMessage> remoteDevice) {
+ final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> remoteDevice) {
+ this(id, remoteDevice, Optional.<NetconfSessionPreferences>absent());
+ }
+
+ private NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> remoteDevice,
+ final Optional<NetconfSessionPreferences> overrideNetconfCapabilities) {
this.id = id;
this.remoteDevice = remoteDevice;
+ this.overrideNetconfCapabilities = overrideNetconfCapabilities;
}
- private final Queue<Request> requests = new ArrayDeque<>();
- private NetconfClientSession session;
-
@Override
public void onSessionUp(final NetconfClientSession session) {
sessionLock.lock();
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<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();
+ }
}
private void tearDown( String reason ) {
}
}
- 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 )
- {
- return new FailedRpcResult<NetconfMessage>( RpcErrors.getRpcError( null,
- NetconfDocumentedException.ErrorTag.operation_failed.getTagValue(),
- null, RpcError.ErrorSeverity.ERROR, message, errorType, null ) );
+ private RpcResult<NetconfMessage> createErrorRpcResult( RpcError.ErrorType errorType, String message ) {
+ return RpcResultBuilder.<NetconfMessage>failed()
+ .withError(errorType, NetconfDocumentedException.ErrorTag.operation_failed.getTagValue(), message).build();
}
@Override
@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
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));
}
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<NetconfMessage>(
- NetconfMessageTransformUtil.toRpcError( 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 );
- request.future.set( new FailedRpcResult<NetconfMessage>(
- NetconfMessageTransformUtil.toRpcError( e ) ) );
+ request.future.set( RpcResultBuilder.<NetconfMessage>failed()
+ .withRpcError( NetconfMessageTransformUtil.toRpcError( e ) ).build() );
return;
}
- request.future.set(Rpcs.getRpcResult( true, message, Collections.<RpcError>emptySet() ) );
+ request.future.set( RpcResultBuilder.success( message ).build() );
}
}
}
@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();
}
}
Code Review / controller.git / blobdiff