Rename AsyncSshHandler.handleSshSetupFailure()
[netconf.git] / netconf / netconf-netty-util / src / main / java / org / opendaylight / netconf / nettyutil / handler / ssh / client / AsyncSshHandler.java
index 0eb635ef3d0866599909888ecdbe175eb2269162..c8312e9419af95fe90a91c8130f35c74319a0ff1 100644 (file)
@@ -14,7 +14,7 @@ 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;
@@ -24,6 +24,7 @@ import org.opendaylight.netconf.nettyutil.handler.ssh.authentication.Authenticat
 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.future.OpenFuture;
 import org.opendaylight.netconf.shaded.sshd.client.session.ClientSession;
 import org.opendaylight.netconf.shaded.sshd.core.CoreModuleProperties;
 import org.slf4j.Logger;
@@ -47,6 +48,7 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
         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);
@@ -64,7 +66,7 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
     private NettyAwareChannelSubsystem channel;
     private ClientSession session;
     private ChannelPromise connectPromise;
-    private GenericFutureListener negotiationFutureListener;
+    private FutureListener<Object> negotiationFutureListener;
 
     public AsyncSshHandler(final AuthenticationHandler authenticationHandler, final NetconfSshClient sshClient,
             final Future<?> negotiationFuture) {
@@ -101,65 +103,103 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
                 negotiationFuture);
     }
 
-    private void startSsh(final ChannelHandlerContext ctx, final SocketAddress address) throws IOException {
-        LOG.debug("Starting SSH to {} on channel: {}", address, ctx.channel());
-
-        final ConnectFuture sshConnectionFuture = sshClient.connect(authenticationHandler.getUsername(), address)
-               .verify(ctx.channel().config().getConnectTimeoutMillis(), TimeUnit.MILLISECONDS);
-        sshConnectionFuture.addListener(future -> {
-            if (future.isConnected()) {
-                handleSshSessionCreated(future, ctx);
-            } else {
-                handleSshSetupFailure(ctx, future.getException());
-            }
-        });
+    @Override
+    public synchronized void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) {
+        sshWriteAsyncHandler.write(ctx, msg, promise);
     }
 
