*/
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;
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> 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> remoteDevice,
+ final NetconfSessionPreferences netconfSessionPreferences) {
+ this(id, remoteDevice, Optional.of(netconfSessionPreferences));
+ }
+
public NetconfDeviceCommunicator(final RemoteDeviceId id,
- final RemoteDevice<NetconfSessionCapabilities, NetconfMessage> remoteDevice) {
+ final RemoteDevice<NetconfSessionPreferences, NetconfMessage> remoteDevice) {
+ this(id, remoteDevice, Optional.<NetconfSessionPreferences>absent());
+ }
+
+ private NetconfDeviceCommunicator(final RemoteDeviceId id, final RemoteDevice<NetconfSessionPreferences, NetconfMessage> 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 NetconfClientConfiguration config) {
if(config instanceof NetconfReconnectingClientConfiguration) {
- dispatch.createReconnectingClient((NetconfReconnectingClientConfiguration) config);
+ initFuture = dispatch.createReconnectingClient((NetconfReconnectingClientConfiguration) config);
} else {
- dispatch.createClient(config);
+ initFuture = dispatch.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());
+ }
+ }
+ });
}
private void tearDown( String reason ) {
@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));
}
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 );
return;
}
- request.future.set( RpcResultBuilder.<NetconfMessage>success( message ).build() );
+ request.future.set( RpcResultBuilder.success( message ).build() );
}
}