From: Tony Tkacik Date: Mon, 21 Jul 2014 09:11:50 +0000 (+0000) Subject: Merge "Bug 1362: New AsyncWriteTransaction#submit method" X-Git-Tag: release/helium~466 X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=commitdiff_plain;h=3d81e6fb622fb7d5b239a8697d51bcb6d96658db;hp=0219d667fd81b48cd3f05faee7d39aa1acce73a4 Merge "Bug 1362: New AsyncWriteTransaction#submit method" --- diff --git a/opendaylight/netconf/netconf-client/src/main/java/org/opendaylight/controller/netconf/client/SshClientChannelInitializer.java b/opendaylight/netconf/netconf-client/src/main/java/org/opendaylight/controller/netconf/client/SshClientChannelInitializer.java index 829ac304bd..87b3f837e8 100644 --- a/opendaylight/netconf/netconf-client/src/main/java/org/opendaylight/controller/netconf/client/SshClientChannelInitializer.java +++ b/opendaylight/netconf/netconf-client/src/main/java/org/opendaylight/controller/netconf/client/SshClientChannelInitializer.java @@ -11,9 +11,8 @@ import io.netty.channel.Channel; import io.netty.util.concurrent.Promise; import java.io.IOException; import org.opendaylight.controller.netconf.nettyutil.AbstractChannelInitializer; -import org.opendaylight.controller.netconf.nettyutil.handler.ssh.SshHandler; import org.opendaylight.controller.netconf.nettyutil.handler.ssh.authentication.AuthenticationHandler; -import org.opendaylight.controller.netconf.nettyutil.handler.ssh.client.Invoker; +import org.opendaylight.controller.netconf.nettyutil.handler.ssh.client.SshHandler; import org.opendaylight.protocol.framework.SessionListenerFactory; final class SshClientChannelInitializer extends AbstractChannelInitializer { @@ -33,8 +32,7 @@ final class SshClientChannelInitializer extends AbstractChannelInitializer promise) { try { - final Invoker invoker = Invoker.subsystem("netconf"); - ch.pipeline().addFirst(new SshHandler(authenticationHandler, invoker)); + ch.pipeline().addFirst(SshHandler.createForNetconfSubsystem(authenticationHandler)); super.initialize(ch,promise); } catch (final IOException e) { throw new RuntimeException(e); diff --git a/opendaylight/netconf/netconf-client/src/test/java/org/opendaylight/controller/netconf/client/test/TestingNetconfClient.java b/opendaylight/netconf/netconf-client/src/test/java/org/opendaylight/controller/netconf/client/test/TestingNetconfClient.java index afa17532d5..18ed18e4ae 100644 --- a/opendaylight/netconf/netconf-client/src/test/java/org/opendaylight/controller/netconf/client/test/TestingNetconfClient.java +++ b/opendaylight/netconf/netconf-client/src/test/java/org/opendaylight/controller/netconf/client/test/TestingNetconfClient.java @@ -56,7 +56,7 @@ public class TestingNetconfClient implements Closeable { this.label = clientLabel; sessionListener = config.getSessionListener(); Future clientFuture = netconfClientDispatcher.createClient(config); - clientSession = get(clientFuture); + clientSession = get(clientFuture);//TODO: make static this.sessionId = clientSession.getSessionId(); } diff --git a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/exi/EXIParameters.java b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/exi/EXIParameters.java index 84353a4646..993709258a 100644 --- a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/exi/EXIParameters.java +++ b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/exi/EXIParameters.java @@ -7,14 +7,12 @@ */ package org.opendaylight.controller.netconf.nettyutil.handler.exi; -import org.opendaylight.controller.netconf.api.NetconfMessage; +import com.google.common.base.Preconditions; import org.opendaylight.controller.netconf.util.xml.XmlElement; import org.openexi.proc.common.AlignmentType; import org.openexi.proc.common.EXIOptions; import org.openexi.proc.common.EXIOptionsException; -import com.google.common.base.Preconditions; - public final class EXIParameters { private static final String EXI_PARAMETER_ALIGNMENT = "alignment"; private static final String EXI_PARAMETER_BYTE_ALIGNED = "byte-aligned"; @@ -29,20 +27,12 @@ public final class EXIParameters { private static final String EXI_FIDELITY_PIS = "pis"; private static final String EXI_FIDELITY_PREFIXES = "prefixes"; - private static final String EXI_PARAMETER_SCHEMA = "schema"; - private static final String EXI_PARAMETER_SCHEMA_NONE = "none"; - private static final String EXI_PARAMETER_SCHEMA_BUILT_IN = "builtin"; - private static final String EXI_PARAMETER_SCHEMA_BASE_1_1 = "base:1.1"; - private final EXIOptions options; private EXIParameters(final EXIOptions options) { this.options = Preconditions.checkNotNull(options); } - public static EXIParameters fromNetconfMessage(final NetconfMessage root) throws EXIOptionsException { - return fromXmlElement(XmlElement.fromDomDocument(root.getDocument())); - } public static EXIParameters fromXmlElement(final XmlElement root) throws EXIOptionsException { final EXIOptions options = new EXIOptions(); @@ -77,30 +67,6 @@ public final class EXIParameters { options.setPreserveNS(true); } } - - if (root.getElementsByTagName(EXI_PARAMETER_SCHEMA).getLength() > 0) { -/* - GrammarFactory grammarFactory = GrammarFactory.newInstance(); - if (operationElement - .getElementsByTagName(EXI_PARAMETER_SCHEMA_NONE) - .getLength() > 0) { - this.grammars = grammarFactory.createSchemaLessGrammars(); - } - - if (operationElement.getElementsByTagName( - EXI_PARAMETER_SCHEMA_BUILT_IN).getLength() > 0) { - this.grammars = grammarFactory.createXSDTypesOnlyGrammars(); - } - - if (operationElement.getElementsByTagName( - EXI_PARAMETER_SCHEMA_BASE_1_1).getLength() > 0) { - this.grammars = grammarFactory - .createGrammars(NETCONF_XSD_LOCATION); - } -*/ - - } - return new EXIParameters(options); } diff --git a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/exi/NetconfStartExiMessage.java b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/exi/NetconfStartExiMessage.java index 1b0a34d7e0..72eb774b53 100644 --- a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/exi/NetconfStartExiMessage.java +++ b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/exi/NetconfStartExiMessage.java @@ -48,7 +48,7 @@ public final class NetconfStartExiMessage extends NetconfMessage { Element startExiElement = doc.createElementNS(XmlNetconfConstants.URN_IETF_PARAMS_XML_NS_NETCONF_EXI_1_0, START_EXI); - addAlignemnt(exiOptions, doc, startExiElement); + addAlignment(exiOptions, doc, startExiElement); addFidelity(exiOptions, doc, startExiElement); rpcElement.appendChild(startExiElement); @@ -75,7 +75,7 @@ public final class NetconfStartExiMessage extends NetconfMessage { } } - private static void addAlignemnt(EXIOptions exiOptions, Document doc, Element startExiElement) { + private static void addAlignment(EXIOptions exiOptions, Document doc, Element startExiElement) { Element alignmentElement = doc.createElementNS(XmlNetconfConstants.URN_IETF_PARAMS_XML_NS_NETCONF_EXI_1_0, ALIGNMENT_KEY); alignmentElement.setTextContent(exiOptions.getAlignmentType().toString()); diff --git a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/authentication/LoginPassword.java b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/authentication/LoginPassword.java index 67027d8014..b67aa0f96d 100644 --- a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/authentication/LoginPassword.java +++ b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/authentication/LoginPassword.java @@ -14,7 +14,7 @@ import java.io.IOException; /** * Class Providing username/password authentication option to - * {@link org.opendaylight.controller.netconf.nettyutil.handler.ssh.SshHandler} + * {@link org.opendaylight.controller.netconf.nettyutil.handler.ssh.client.SshHandler} */ public class LoginPassword extends AuthenticationHandler { private final String username; diff --git a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/Invoker.java b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/Invoker.java index d542e1952a..eab2546d6e 100644 --- a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/Invoker.java +++ b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/Invoker.java @@ -13,33 +13,38 @@ import java.io.IOException; * Abstract class providing mechanism of invoking various SSH level services. * Class is not allowed to be extended, as it provides its own implementations via instance initiators. */ -public abstract class Invoker { +abstract class Invoker { private boolean invoked = false; - private Invoker(){} + private Invoker() { + } protected boolean isInvoked() { - // TODO invoked is always false return invoked; } + public void setInvoked() { + this.invoked = true; + } + abstract void invoke(SshSession session) throws IOException; - /** - * Invoker implementation to invokes subsystem SSH service. - * - * @param subsystem - * @return - */ + public static Invoker netconfSubsystem(){ + return subsystem("netconf"); + } + public static Invoker subsystem(final String subsystem) { return new Invoker() { @Override - void invoke(SshSession session) throws IOException { + synchronized void invoke(SshSession session) throws IOException { if (isInvoked()) { throw new IllegalStateException("Already invoked."); } - - session.startSubSystem(subsystem); + try { + session.startSubSystem(subsystem); + } finally { + setInvoked(); + } } }; } diff --git a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/SshClient.java b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/SshClient.java index 3520fe029d..271b781b99 100644 --- a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/SshClient.java +++ b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/SshClient.java @@ -10,18 +10,16 @@ package org.opendaylight.controller.netconf.nettyutil.handler.ssh.client; import ch.ethz.ssh2.Connection; import ch.ethz.ssh2.Session; -import ch.ethz.ssh2.channel.Channel; -import org.opendaylight.controller.netconf.nettyutil.handler.ssh.authentication.AuthenticationHandler; -import org.opendaylight.controller.netconf.nettyutil.handler.ssh.virtualsocket.VirtualSocket; - import java.io.IOException; import java.util.HashMap; import java.util.Map; +import org.opendaylight.controller.netconf.nettyutil.handler.ssh.authentication.AuthenticationHandler; +import org.opendaylight.controller.netconf.nettyutil.handler.ssh.virtualsocket.VirtualSocket; /** * Wrapper class around GANYMED SSH java library. */ -public class SshClient { +class SshClient { private final VirtualSocket socket; private final Map openSessions = new HashMap<>(); private final AuthenticationHandler authenticationHandler; @@ -51,15 +49,10 @@ public class SshClient { authenticationHandler.authenticate(connection); } - public void closeSession(SshSession session) { - if (session.getState() == Channel.STATE_OPEN || session.getState() == Channel.STATE_OPENING) { - session.close(); - } - } public void close() { for (SshSession session : openSessions.values()){ - closeSession(session); + session.close(); } openSessions.clear(); @@ -68,4 +61,11 @@ public class SshClient { connection.close(); } } + + @Override + public String toString() { + return "SshClient{" + + "socket=" + socket + + '}'; + } } diff --git a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/SshClientAdapter.java b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/SshClientAdapter.java index ad8b25ff21..1a2eb3f1ab 100644 --- a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/SshClientAdapter.java +++ b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/SshClientAdapter.java @@ -8,8 +8,13 @@ package org.opendaylight.controller.netconf.nettyutil.handler.ssh.client; +import static com.google.common.base.Preconditions.checkArgument; +import static com.google.common.base.Preconditions.checkNotNull; +import static com.google.common.base.Preconditions.checkState; + import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelPromise; import java.io.IOException; @@ -18,7 +23,6 @@ import java.io.OutputStream; import java.util.LinkedList; import java.util.Queue; import java.util.concurrent.atomic.AtomicBoolean; -import org.opendaylight.controller.netconf.nettyutil.handler.ssh.virtualsocket.VirtualSocketException; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -27,7 +31,7 @@ import org.slf4j.LoggerFactory; * Worker thread class. Handles all downstream and upstream events in SSH Netty * pipeline. */ -public class SshClientAdapter implements Runnable { +class SshClientAdapter implements Runnable { private static final Logger logger = LoggerFactory.getLogger(SshClientAdapter.class); private static final int BUFFER_SIZE = 1024; @@ -51,6 +55,7 @@ public class SshClientAdapter implements Runnable { this.invoker = invoker; } + // TODO: refactor public void run() { try { SshSession session = sshClient.openSession(); @@ -80,12 +85,6 @@ public class SshClientAdapter implements Runnable { byteBuf.writeBytes(tranBuff); ctx.fireChannelRead(byteBuf); } - - } catch (VirtualSocketException e) { - // Netty closed connection prematurely. - // Or maybe tried to open ganymed connection without having initialized session - // (ctx.channel().remoteAddress() is null) - // Just pass and move on. } catch (Exception e) { logger.error("Unexpected exception", e); } finally { @@ -123,12 +122,23 @@ public class SshClientAdapter implements Runnable { } } - public void start(ChannelHandlerContext ctx) { - if (this.ctx != null) { - // context is already associated. - return; + public Thread start(ChannelHandlerContext ctx, ChannelFuture channelFuture) { + checkArgument(channelFuture.isSuccess()); + checkNotNull(ctx.channel().remoteAddress()); + synchronized (this) { + checkState(this.ctx == null); + this.ctx = ctx; } - this.ctx = ctx; - new Thread(this).start(); + String threadName = toString(); + Thread thread = new Thread(this, threadName); + thread.start(); + return thread; + } + + @Override + public String toString() { + return "SshClientAdapter{" + + "sshClient=" + sshClient + + '}'; } } diff --git a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/SshHandler.java b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/SshHandler.java similarity index 83% rename from opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/SshHandler.java rename to opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/SshHandler.java index 1427c6f3a7..c710a010e2 100644 --- a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/SshHandler.java +++ b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/SshHandler.java @@ -6,7 +6,7 @@ * and is available at http://www.eclipse.org/legal/epl-v10.html */ -package org.opendaylight.controller.netconf.nettyutil.handler.ssh; +package org.opendaylight.controller.netconf.nettyutil.handler.ssh.client; import io.netty.buffer.ByteBuf; import io.netty.channel.ChannelFuture; @@ -14,26 +14,30 @@ import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelOutboundHandlerAdapter; import io.netty.channel.ChannelPromise; - import java.io.IOException; import java.net.SocketAddress; - -import org.opendaylight.controller.netconf.nettyutil.handler.ssh.client.SshClient; import org.opendaylight.controller.netconf.nettyutil.handler.ssh.authentication.AuthenticationHandler; -import org.opendaylight.controller.netconf.nettyutil.handler.ssh.client.Invoker; -import org.opendaylight.controller.netconf.nettyutil.handler.ssh.client.SshClientAdapter; import org.opendaylight.controller.netconf.nettyutil.handler.ssh.virtualsocket.VirtualSocket; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * Netty SSH handler class. Acts as interface between Netty and SSH library. All standard Netty message handling * stops at instance of this class. All downstream events are handed of to wrapped {@link org.opendaylight.controller.netconf.nettyutil.handler.ssh.client.SshClientAdapter}; */ public class SshHandler extends ChannelOutboundHandlerAdapter { + private static final Logger logger = LoggerFactory.getLogger(SshHandler.class); private static final String SOCKET = "socket"; private final VirtualSocket virtualSocket = new VirtualSocket(); private final SshClientAdapter sshClientAdapter; + + public static SshHandler createForNetconfSubsystem(AuthenticationHandler authenticationHandler) throws IOException { + return new SshHandler(authenticationHandler, Invoker.netconfSubsystem()); + } + + public SshHandler(AuthenticationHandler authenticationHandler, Invoker invoker) throws IOException { SshClient sshClient = new SshClient(virtualSocket, authenticationHandler); this.sshClientAdapter = new SshClientAdapter(sshClient, invoker); @@ -67,7 +71,11 @@ public class SshHandler extends ChannelOutboundHandlerAdapter { promise.addListener(new ChannelFutureListener() { public void operationComplete(ChannelFuture channelFuture) { - sshClientAdapter.start(ctx); + if (channelFuture.isSuccess()) { + sshClientAdapter.start(ctx, channelFuture); + } else { + logger.debug("Failed to connect to remote host"); + } }} ); } diff --git a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/SshSession.java b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/SshSession.java index 8311554cda..44893b8794 100644 --- a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/SshSession.java +++ b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/SshSession.java @@ -11,6 +11,7 @@ package org.opendaylight.controller.netconf.nettyutil.handler.ssh.client; import ch.ethz.ssh2.Session; import ch.ethz.ssh2.StreamGobbler; +import ch.ethz.ssh2.channel.Channel; import java.io.Closeable; import java.io.IOException; import java.io.InputStream; @@ -19,33 +20,18 @@ import java.io.OutputStream; /** * Wrapper class for proprietary SSH sessions implementations */ -public class SshSession implements Closeable { +class SshSession implements Closeable { private final Session session; public SshSession(Session session) { this.session = session; } - public void execCommand(String cmd) throws IOException { - session.execCommand(cmd); - } - - public void execCommand(String cmd, String charsetName) throws IOException { - session.execCommand(cmd, charsetName); - } - - public void startShell() throws IOException { - session.startShell(); - } public void startSubSystem(String name) throws IOException { session.startSubSystem(name); } - public int getState() { - return session.getState(); - } - public InputStream getStdout() { return new StreamGobbler(session.getStdout()); } @@ -58,24 +44,10 @@ public class SshSession implements Closeable { return session.getStdin(); } - public int waitUntilDataAvailable(long timeout) throws IOException { - return session.waitUntilDataAvailable(timeout); - } - - public int waitForCondition(int conditionSet, long timeout) { - return session.waitForCondition(conditionSet, timeout); - } - - public Integer getExitStatus() { - return session.getExitStatus(); - } - - public String getExitSignal() { - return session.getExitSignal(); - } - @Override public void close() { - session.close(); + if (session.getState() == Channel.STATE_OPEN || session.getState() == Channel.STATE_OPENING) { + session.close(); + } } } diff --git a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/virtualsocket/VirtualSocket.java b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/virtualsocket/VirtualSocket.java index 6debeba97e..69cce8057e 100644 --- a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/virtualsocket/VirtualSocket.java +++ b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/virtualsocket/VirtualSocket.java @@ -25,32 +25,33 @@ import java.nio.channels.SocketChannel; * use OIO application in asynchronous environment and NIO EventLoop. Using VirtualSocket OIO applications * are able to use full potential of NIO environment. */ +//TODO: refactor - socket should be created when connection is established public class VirtualSocket extends Socket implements ChannelHandler { private static final String INPUT_STREAM = "inputStream"; private static final String OUTPUT_STREAM = "outputStream"; - private final ChannelInputStream chis = new ChannelInputStream(); - private final ChannelOutputStream chos = new ChannelOutputStream(); + private final ChannelInputStream chais = new ChannelInputStream(); + private final ChannelOutputStream chaos = new ChannelOutputStream(); private ChannelHandlerContext ctx; public InputStream getInputStream() { - return this.chis; + return this.chais; } public OutputStream getOutputStream() { - return this.chos; + return this.chaos; } public void handlerAdded(ChannelHandlerContext ctx) { this.ctx = ctx; if (ctx.channel().pipeline().get(OUTPUT_STREAM) == null) { - ctx.channel().pipeline().addFirst(OUTPUT_STREAM, chos); + ctx.channel().pipeline().addFirst(OUTPUT_STREAM, chaos); } if (ctx.channel().pipeline().get(INPUT_STREAM) == null) { - ctx.channel().pipeline().addFirst(INPUT_STREAM, chis); + ctx.channel().pipeline().addFirst(INPUT_STREAM, chais); } } @@ -69,7 +70,6 @@ public class VirtualSocket extends Socket implements ChannelHandler { ctx.fireExceptionCaught(throwable); } - public VirtualSocket() {super();} @Override public void connect(SocketAddress endpoint) throws IOException {} @@ -83,12 +83,7 @@ public class VirtualSocket extends Socket implements ChannelHandler { @Override public InetAddress getInetAddress() { InetSocketAddress isa = getInetSocketAddress(); - - if (isa == null) { - throw new VirtualSocketException(); - } - - return getInetSocketAddress().getAddress(); + return isa.getAddress(); } @Override @@ -187,7 +182,7 @@ public class VirtualSocket extends Socket implements ChannelHandler { @Override public String toString() { - return "Virtual socket InetAdress["+getInetAddress()+"], Port["+getPort()+"]"; + return "VirtualSocket{" + getInetAddress() + ":" + getPort() + "}"; } @Override diff --git a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/virtualsocket/VirtualSocketException.java b/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/virtualsocket/VirtualSocketException.java deleted file mode 100644 index 626ebe937e..0000000000 --- a/opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/virtualsocket/VirtualSocketException.java +++ /dev/null @@ -1,17 +0,0 @@ -/* - * Copyright (c) 2013 Cisco Systems, Inc. and others. All rights reserved. - * - * This program and the accompanying materials are made available under the - * terms of the Eclipse Public License v1.0 which accompanies this distribution, - * and is available at http://www.eclipse.org/legal/epl-v10.html - */ - -package org.opendaylight.controller.netconf.nettyutil.handler.ssh.virtualsocket; - -/** - * Exception class which provides notification about exceptional situations at the virtual socket layer. - */ -// FIXME: Switch to checked exception, create a runtime exception to workaround Socket API -public class VirtualSocketException extends RuntimeException { - private static final long serialVersionUID = 1L; -} diff --git a/opendaylight/netconf/netconf-ssh/pom.xml b/opendaylight/netconf/netconf-ssh/pom.xml index 8a2387d2c1..febf3abf8e 100644 --- a/opendaylight/netconf/netconf-ssh/pom.xml +++ b/opendaylight/netconf/netconf-ssh/pom.xml @@ -49,6 +49,16 @@ mockito-configuration test + + org.opendaylight.controller + netconf-netty-util + test + + + org.opendaylight.controller + netconf-client + test + diff --git a/opendaylight/netconf/netconf-ssh/src/main/java/org/opendaylight/controller/netconf/ssh/NetconfSSHServer.java b/opendaylight/netconf/netconf-ssh/src/main/java/org/opendaylight/controller/netconf/ssh/NetconfSSHServer.java index 08bf9836b2..670f50ddd0 100644 --- a/opendaylight/netconf/netconf-ssh/src/main/java/org/opendaylight/controller/netconf/ssh/NetconfSSHServer.java +++ b/opendaylight/netconf/netconf-ssh/src/main/java/org/opendaylight/controller/netconf/ssh/NetconfSSHServer.java @@ -7,9 +7,11 @@ */ package org.opendaylight.controller.netconf.ssh; +import com.google.common.annotations.VisibleForTesting; import io.netty.channel.EventLoopGroup; import io.netty.channel.local.LocalAddress; import java.io.IOException; +import java.net.InetSocketAddress; import java.net.ServerSocket; import java.net.Socket; import java.util.concurrent.ExecutorService; @@ -68,6 +70,11 @@ public final class NetconfSSHServer extends Thread implements AutoCloseable { logger.trace("SSH server socket closed."); } + @VisibleForTesting + public InetSocketAddress getLocalSocketAddress() { + return (InetSocketAddress) serverSocket.getLocalSocketAddress(); + } + @Override public void run() { while (up) { diff --git a/opendaylight/netconf/netconf-ssh/src/main/java/org/opendaylight/controller/netconf/ssh/threads/Handshaker.java b/opendaylight/netconf/netconf-ssh/src/main/java/org/opendaylight/controller/netconf/ssh/threads/Handshaker.java index 8045d32a50..6300c56e72 100644 --- a/opendaylight/netconf/netconf-ssh/src/main/java/org/opendaylight/controller/netconf/ssh/threads/Handshaker.java +++ b/opendaylight/netconf/netconf-ssh/src/main/java/org/opendaylight/controller/netconf/ssh/threads/Handshaker.java @@ -32,6 +32,7 @@ import io.netty.channel.EventLoopGroup; import io.netty.channel.local.LocalAddress; import io.netty.channel.local.LocalChannel; import io.netty.handler.stream.ChunkedStream; +import java.io.BufferedOutputStream; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; @@ -119,14 +120,14 @@ public class Handshaker implements Runnable { class SSHClientHandler extends ChannelInboundHandlerAdapter { private static final Logger logger = LoggerFactory.getLogger(SSHClientHandler.class); private final AutoCloseable remoteConnection; - private final OutputStream remoteOutputStream; + private final BufferedOutputStream remoteOutputStream; private final String session; private ChannelHandlerContext channelHandlerContext; public SSHClientHandler(AutoCloseable remoteConnection, OutputStream remoteOutputStream, String session) { this.remoteConnection = remoteConnection; - this.remoteOutputStream = remoteOutputStream; + this.remoteOutputStream = new BufferedOutputStream(remoteOutputStream); this.session = session; } @@ -137,7 +138,7 @@ class SSHClientHandler extends ChannelInboundHandlerAdapter { } @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) { + public void channelRead(ChannelHandlerContext ctx, Object msg) throws IOException { ByteBuf bb = (ByteBuf) msg; // we can block the server here so that slow client does not cause memory pressure try { diff --git a/opendaylight/netconf/netconf-ssh/src/test/java/org/opendaylight/controller/netconf/netty/EchoClient.java b/opendaylight/netconf/netconf-ssh/src/test/java/org/opendaylight/controller/netconf/netty/EchoClient.java index 5d0c71aa62..b768e2b1d1 100644 --- a/opendaylight/netconf/netconf-ssh/src/test/java/org/opendaylight/controller/netconf/netty/EchoClient.java +++ b/opendaylight/netconf/netconf-ssh/src/test/java/org/opendaylight/controller/netconf/netty/EchoClient.java @@ -25,29 +25,36 @@ import org.slf4j.LoggerFactory; * traffic between the echo client and server by sending the first message to * the server. */ -public class EchoClient implements Runnable { +public class EchoClient extends Thread { private static final Logger logger = LoggerFactory.getLogger(EchoClient.class); - private final ChannelHandler clientHandler; + private final ChannelInitializer channelInitializer; - public EchoClient(ChannelHandler clientHandler) { - this.clientHandler = clientHandler; + + public EchoClient(final ChannelHandler clientHandler) { + channelInitializer = new ChannelInitializer() { + @Override + public void initChannel(LocalChannel ch) throws Exception { + ch.pipeline().addLast(clientHandler); + } + }; + } + + public EchoClient(ChannelInitializer channelInitializer) { + this.channelInitializer = channelInitializer; } + @Override public void run() { // Configure the client. EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap b = new Bootstrap(); + b.group(group) .channel(LocalChannel.class) - .handler(new ChannelInitializer() { - @Override - public void initChannel(LocalChannel ch) throws Exception { - ch.pipeline().addLast(clientHandler); - } - }); + .handler(channelInitializer); // Start the client. LocalAddress localAddress = new LocalAddress("foo"); diff --git a/opendaylight/netconf/netconf-ssh/src/test/java/org/opendaylight/controller/netconf/netty/EchoClientHandler.java b/opendaylight/netconf/netconf-ssh/src/test/java/org/opendaylight/controller/netconf/netty/EchoClientHandler.java index 81182a580e..2a5791710a 100644 --- a/opendaylight/netconf/netconf-ssh/src/test/java/org/opendaylight/controller/netconf/netty/EchoClientHandler.java +++ b/opendaylight/netconf/netconf-ssh/src/test/java/org/opendaylight/controller/netconf/netty/EchoClientHandler.java @@ -13,6 +13,8 @@ import static com.google.common.base.Preconditions.checkState; import com.google.common.base.Charsets; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; +import io.netty.channel.ChannelFuture; +import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; import org.slf4j.Logger; @@ -23,31 +25,41 @@ import org.slf4j.LoggerFactory; * traffic between the echo client and server by sending the first message to * the server. */ -public class EchoClientHandler extends ChannelInboundHandlerAdapter { +public class EchoClientHandler extends ChannelInboundHandlerAdapter implements ChannelFutureListener { private static final Logger logger = LoggerFactory.getLogger(EchoClientHandler.class); private ChannelHandlerContext ctx; + private final StringBuilder fromServer = new StringBuilder(); + + public static enum State {CONNECTING, CONNECTED, FAILED_TO_CONNECT, CONNECTION_CLOSED} + + + private State state = State.CONNECTING; @Override - public void channelActive(ChannelHandlerContext ctx) { + public synchronized void channelActive(ChannelHandlerContext ctx) { checkState(this.ctx == null); - logger.info("client active"); + logger.info("channelActive"); this.ctx = ctx; + state = State.CONNECTED; } @Override - public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { - ByteBuf bb = (ByteBuf) msg; - logger.info(">{}", bb.toString(Charsets.UTF_8)); - bb.release(); + public synchronized void channelInactive(ChannelHandlerContext ctx) throws Exception { + state = State.CONNECTION_CLOSED; } @Override - public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { + public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { + ByteBuf bb = (ByteBuf) msg; + String string = bb.toString(Charsets.UTF_8); + fromServer.append(string); + logger.info(">{}", string); + bb.release(); } @Override - public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { + public synchronized void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // Close the connection when an exception is raised. logger.warn("Unexpected exception from downstream.", cause); checkState(this.ctx.equals(ctx)); @@ -55,8 +67,30 @@ public class EchoClientHandler extends ChannelInboundHandlerAdapter { this.ctx = null; } - public void write(String message) { + public synchronized void write(String message) { ByteBuf byteBuf = Unpooled.copiedBuffer(message.getBytes()); ctx.writeAndFlush(byteBuf); } + + public synchronized boolean isConnected() { + return state == State.CONNECTED; + } + + public synchronized String read() { + return fromServer.toString(); + } + + @Override + public synchronized void operationComplete(ChannelFuture future) throws Exception { + checkState(state == State.CONNECTING); + if (future.isSuccess()) { + logger.trace("Successfully connected, state will be switched in channelActive"); + } else { + state = State.FAILED_TO_CONNECT; + } + } + + public State getState() { + return state; + } } diff --git a/opendaylight/netconf/netconf-ssh/src/test/java/org/opendaylight/controller/netconf/netty/SSHTest.java b/opendaylight/netconf/netconf-ssh/src/test/java/org/opendaylight/controller/netconf/netty/SSHTest.java index 2bda51b495..488c370145 100644 --- a/opendaylight/netconf/netconf-ssh/src/test/java/org/opendaylight/controller/netconf/netty/SSHTest.java +++ b/opendaylight/netconf/netconf-ssh/src/test/java/org/opendaylight/controller/netconf/netty/SSHTest.java @@ -8,12 +8,28 @@ package org.opendaylight.controller.netconf.netty; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; import static org.mockito.Matchers.anyString; import static org.mockito.Mockito.doReturn; import static org.mockito.Mockito.mock; +import com.google.common.base.Stopwatch; +import io.netty.bootstrap.Bootstrap; +import io.netty.channel.ChannelInitializer; +import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.socket.nio.NioSocketChannel; +import io.netty.util.HashedWheelTimer; +import java.net.InetSocketAddress; +import java.util.concurrent.TimeUnit; +import org.junit.After; +import org.junit.Before; import org.junit.Test; +import org.opendaylight.controller.netconf.netty.EchoClientHandler.State; +import org.opendaylight.controller.netconf.nettyutil.handler.ssh.authentication.LoginPassword; +import org.opendaylight.controller.netconf.nettyutil.handler.ssh.client.SshHandler; import org.opendaylight.controller.netconf.ssh.NetconfSSHServer; import org.opendaylight.controller.netconf.ssh.authentication.AuthProvider; import org.opendaylight.controller.netconf.ssh.authentication.PEMGenerator; @@ -23,6 +39,21 @@ import org.slf4j.LoggerFactory; public class SSHTest { public static final Logger logger = LoggerFactory.getLogger(SSHTest.class); + public static final String AHOJ = "ahoj\n"; + private EventLoopGroup nettyGroup; + HashedWheelTimer hashedWheelTimer; + + @Before + public void setUp() throws Exception { + hashedWheelTimer = new HashedWheelTimer(); + nettyGroup = new NioEventLoopGroup(); + } + + @After + public void tearDown() throws Exception { + hashedWheelTimer.stop(); + nettyGroup.shutdownGracefully(); + } @Test public void test() throws Exception { @@ -30,10 +61,63 @@ public class SSHTest { AuthProvider authProvider = mock(AuthProvider.class); doReturn(PEMGenerator.generate().toCharArray()).when(authProvider).getPEMAsCharArray(); doReturn(true).when(authProvider).authenticated(anyString(), anyString()); - NetconfSSHServer thread = NetconfSSHServer.start(10831, NetconfConfigUtil.getNetconfLocalAddress(), authProvider, new NioEventLoopGroup()); - Thread.sleep(2000); - logger.info("Closing socket"); - thread.close(); - thread.join(); + NetconfSSHServer netconfSSHServer = NetconfSSHServer.start(10831, NetconfConfigUtil.getNetconfLocalAddress(), + authProvider, new NioEventLoopGroup()); + + InetSocketAddress address = netconfSSHServer.getLocalSocketAddress(); + final EchoClientHandler echoClientHandler = connectClient(address); + Stopwatch stopwatch = new Stopwatch().start(); + while(echoClientHandler.isConnected() == false && stopwatch.elapsed(TimeUnit.SECONDS) < 5) { + Thread.sleep(100); + } + assertTrue(echoClientHandler.isConnected()); + logger.info("connected, writing to client"); + echoClientHandler.write(AHOJ); + // check that server sent back the same string + stopwatch = stopwatch.reset().start(); + while (echoClientHandler.read().endsWith(AHOJ) == false && stopwatch.elapsed(TimeUnit.SECONDS) < 5) { + Thread.sleep(100); + } + try { + String read = echoClientHandler.read(); + assertTrue(read + " should end with " + AHOJ, read.endsWith(AHOJ)); + } finally { + logger.info("Closing socket"); + netconfSSHServer.close(); + netconfSSHServer.join(); + } } + + public EchoClientHandler connectClient(InetSocketAddress address) { + final EchoClientHandler echoClientHandler = new EchoClientHandler(); + ChannelInitializer channelInitializer = new ChannelInitializer() { + @Override + public void initChannel(NioSocketChannel ch) throws Exception { + ch.pipeline().addFirst(SshHandler.createForNetconfSubsystem(new LoginPassword("a", "a"))); + ch.pipeline().addLast(echoClientHandler); + } + }; + Bootstrap b = new Bootstrap(); + + b.group(nettyGroup) + .channel(NioSocketChannel.class) + .handler(channelInitializer); + + // Start the client. + b.connect(address).addListener(echoClientHandler); + return echoClientHandler; + } + + @Test + public void testClientWithoutServer() throws Exception { + InetSocketAddress address = new InetSocketAddress(12345); + final EchoClientHandler echoClientHandler = connectClient(address); + Stopwatch stopwatch = new Stopwatch().start(); + while(echoClientHandler.getState() == State.CONNECTING && stopwatch.elapsed(TimeUnit.SECONDS) < 5) { + Thread.sleep(100); + } + assertFalse(echoClientHandler.isConnected()); + assertEquals(State.FAILED_TO_CONNECT, echoClientHandler.getState()); + } + }