X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;ds=sidebyside;f=opendaylight%2Fnetconf%2Fnetconf-netty-util%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fnetconf%2Fnettyutil%2Fhandler%2Fssh%2Fclient%2FAsyncSshHandler.java;h=064ae72bc7c512db762a45007a73e9aeebebbe45;hb=edcc020c8fda4b13f22a31d79c13feef0b53b0ee;hp=3bd72320232bb7f912a316469b816fa76952f0c4;hpb=93f1b5b561183147a8a8f91a8af89fd6eb56a0a9;p=controller.git diff --git a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java index 3bd7232023..064ae72bc7 100644 --- a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java +++ b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java @@ -9,6 +9,7 @@ package org.opendaylight.controller.netconf.nettyutil.handler.ssh.client; import com.google.common.base.Preconditions; +import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelOutboundHandlerAdapter; import io.netty.channel.ChannelPromise; @@ -31,7 +32,7 @@ import org.slf4j.LoggerFactory; */ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { - private static final Logger logger = LoggerFactory.getLogger(AsyncSshHandler.class); + private static final Logger LOG = LoggerFactory.getLogger(AsyncSshHandler.class); public static final String SUBSYSTEM = "netconf"; public static final SshClient DEFAULT_CLIENT = SshClient.setUpDefaultClient(); @@ -47,7 +48,7 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { private final AuthenticationHandler authenticationHandler; private final SshClient sshClient; - private AsyncSshHanderReader sshReadAsyncListener; + private AsyncSshHandlerReader sshReadAsyncListener; private AsyncSshHandlerWriter sshWriteAsyncHandler; private ClientChannel channel; @@ -73,7 +74,7 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { } private void startSsh(final ChannelHandlerContext ctx, final SocketAddress address) { - logger.debug("Starting SSH to {} on channel: {}", address, ctx.channel()); + LOG.debug("Starting SSH to {} on channel: {}", address, ctx.channel()); final ConnectFuture sshConnectionFuture = sshClient.connect(authenticationHandler.getUsername(), address); sshConnectionFuture.addListener(new SshFutureListener() { @@ -90,7 +91,7 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { private synchronized void handleSshSessionCreated(final ConnectFuture future, final ChannelHandlerContext ctx) { try { - logger.trace("SSH session created on channel: {}", ctx.channel()); + LOG.trace("SSH session created on channel: {}", ctx.channel()); session = future.getSession(); final AuthFuture authenticateFuture = authenticationHandler.authenticate(session); @@ -100,7 +101,11 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { if (future.isSuccess()) { handleSshAuthenticated(session, ctx); } else { - handleSshSetupFailure(ctx, future.getException()); + // Exception does not have to be set in the future, add simple exception in such case + final Throwable exception = future.getException() == null ? + new IllegalStateException("Authentication failed") : + future.getException(); + handleSshSetupFailure(ctx, exception); } } }); @@ -111,7 +116,7 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { private synchronized void handleSshAuthenticated(final ClientSession session, final ChannelHandlerContext ctx) { try { - logger.debug("SSH session authenticated on channel: {}, server version: {}", ctx.channel(), session.getServerVersion()); + LOG.debug("SSH session authenticated on channel: {}, server version: {}", ctx.channel(), session.getServerVersion()); channel = session.createSubsystemChannel(SUBSYSTEM); channel.setStreaming(ClientChannel.Streaming.Async); @@ -133,12 +138,24 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { } private synchronized void handleSshChanelOpened(final ChannelHandlerContext ctx) { - logger.trace("SSH subsystem channel opened successfully on channel: {}", ctx.channel()); + LOG.trace("SSH subsystem channel opened successfully on channel: {}", ctx.channel()); connectPromise.setSuccess(); - connectPromise = null; - sshReadAsyncListener = new AsyncSshHanderReader(this, ctx, channel.getAsyncOut()); + // TODO we should also read from error stream and at least log from that + + sshReadAsyncListener = new AsyncSshHandlerReader(new AutoCloseable() { + @Override + public void close() throws Exception { + AsyncSshHandler.this.disconnect(ctx, ctx.newPromise()); + } + }, new AsyncSshHandlerReader.ReadMsgHandler() { + @Override + public void onMessageRead(final ByteBuf msg) { + ctx.fireChannelRead(msg); + } + }, channel.toString(), channel.getAsyncOut()); + // if readAsyncListener receives immediate close, it will close this handler and closing this handler sets channel variable to null if(channel != null) { sshWriteAsyncHandler = new AsyncSshHandlerWriter(channel.getAsyncIn()); @@ -147,10 +164,13 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { } private synchronized void handleSshSetupFailure(final ChannelHandlerContext ctx, final Throwable e) { - logger.warn("Unable to setup SSH connection on channel: {}", ctx.channel(), e); - connectPromise.setFailure(e); - connectPromise = null; - throw new IllegalStateException("Unable to setup SSH connection on channel: " + ctx.channel(), e); + LOG.warn("Unable to setup SSH connection on channel: {}", ctx.channel(), e); + disconnect(ctx, ctx.newPromise()); + + // If the promise is not yet done, we have failed with initial connect and set connectPromise to failure + if(!connectPromise.isDone()) { + connectPromise.setFailure(e); + } } @Override @@ -160,6 +180,7 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { @Override public synchronized void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) throws Exception { + LOG.debug("SSH session connecting on channel {}. promise: {} ", ctx.channel(), connectPromise); this.connectPromise = promise; startSsh(ctx, remoteAddress); } @@ -171,14 +192,21 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { @Override public synchronized void disconnect(final ChannelHandlerContext ctx, final ChannelPromise promise) { - if(sshReadAsyncListener != null) { - sshReadAsyncListener.close(); + LOG.trace("Closing SSH session on channel: {} with connect promise in state: {}", ctx.channel(), connectPromise); + + // If we have already succeeded and the session was dropped after, we need to fire inactive to notify reconnect logic + if(connectPromise.isSuccess()) { + ctx.fireChannelInactive(); } if(sshWriteAsyncHandler != null) { sshWriteAsyncHandler.close(); } + if(sshReadAsyncListener != null) { + sshReadAsyncListener.close(); + } + if(session!= null && !session.isClosed() && !session.isClosing()) { session.close(false).addListener(new SshFutureListener() { @Override @@ -191,11 +219,19 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { }); } + // Super disconnect is necessary in this case since we are using NioSocketChannel and it needs to cleanup its resources + // e.g. Socket that it tries to open in its constructor (https://bugs.opendaylight.org/show_bug.cgi?id=2430) + // TODO better solution would be to implement custom ChannelFactory + Channel that will use mina SSH lib internally: port this to custom channel implementation + try { + // Disconnect has to be closed after inactive channel event was fired, because it interferes with it + super.disconnect(ctx, ctx.newPromise()); + } catch (final Exception e) { + LOG.warn("Unable to cleanup all resources for channel: {}. Ignoring.", ctx.channel(), e); + } + channel = null; promise.setSuccess(); - - logger.debug("SSH session closed on channel: {}", ctx.channel()); - ctx.fireChannelInactive(); + LOG.debug("SSH session closed on channel: {}", ctx.channel()); } }