updated udt and connection test

pom.xml

@@ -144,7 +144,7 @@
       <dependency>
         <groupId>com.barchart.udt</groupId>
         <artifactId>barchart-udt-bundle</artifactId>
-        <version>2.2.6</version>
+        <version>2.3.0-SNAPSHOT</version>
       </dependency>
 
       <dependency>

testsuite/.gitignore

@@ -0,0 +1,6 @@
+
+#
+# UDT native libraries extract location.
+#
+
+/lib

testsuite/src/test/java/io/netty/testsuite/transport/udt/UDTClientServerConnectionTest.java

@@ -15,10 +15,9 @@
  */
 package io.netty.testsuite.transport.udt;
 
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.*;
 import io.netty.bootstrap.Bootstrap;
 import io.netty.bootstrap.ServerBootstrap;
-import io.netty.buffer.BufType;
 import io.netty.channel.Channel;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundMessageHandlerAdapter;
@@ -33,164 +32,251 @@ import io.netty.handler.codec.DelimiterBasedFrameDecoder;
 import io.netty.handler.codec.Delimiters;
 import io.netty.handler.codec.string.StringDecoder;
 import io.netty.handler.codec.string.StringEncoder;
-import io.netty.testsuite.transport.udt.UDTClientServerConnectionTest.Server.TestThreadFactory;
 import io.netty.util.CharsetUtil;
 
-import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.logging.Level;
-import java.util.logging.Logger;
 
 import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
