UDTClientServerConnectionTest Class — netty Architecture

Source Code

testsuite/src/main/java/io/netty/testsuite/transport/udt/UDTClientServerConnectionTest.java lines 57–400

public class UDTClientServerConnectionTest {

    static class Client implements Runnable {

        static final Logger log = LoggerFactory.getLogger(Client.class);

        private final InetSocketAddress address;

        volatile Channel channel;
        volatile boolean isRunning;
        volatile boolean isShutdown;

        Client(InetSocketAddress address) {
            this.address = address;
        }

        @Override
        public void run() {
            final Bootstrap boot = new Bootstrap();
            final ThreadFactory clientFactory = new DefaultThreadFactory("client");
            final EventLoopGroup connectGroup = new MultiThreadIoEventLoopGroup(1,
                    clientFactory, NioIoHandler.newFactory(NioUdtProvider.BYTE_PROVIDER));
            try {
                boot.group(connectGroup)
                        .channelFactory(NioUdtProvider.BYTE_CONNECTOR)
                        .handler(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 ClientHandler());
                            }
                        });
                channel = boot.connect(address).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().syncUninterruptibly();
            }
        }

        void shutdown() {
            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 {