X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;f=netconf%2Fnetconf-netty-util%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fnetconf%2Fnettyutil%2Fhandler%2Fssh%2Fclient%2FAsyncSshHandler.java;h=0d2f6e1aa5aa42868fb6bcec746a87ee168139a9;hb=22755ee880d798830f42cc0d60e84c8602fe3794;hp=5132df6e7145f8a5ca9c689a17e6505c2c027420;hpb=c1999af65772cfa7d9a6789926c496c31637772f;p=netconf.git diff --git a/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java b/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java index 5132df6e71..0d2f6e1aa5 100644 --- a/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java +++ b/netconf/netconf-netty-util/src/main/java/org/opendaylight/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java @@ -7,24 +7,25 @@ */ package org.opendaylight.netconf.nettyutil.handler.ssh.client; +import static com.google.common.base.Verify.verify; import static java.util.Objects.requireNonNull; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelOutboundHandlerAdapter; import io.netty.channel.ChannelPromise; import io.netty.util.concurrent.Future; -import io.netty.util.concurrent.GenericFutureListener; +import io.netty.util.concurrent.FutureListener; import java.io.IOException; import java.net.SocketAddress; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import org.eclipse.jdt.annotation.Nullable; import org.opendaylight.netconf.nettyutil.handler.ssh.authentication.AuthenticationHandler; -import org.opendaylight.netconf.shaded.sshd.client.SshClient; import org.opendaylight.netconf.shaded.sshd.client.channel.ClientChannel; import org.opendaylight.netconf.shaded.sshd.client.future.AuthFuture; import org.opendaylight.netconf.shaded.sshd.client.future.ConnectFuture; import org.opendaylight.netconf.shaded.sshd.client.session.ClientSession; +import org.opendaylight.netconf.shaded.sshd.core.CoreModuleProperties; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -37,15 +38,16 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { public static final String SUBSYSTEM = "netconf"; public static final int SSH_DEFAULT_NIO_WORKERS = 8; - // Disable default timeouts from mina sshd - private static final long DEFAULT_TIMEOUT = -1L; - public static final SshClient DEFAULT_CLIENT; + public static final NetconfSshClient DEFAULT_CLIENT; static { - final SshClient c = SshClient.setUpDefaultClient(); - c.getProperties().put(SshClient.AUTH_TIMEOUT, Long.toString(DEFAULT_TIMEOUT)); - c.getProperties().put(SshClient.IDLE_TIMEOUT, Long.toString(DEFAULT_TIMEOUT)); + final NetconfSshClient c = new NetconfClientBuilder().build(); + // Disable default timeouts from mina sshd + c.getProperties().put(CoreModuleProperties.AUTH_TIMEOUT.getName(), "0"); + c.getProperties().put(CoreModuleProperties.IDLE_TIMEOUT.getName(), "0"); + c.getProperties().put(CoreModuleProperties.NIO2_READ_TIMEOUT.getName(), "0"); + c.getProperties().put(CoreModuleProperties.TCP_NODELAY.getName(), true); // TODO make configurable, or somehow reuse netty threadpool c.setNioWorkers(SSH_DEFAULT_NIO_WORKERS); @@ -55,18 +57,17 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { private final AtomicBoolean isDisconnected = new AtomicBoolean(); private final AuthenticationHandler authenticationHandler; - private final SshClient sshClient; private final Future negotiationFuture; + private final NetconfSshClient sshClient; - private AsyncSshHandlerReader sshReadAsyncListener; private AsyncSshHandlerWriter sshWriteAsyncHandler; - private ClientChannel channel; + private NettyAwareChannelSubsystem channel; private ClientSession session; private ChannelPromise connectPromise; - private GenericFutureListener negotiationFutureListener; + private FutureListener negotiationFutureListener; - public AsyncSshHandler(final AuthenticationHandler authenticationHandler, final SshClient sshClient, + public AsyncSshHandler(final AuthenticationHandler authenticationHandler, final NetconfSshClient sshClient, final Future negotiationFuture) { this.authenticationHandler = requireNonNull(authenticationHandler); this.sshClient = requireNonNull(sshClient); @@ -79,7 +80,7 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { * @param authenticationHandler authentication handler * @param sshClient started SshClient */ - public AsyncSshHandler(final AuthenticationHandler authenticationHandler, final SshClient sshClient) { + public AsyncSshHandler(final AuthenticationHandler authenticationHandler, final NetconfSshClient sshClient) { this(authenticationHandler, sshClient, null); } @@ -96,7 +97,7 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { * @return {@code AsyncSshHandler} */ public static AsyncSshHandler createForNetconfSubsystem(final AuthenticationHandler authenticationHandler, - final Future negotiationFuture, @Nullable final SshClient sshClient) { + final Future negotiationFuture, final @Nullable NetconfSshClient sshClient) { return new AsyncSshHandler(authenticationHandler, sshClient != null ? sshClient : DEFAULT_CLIENT, negotiationFuture); } @@ -120,8 +121,10 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { LOG.trace("SSH session created on channel: {}", ctx.channel()); session = future.getSession(); + verify(session instanceof NettyAwareClientSession, "Unexpected session %s", session); + final AuthFuture authenticateFuture = authenticationHandler.authenticate(session); - final ClientSession localSession = session; + final NettyAwareClientSession localSession = (NettyAwareClientSession) session; authenticateFuture.addListener(future1 -> { if (future1.isSuccess()) { handleSshAuthenticated(localSession, ctx); @@ -135,12 +138,13 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { } } - private synchronized void handleSshAuthenticated(final ClientSession newSession, final ChannelHandlerContext ctx) { - try { - LOG.debug("SSH session authenticated on channel: {}, server version: {}", ctx.channel(), - newSession.getServerVersion()); + private synchronized void handleSshAuthenticated(final NettyAwareClientSession newSession, + final ChannelHandlerContext ctx) { + LOG.debug("SSH session authenticated on channel: {}, server version: {}", ctx.channel(), + newSession.getServerVersion()); - channel = newSession.createSubsystemChannel(SUBSYSTEM); + try { + channel = newSession.createSubsystemChannel(SUBSYSTEM, ctx); channel.setStreaming(ClientChannel.Streaming.Async); channel.open().addListener(future -> { if (future.isOpened()) { @@ -149,8 +153,6 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { handleSshSetupFailure(ctx, future.getException()); } }); - - } catch (final IOException e) { handleSshSetupFailure(ctx, e); } @@ -163,18 +165,9 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { connectPromise.setSuccess(); } - // TODO we should also read from error stream and at least log from that - - ClientChannel localChannel = channel; - sshReadAsyncListener = new AsyncSshHandlerReader(() -> AsyncSshHandler.this.disconnect(ctx, ctx.newPromise()), - ctx::fireChannelRead, localChannel.toString(), localChannel.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()); - ctx.fireChannelActive(); - } + sshWriteAsyncHandler = new AsyncSshHandlerWriter(channel.getAsyncIn()); + ctx.fireChannelActive(); + channel.onClose(() -> disconnect(ctx, ctx.newPromise())); } private synchronized void handleSshSetupFailure(final ChannelHandlerContext ctx, final Throwable error) { @@ -196,8 +189,8 @@ 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; + LOG.debug("SSH session connecting on channel {}. promise: {}", ctx.channel(), promise); + connectPromise = requireNonNull(promise); if (negotiationFuture != null) { negotiationFutureListener = future -> { @@ -238,10 +231,6 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { sshWriteAsyncHandler.close(); } - if (sshReadAsyncListener != null) { - sshReadAsyncListener.close(); - } - //If connection promise is not already set, it means negotiation failed //we must set connection promise to failure if (!connectPromise.isDone()) { @@ -277,7 +266,12 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter { LOG.warn("Unable to cleanup all resources for channel: {}. Ignoring.", ctx.channel(), e); } - channel = null; + if (channel != null) { + //TODO: see if calling just close() is sufficient + //channel.close(false); + channel.close(); + channel = null; + } promise.setSuccess(); LOG.debug("SSH session closed on channel: {}", ctx.channel()); }