Server Class — netty Architecture

Source Code

testsuite/src/main/java/io/netty/testsuite/transport/udt/UDTClientServerConnectionTest.java lines 182–281

    static class Server implements Runnable {

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

        final ChannelGroup group = new DefaultChannelGroup("server group", GlobalEventExecutor.INSTANCE);

        private final InetSocketAddress address;

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

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

        @Override
        public void run() {
            final ServerBootstrap boot = new ServerBootstrap();
            final ThreadFactory factory = new DefaultThreadFactory("udp");
            final EventLoopGroup eventLoopGroup = new MultiThreadIoEventLoopGroup(1,
                    factory, NioIoHandler.newFactory(NioUdtProvider.BYTE_PROVIDER));
            try {
                boot.group(eventLoopGroup)
                        .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));
                            }
                        });
                channel = boot.bind(address).sync().channel();
                isRunning = true;
                log.info("Server ready.");
                waitForRunning(false);
                log.info("Server closing acceptor...");
                channel.close().sync();
                log.info("Server closing connectors...");
                group.close().sync();
                isShutdown = true;
                log.info("Server is done.");
            } catch (final Throwable e) {
                log.error("Server failure.", e);
            } finally {
                eventLoopGroup.shutdownGracefully();
                eventLoopGroup.terminationFuture().syncUninterruptibly();
            }
        }

        void shutdown() {
            isRunning = false;
        }

        void waitForActive(final boolean isActive) throws Exception {
            for (int k = 0; k < WAIT_COUNT; k++) {
                Thread.sleep(WAIT_SLEEP);
                if (isActive) {
                    for (final Channel channel : group) {
                        final ServerHandler handler = channel.pipeline().get(
                                ServerHandler.class);
                        if (handler != null && handler.isActive) {
                            return;
                        }
                    }
                } else {
                    if (group.isEmpty()) {
                        return;
                    }
                }
            }
        }