import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.SettableFuture;
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;
private NetconfClientSession session;
private Future<?> initFuture;
- private SettableFuture<NetconfDeviceCapabilities> firstConnectionFuture;
+ private final SettableFuture<NetconfDeviceCapabilities> firstConnectionFuture;
// isSessionClosing indicates a close operation on the session is issued and
// tearDown will surely be called later to finish the close.
// Used to allow only one thread to enter tearDown and other threads should
// NOT enter it simultaneously and should end its close operation without
// calling tearDown to release the locks they hold to avoid deadlock.
- private volatile AtomicBoolean isSessionClosing = new AtomicBoolean(false);
+ private final AtomicBoolean isSessionClosing = new AtomicBoolean(false);
public Boolean isSessionClosing() {
return isSessionClosing.get();
LOG.trace("{}: Session advertised capabilities: {}", id,
netconfSessionPreferences);
- if(overrideNetconfCapabilities.isPresent()) {
- netconfSessionPreferences = overrideNetconfCapabilities.get().isOverride() ?
- netconfSessionPreferences.replaceModuleCaps(overrideNetconfCapabilities.get().getSessionPreferences()) :
- netconfSessionPreferences.addModuleCaps(overrideNetconfCapabilities.get().getSessionPreferences());
- LOG.debug(
- "{}: Session capabilities overridden, capabilities that will be used: {}",
- id, netconfSessionPreferences);
+ if (overrideNetconfCapabilities.isPresent()) {
+ final NetconfSessionPreferences sessionPreferences = overrideNetconfCapabilities
+ .get().getSessionPreferences();
+ netconfSessionPreferences = overrideNetconfCapabilities.get().moduleBasedCapsOverrided()
+ ? netconfSessionPreferences.replaceModuleCaps(sessionPreferences)
+ : netconfSessionPreferences.addModuleCaps(sessionPreferences);
+
+ netconfSessionPreferences = overrideNetconfCapabilities.get().nonModuleBasedCapsOverrided()
+ ? netconfSessionPreferences.replaceNonModuleCaps(sessionPreferences)
+ : netconfSessionPreferences.addNonModuleCaps(sessionPreferences);
+ LOG.debug("{}: Session capabilities overridden, capabilities that will be used: {}", id,
+ netconfSessionPreferences);
}
-
remoteDevice.onRemoteSessionUp(netconfSessionPreferences, this);
if (!firstConnectionFuture.isDone()) {
firstConnectionFuture.set(netconfSessionPreferences.getNetconfDeviceCapabilities());
}
- initFuture.addListener(new GenericFutureListener<Future<Object>>(){
-
- @Override
- public void operationComplete(Future<Object> future) throws Exception {
- if (!future.isSuccess() && !future.isCancelled()) {
- LOG.debug("{}: Connection failed", id, future.cause());
- NetconfDeviceCommunicator.this.remoteDevice.onRemoteSessionFailed(future.cause());
- if (firstConnectionFuture.isDone()) {
- firstConnectionFuture.setException(future.cause());
- }
+ initFuture.addListener(future -> {
+ if (!future.isSuccess() && !future.isCancelled()) {
+ LOG.debug("{}: Connection failed", id, future.cause());
+ NetconfDeviceCommunicator.this.remoteDevice.onRemoteSessionFailed(future.cause());
+ if (firstConnectionFuture.isDone()) {
+ firstConnectionFuture.setException(future.cause());
}
}
});
}
}
- private void tearDown( String reason ) {
+ private void tearDown(final String reason) {
if (!isSessionClosing()) {
LOG.warn("It's curious that no one to close the session but tearDown is called!");
}
LOG.debug("Tearing down {}", reason);
- List<UncancellableFuture<RpcResult<NetconfMessage>>> futuresToCancel = Lists.newArrayList();
+ final List<UncancellableFuture<RpcResult<NetconfMessage>>> futuresToCancel = Lists.newArrayList();
sessionLock.lock();
try {
if( session != null ) {
// Notify pending request futures outside of the sessionLock to avoid unnecessarily
// blocking the caller.
- for( UncancellableFuture<RpcResult<NetconfMessage>> future: futuresToCancel ) {
+ for (final UncancellableFuture<RpcResult<NetconfMessage>> future : futuresToCancel) {
if( Strings.isNullOrEmpty( reason ) ) {
future.set( createSessionDownRpcResult() );
} else {
String.format( "The netconf session to %1$s is disconnected", id.getName() ) );
}
- private RpcResult<NetconfMessage> createErrorRpcResult( RpcError.ErrorType errorType, String message ) {
+ private static RpcResult<NetconfMessage> createErrorRpcResult(final RpcError.ErrorType errorType,
+ final String message) {
return RpcResultBuilder.<NetconfMessage>failed()
- .withError(errorType, NetconfDocumentedException.ErrorTag.operation_failed.getTagValue(), message).build();
+ .withError(errorType, NetconfDocumentedException.ErrorTag.OPERATION_FAILED.getTagValue(), message).build();
}
@Override
return Futures.immediateFuture( createSessionDownRpcResult() );
}
- final Request req = new Request( new UncancellableFuture<RpcResult<NetconfMessage>>(true),
- message );
+ final Request req = new Request(new UncancellableFuture<>(true), message);
requests.add(req);
- session.sendMessage(req.request).addListener(new FutureListener<Void>() {
- @Override
- public void operationComplete(final Future<Void> future) throws Exception {
- if( !future.isSuccess() ) {
- // We expect that a session down will occur at this point
- LOG.debug("{}: Failed to send request {}", id,
- XmlUtil.toString(req.request.getDocument()),
- future.cause());
-
- if( future.cause() != null ) {
- req.future.set( createErrorRpcResult( RpcError.ErrorType.TRANSPORT,
- future.cause().getLocalizedMessage() ) );
- } else {
- req.future.set( createSessionDownRpcResult() ); // assume session is down
- }
- req.future.setException( future.cause() );
- }
- else {
- LOG.trace("Finished sending request {}", req.request);
+ session.sendMessage(req.request).addListener(future -> {
+ if( !future.isSuccess() ) {
+ // We expect that a session down will occur at this point
+ LOG.debug("{}: Failed to send request {}", id,
+ XmlUtil.toString(req.request.getDocument()),
+ future.cause());
+
+ if( future.cause() != null ) {
+ req.future.set( createErrorRpcResult( RpcError.ErrorType.TRANSPORT,
+ future.cause().getLocalizedMessage() ) );
+ } else {
+ req.future.set( createSessionDownRpcResult() ); // assume session is down
}
+ req.future.setException( future.cause() );
+ }
+ else {
+ LOG.trace("Finished sending request {}", req.request);
}
});
Code Review / netconf.git / blobdiff