-    private synchronized void handleSshSessionCreated(final ConnectFuture future, final ChannelHandlerContext ctx) {
-        try {
-            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 NettyAwareClientSession localSession = (NettyAwareClientSession) session;
-            authenticateFuture.addListener(future1 -> {
-                if (future1.isSuccess()) {
-                    handleSshAuthenticated(localSession, ctx);
-                } else {
-                    handleSshSetupFailure(ctx, new AuthenticationFailedException("Authentication failed",
-                        future1.getException()));
+    @Override
+    public synchronized void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress,
+            final SocketAddress localAddress, final ChannelPromise promise) throws IOException {
+        LOG.debug("SSH session connecting on channel {}. promise: {}", ctx.channel(), promise);
+        connectPromise = requireNonNull(promise);
+
+        if (negotiationFuture != null) {
+            negotiationFutureListener = future -> {
+                if (future.isSuccess()) {
+                    promise.setSuccess();
                 }
-            });
-        } catch (final IOException e) {
-            handleSshSetupFailure(ctx, e);
+            };
+            //complete connection promise with netconf negotiation future
+            negotiationFuture.addListener(negotiationFutureListener);
+        }
+
+        LOG.debug("Starting SSH to {} on channel: {}", remoteAddress, ctx.channel());
+        sshClient.connect(authenticationHandler.getUsername(), remoteAddress)
+            // FIXME: this is a blocking call, we should handle this with a concurrently-scheduled timeout. We do not
+            //        have a Timer ready, so perhaps we should be using the event loop?
+            .verify(ctx.channel().config().getConnectTimeoutMillis(), TimeUnit.MILLISECONDS)
+            .addListener(future -> onConnectComplete(future, ctx));
+    }
+
+    private void onConnectComplete(final ConnectFuture future, final ChannelHandlerContext ctx) {
+        final var cause = future.getException();
+        if (cause != null) {
+            onOpenFailure(ctx, cause);
+            return;
         }
+
+        final var clientSession = future.getSession();
+        LOG.trace("SSH session {} created on channel: {}", clientSession, ctx.channel());
+        verify(clientSession instanceof NettyAwareClientSession, "Unexpected session %s", clientSession);
+        onConnectComplete((NettyAwareClientSession) clientSession, ctx);
     }
 
-    private synchronized void handleSshAuthenticated(final NettyAwareClientSession newSession,
+    private synchronized void onConnectComplete(final NettyAwareClientSession clientSession,
             final ChannelHandlerContext ctx) {
+        session = clientSession;
+
+        final AuthFuture authFuture;
         try {
-            LOG.debug("SSH session authenticated on channel: {}, server version: {}", ctx.channel(),
-                    newSession.getServerVersion());
+            authFuture = authenticationHandler.authenticate(clientSession);
+        } catch (final IOException e) {
+            onOpenFailure(ctx, e);
+            return;
+        }
 
-            channel = newSession.createSubsystemChannel(SUBSYSTEM, ctx);
-            channel.setStreaming(ClientChannel.Streaming.Async);
-            channel.open().addListener(future -> {
-                if (future.isOpened()) {
-                    handleSshChanelOpened(ctx);
-                } else {
-                    handleSshSetupFailure(ctx, future.getException());
-                }
-            });
+        authFuture.addListener(future -> onAuthComplete(future, clientSession, ctx));
+    }
+
+    private void onAuthComplete(final AuthFuture future, final NettyAwareClientSession clientSession,
+            final ChannelHandlerContext ctx) {
+        final var cause = future.getException();
+        if (cause != null) {
+            onOpenFailure(ctx, new AuthenticationFailedException("Authentication failed", cause));
+            return;
+        }
+
+        onAuthComplete(clientSession, ctx);
+    }
 
+    private synchronized void onAuthComplete(final NettyAwareClientSession clientSession,
+            final ChannelHandlerContext ctx) {
+        LOG.debug("SSH session authenticated on channel: {}, server version: {}", ctx.channel(),
+            clientSession.getServerVersion());
 
+        final OpenFuture openFuture;
+        try {
+            channel = clientSession.createSubsystemChannel(SUBSYSTEM, ctx);
+            channel.setStreaming(ClientChannel.Streaming.Async);
+            openFuture = channel.open();
         } catch (final IOException e) {
-            handleSshSetupFailure(ctx, e);
+            onOpenFailure(ctx, e);
+            return;
         }
+
+        openFuture.addListener(future -> onOpenComplete(future, ctx));
     }
 
-    private synchronized void handleSshChanelOpened(final ChannelHandlerContext ctx) {
+    private void onOpenComplete(final OpenFuture future, final ChannelHandlerContext ctx) {
+        final var cause = future.getException();
+        if (cause != null) {
+            onOpenFailure(ctx, cause);
+            return;
+        }
+
+        onOpenComplete(ctx);
+    }
+
+    private synchronized void onOpenComplete(final ChannelHandlerContext ctx) {
         LOG.trace("SSH subsystem channel opened successfully on channel: {}", ctx.channel());
 
         if (negotiationFuture == null) {
@@ -171,40 +211,17 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
         channel.onClose(() -> disconnect(ctx, ctx.newPromise()));
     }
 
-    private synchronized void handleSshSetupFailure(final ChannelHandlerContext ctx, final Throwable error) {
-        LOG.warn("Unable to setup SSH connection on channel: {}", ctx.channel(), error);
+    private synchronized void onOpenFailure(final ChannelHandlerContext ctx, final Throwable cause) {
+        LOG.warn("Unable to setup SSH connection on channel: {}", ctx.channel(), cause);
 
         // If the promise is not yet done, we have failed with initial connect and set connectPromise to failure
         if (!connectPromise.isDone()) {
-            connectPromise.setFailure(error);
+            connectPromise.setFailure(cause);
         }
 
         disconnect(ctx, ctx.newPromise());
     }
 
-    @Override
-    public synchronized void write(final ChannelHandlerContext ctx, final Object msg, final ChannelPromise promise) {
-        sshWriteAsyncHandler.write(ctx, msg, promise);
-    }
-
-    @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;
-
-        if (negotiationFuture != null) {
-            negotiationFutureListener = future -> {
-                if (future.isSuccess()) {
-                    promise.setSuccess();
-                }
-            };
-            //complete connection promise with netconf negotiation future
-            negotiationFuture.addListener(negotiationFutureListener);
-        }
-        startSsh(ctx, remoteAddress);
-    }
-
     @Override
     public void close(final ChannelHandlerContext ctx, final ChannelPromise promise) {
         disconnect(ctx, promise);