+/**
+ * Verify UDT connect/disconnect life cycle.
+ */
 public class UDTClientServerConnectionTest {
 
-    @Test
+    static class Client implements Runnable {
-    public void test() throws InterruptedException {
-        // first start server
 
-        int port = 1234;
+        static final Logger log = LoggerFactory.getLogger(Client.class);
-        Server server = new Server(port);
-        Thread serverTread = new Thread(server);
-        serverTread.start();
 
-        Thread.sleep(1000);
+        final String host;
+        final int port;
 
-        // start a client
+        volatile Channel channel;
-        final String host = "localhost";
+        volatile boolean isRunning;
+        volatile boolean isShutdown;
 
-        TestClient client = new TestClient(host, port);
+        Client(final String host, final int port) {
-        Thread clientThread = new Thread(client);
+            this.host = host;
-        clientThread.start();
-
-        Thread.sleep(1000);
-
-        // check number of connections
-        assertTrue(server.connectedClients() == 1);
-
-        // close client
-        client.shutdown();
-
-        Thread.sleep(1000);
-
-        // check connections again
-        assertTrue(server.connectedClients() == 0);
-    }
-
-    static class Server implements Runnable {
-        private final ChannelGroup channels = new DefaultChannelGroup(
-                "all channels");
-        private Channel serverChannel;
-
-        private static final Logger log = Logger.getLogger(Server.class
-                .getName());
-
-        private final int port;
-        private ServerBootstrap b;
-        private boolean running;
-        private boolean shutdown;
-
-        public Server(final int port) {
             this.port = port;
         }
 
-        public void shutdown() {
+        @Override
-            log.info("shutting down server...");
-            running = false;
-            while (!shutdown) {
-                try {
-                    Thread.sleep(100);
-                } catch (InterruptedException e) {
-                    //ignore
-                }
-            }
-        }
-
         public void run() {
-            running = true;
+            final Bootstrap boot = new Bootstrap();
-            b = new ServerBootstrap();
+            final ThreadFactory clientFactory = new ThreadFactory("client");
-            final ThreadFactory acceptFactory = new TestThreadFactory("accept");
-            final ThreadFactory connectFactory = new TestThreadFactory(
-                    "connect");
-            final NioEventLoopGroup acceptGroup = new NioEventLoopGroup(1,
-                    acceptFactory, NioUdtProvider.BYTE_PROVIDER);
             final NioEventLoopGroup connectGroup = new NioEventLoopGroup(1,
-                    connectFactory, NioUdtProvider.BYTE_PROVIDER);
+                    clientFactory, NioUdtProvider.BYTE_PROVIDER);
             try {
-                // Configure the server.
+                boot.group(connectGroup)
-                b.group(acceptGroup, connectGroup)
+                        .channelFactory(NioUdtProvider.BYTE_CONNECTOR)
-                        .channelFactory(NioUdtProvider.BYTE_ACCEPTOR)
+                        .handler(new ChannelInitializer<UdtChannel>() {
-                        .childHandler(new ChannelInitializer<UdtChannel>() {
 
                             @Override
-                            protected void initChannel(UdtChannel ch)
+                            protected void initChannel(final UdtChannel ch)
                                     throws Exception {
-                                ChannelPipeline pipeline = ch.pipeline();
+                                final ChannelPipeline pipeline = ch.pipeline();
                                 pipeline.addLast("framer",
                                         new DelimiterBasedFrameDecoder(8192,
                                                 Delimiters.lineDelimiter()));
                                 pipeline.addLast("decoder", new StringDecoder(
                                         CharsetUtil.UTF_8));
                                 pipeline.addLast("encoder", new StringEncoder(
-                                        BufType.BYTE));
+                                        CharsetUtil.UTF_8));
+                                pipeline.addLast("handler", new ClientHandler());
+                            }
+                        });
+                channel = boot.connect(host, port).sync().channel();
+                isRunning = true;
+                log.info("Client ready.");
+                waitForRunning(false);
+                log.info("Client closing...");
+                channel.close().sync();
+                isShutdown = true;
+                log.info("Client is done.");
+            } catch (final Throwable e) {
+                log.error("Client failed.", e);
+            } finally {
+                connectGroup.shutdownGracefully();
+            }
+        }
 
-                                pipeline.addLast("handler", new ServerHandler());
+        void shutdown() {
-                                channels.add(ch);
+            isRunning = false;
+        }
+
+        void waitForActive(final boolean isActive) throws Exception {
+            for (int k = 0; k < WAIT_COUNT; k++) {
+                Thread.sleep(WAIT_SLEEP);
+                final ClientHandler handler = channel.pipeline().get(
+                        ClientHandler.class);
+                if (handler != null && isActive == handler.isActive) {
+                    return;
+                }
+            }
+        }
+
+        void waitForRunning(final boolean isRunning) throws Exception {
+            for (int k = 0; k < WAIT_COUNT; k++) {
+                if (isRunning == this.isRunning) {
+                    return;
+                }
+                Thread.sleep(WAIT_SLEEP);
+            }
+        }
+
+        private void waitForShutdown() throws Exception {
+            for (int k = 0; k < WAIT_COUNT; k++) {
+                if (isShutdown) {
+                    return;
+                }
+                Thread.sleep(WAIT_SLEEP);
+            }
+        }
+    }
+
+    static class ClientHandler extends
+            ChannelInboundMessageHandlerAdapter<String> {
+
+        static final Logger log = LoggerFactory.getLogger(ClientHandler.class);
+
+        volatile boolean isActive;
+
+        ClientHandler() {
         }
 
         @Override
-                            public void channelInactive(
+        public void channelActive(final ChannelHandlerContext ctx)
-                                    ChannelHandlerContext ctx) throws Exception {
+                throws Exception {
-                                log.log(Level.INFO,
+            isActive = true;
-                                        "channel inactive, removing from channelgroup");
+            log.info("Client active {}", ctx.channel());
-                                channels.remove(ctx.channel());
+            super.channelActive(ctx);
+        }
+
+        @Override
+        public void channelInactive(final ChannelHandlerContext ctx)
+                throws Exception {
+            isActive = false;
+            log.info("Client inactive {}", ctx.channel());
+            super.channelInactive(ctx);
+        }
+
+        @Override
+        public void exceptionCaught(final ChannelHandlerContext ctx,
+                final Throwable cause) throws Exception {
+            log.warn("Client unexpected exception from downstream.", cause);
+            ctx.close();
+        }
+
+        @Override
+        public void messageReceived(final ChannelHandlerContext ctx,
+                final String msg) throws Exception {
+            log.info("Client received: " + msg);
+        }
+    }
+
+    static class Server implements Runnable {
+
+        static final Logger log = LoggerFactory.getLogger(Server.class);
+
+        final ChannelGroup group = new DefaultChannelGroup("server group");
+
+        final String host;
+        final int port;
+
+        volatile Channel channel;
+        volatile boolean isRunning;
+        volatile boolean isShutdown;
+
+        Server(final String host, final int port) {
+            this.host = host;
+            this.port = port;
+        }
+
+        @Override
+        public void run() {
+            final ServerBootstrap boot = new ServerBootstrap();
+            final ThreadFactory acceptFactory = new ThreadFactory("accept");
+            final ThreadFactory serverFactory = new ThreadFactory("server");
+            final NioEventLoopGroup acceptGroup = new NioEventLoopGroup(1,
+                    acceptFactory, NioUdtProvider.BYTE_PROVIDER);
+            final NioEventLoopGroup connectGroup = new NioEventLoopGroup(1,
+                    serverFactory, NioUdtProvider.BYTE_PROVIDER);
+            try {
+                boot.group(acceptGroup, connectGroup)
+                        .channelFactory(NioUdtProvider.BYTE_ACCEPTOR)
+                        .childHandler(new ChannelInitializer<UdtChannel>() {
+                            @Override
+                            protected void initChannel(final UdtChannel ch)
+                                    throws Exception {
+                                final ChannelPipeline pipeline = ch.pipeline();
+                                pipeline.addLast("framer",
+                                        new DelimiterBasedFrameDecoder(8192,
+                                                Delimiters.lineDelimiter()));
+                                pipeline.addLast("decoder", new StringDecoder(
+                                        CharsetUtil.UTF_8));
+                                pipeline.addLast("encoder", new StringEncoder(
+                                        CharsetUtil.UTF_8));
+                                pipeline.addLast("handler", new ServerHandler(
+                                        group));
                             }
                         });
-                // Start the server.
+                channel = boot.bind(port).sync().channel();
-                serverChannel = b.bind(port).sync().channel();
+                isRunning = true;
-                waitForShutdownCommand();
+                log.info("Server ready.");
-                log.info("closing server channel...");
+                waitForRunning(false);
-                serverChannel.close().sync();
+                log.info("Server closing acceptor...");
-                log.info("closing all accepted gateway channels...");
+                channel.close().sync();
-                channels.close().sync();
+                log.info("Server closing connectors...");
-                log.info("channels closed");
+                group.close().sync();
-            } catch (Exception e) {
+                isShutdown = true;
-                log.log(Level.SEVERE, "GATEWAY SERVER DIED!", e);
+                log.info("Server is done.");
+            } catch (final Throwable e) {
+                log.error("Server failure.", e);
             } finally {
                 acceptGroup.shutdownGracefully();
                 connectGroup.shutdownGracefully();
             }
-            shutdown = true;
         }
 
-        public int connectedClients() {
+        void shutdown() {
-            return channels.size();
+            isRunning = false;
         }
 
-        private void waitForShutdownCommand() {
+        void waitForActive(final boolean isActive) throws Exception {
-            while (running) {
+            for (int k = 0; k < WAIT_COUNT; k++) {
-                try {
+                Thread.sleep(WAIT_SLEEP);
-                    Thread.sleep(100);
+                if (isActive) {
-                } catch (InterruptedException e) {
+                    for (final Channel channel : group) {
-                    e.printStackTrace();
+                        final ServerHandler handler = channel.pipeline().get(
+                                ServerHandler.class);
+                        if (handler != null && handler.isActive) {
+                            return;
+                        }
+                    }
+                } else {
+                    if (group.size() == 0) {
+                        return;
+                    }
                 }
             }
         }
 
-        static class TestThreadFactory implements ThreadFactory {
+        void waitForRunning(final boolean isRunning) throws Exception {
-
+            for (int k = 0; k < WAIT_COUNT; k++) {
-            private static final AtomicInteger counter = new AtomicInteger();
+                if (isRunning == this.isRunning) {
-
+                    return;
-            private final String name;
+                }
-
+                Thread.sleep(WAIT_SLEEP);
-            public TestThreadFactory(final String name) {
+            }
-                this.name = name;
         }
 
-            @Override
+        void waitForShutdown() throws Exception {
-            public Thread newThread(final Runnable runnable) {
+            for (int k = 0; k < WAIT_COUNT; k++) {
-                return new Thread(runnable, name + '-'
+                if (isShutdown) {
-                        + counter.getAndIncrement());
+                    return;
+                }
+                Thread.sleep(WAIT_SLEEP);
             }
         }
     }
@@ -198,140 +284,123 @@ public class UDTClientServerConnectionTest {
     static class ServerHandler extends
             ChannelInboundMessageHandlerAdapter<String> {
 
-        private boolean isClosed;
+        static final Logger log = LoggerFactory.getLogger(ServerHandler.class);
-        private static final Logger log = Logger.getLogger(ServerHandler.class
+
-                .getName());
+        final ChannelGroup group;
+
+        volatile boolean isActive;
+
+        ServerHandler(final ChannelGroup group) {
+            this.group = group;
+        }
+
+        @Override
+        public void channelActive(final ChannelHandlerContext ctx)
+                throws Exception {
+            group.add(ctx.channel());
+            isActive = true;
+            log.info("Server active  : {}", ctx.channel());
+            super.channelActive(ctx);
+        }
+
+        @Override
+        public void channelInactive(final ChannelHandlerContext ctx)
+                throws Exception {
+            group.remove(ctx.channel());
+            isActive = false;
+            log.info("Server inactive: {}", ctx.channel());
+            super.channelInactive(ctx);
+        }
 
         @Override
         public void exceptionCaught(final ChannelHandlerContext ctx,
                 final Throwable cause) {
-            log.log(Level.WARNING,
+            log.warn("Server close on exception.", cause);
-                    "close the connection when an exception is raised", cause);
             ctx.close();
         }
 
         @Override
-        public void messageReceived(ChannelHandlerContext ctx, String message)
+        public void messageReceived(final ChannelHandlerContext ctx,
-                throws Exception {
+                final String message) throws Exception {
-            log.log(Level.INFO, "received: " + message);
+            log.info("Server received: " + message);
+        }
+    }
+
+    static class ThreadFactory implements java.util.concurrent.ThreadFactory {
+
+        static final AtomicInteger counter = new AtomicInteger();
+
+        final String name;
+
+        ThreadFactory(final String name) {
+            this.name = name;
         }
 
         @Override
-        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        public Thread newThread(final Runnable runnable) {
-            log.log(Level.INFO, "channel inactive");
+            return new Thread(runnable, name + '-' + counter.getAndIncrement());
-            isClosed = true;
-        }
-
-        @Override
-        public void channelUnregistered(ChannelHandlerContext ctx)
-                throws Exception {
-            log.log(Level.INFO, "channel unregistered");
-        }
-
-        public boolean isClosed() {
-            return isClosed;
         }
     }
 
-    static class TestClient implements Runnable {
+    static final Logger log = LoggerFactory
+            .getLogger(UDTClientServerConnectionTest.class);
 
-        private static final Logger log = Logger.getLogger(TestClient.class
+    /**
-                .getName());
+     * Maximum wait time is 5 seconds.
+     * <p>
+     * wait-time = {@link #WAIT_COUNT} * {@value #WAIT_SLEEP}
+     */
+    static final int WAIT_COUNT = 50;
+    static final int WAIT_SLEEP = 100;
 
-        private final String host;
+    /**
-        private final int port;
+     * Verify UDT client/server connect and disconnect.
-        private Channel channel;
+     */
-        private boolean running;
+    @Test
+    public void connection() throws Exception {
 
-        public TestClient(final String host, final int port) {
+        final String host = "localhost";
-            this.host = host;
+        final int port = 1234;
-            this.port = port;
+
+        log.info("Starting server.");
+        final Server server = new Server(host, port);
+        final Thread serverTread = new Thread(server, "server-*");
+        serverTread.start();
+        server.waitForRunning(true);
+        assertTrue(server.isRunning);
+
+        log.info("Starting client.");
+        final Client client = new Client(host, port);
+        final Thread clientThread = new Thread(client, "client-*");
+        clientThread.start();
+        client.waitForRunning(true);
+        assertTrue(client.isRunning);
+
+        log.info("Wait till connection is active.");
+        client.waitForActive(true);
+        server.waitForActive(true);
+
+        log.info("Verify connection is active.");
+        assertEquals("group must have one", 1, server.group.size());
+
+        log.info("Stopping client.");
+        client.shutdown();
+        client.waitForShutdown();
+        assertTrue(client.isShutdown);
+
+        log.info("Wait till connection is inactive.");
+        client.waitForActive(false);
+        server.waitForActive(false);
+
+        log.info("Verify connection is inactive.");
+        assertEquals("group must be empty", 0, server.group.size());
+
+        log.info("Stopping server.");
+        server.shutdown();
+        server.waitForShutdown();
+        assertTrue(server.isShutdown);
+
+        log.info("Finished server.");
     }
 
-        public void run() {
-            running = true;
-            // Configure the client.
-            final Bootstrap boot = new Bootstrap();
-            final ThreadFactory connectFactory = new TestThreadFactory(
-                    "connect");
-            final NioEventLoopGroup connectGroup = new NioEventLoopGroup(1,
-                    connectFactory, NioUdtProvider.BYTE_PROVIDER);
-            try {
-                boot.group(connectGroup)
-                        .channelFactory(NioUdtProvider.BYTE_CONNECTOR)
-                        .handler(new ChannelInitializer<UdtChannel>() {
-
-                            @Override
-                            protected void initChannel(UdtChannel ch)
-                                    throws Exception {
-                                ChannelPipeline pipeline = ch.pipeline();
-
-                                // On top of the SSL handler, add the text line
-                                // codec.
-                                pipeline.addLast("framer",
-                                        new DelimiterBasedFrameDecoder(8192,
-                                                Delimiters.lineDelimiter()));
-                                pipeline.addLast("decoder", new StringDecoder());
-                                pipeline.addLast("encoder", new StringEncoder(
-                                        BufType.BYTE));
-
-                                // and then business logic.
-                                pipeline.addLast("handler", new TestClientHandler());
-                            }
-                        });
-                // Start the connection attempt.
-                channel = boot.connect(host, port).sync().channel();
-                waitForShutdownCommand();
-                channel.disconnect().sync();
-                channel.close().sync(); // close the channel and wait until done
-
-                // channel.closeFuture().sync(); //wait for the channel to close
-
-            } catch (Exception e) {
-                e.printStackTrace();
-            } finally {
-                // Shut down the event loop to terminate all threads.
-                connectGroup.shutdownGracefully();
-            }
-            log.info("test client done");
-        }
-
-        private void waitForShutdownCommand() {
-            while (running) {
-                try {
-                    Thread.sleep(100);
-                } catch (InterruptedException e) {
-                    e.printStackTrace();
-                }
-            }
-        }
-
-        public void shutdown() {
-            running = false;
-        }
-    }
-
-    static class TestClientHandler extends
-            ChannelInboundMessageHandlerAdapter<String> {
-
-        TestClientHandler() {
-        }
-
-        private static final Logger logger = Logger
-                .getLogger(TestClientHandler.class.getName());
-
-        @Override
-        public void messageReceived(ChannelHandlerContext ctx, String msg)
-                throws Exception {
-            logger.info("client received: " + msg);
-        }
-
-        @Override
-        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
-                throws Exception {
-            logger.log(Level.WARNING, "Unexpected exception from downstream.",
-                    cause);
-            ctx.close();
-        }
-    }
 }

testsuite/src/test/java/io/netty/testsuite/transport/udt/package-info.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2012 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * Test suite classes for UDT.
+ */
+package io.netty.testsuite.transport.udt;
+