serverCloseWithoutClientReadIsDetected0() — netty Function Reference

Source Code

transport-native-unix-common-tests/src/main/java/io/netty/channel/unix/tests/DetectPeerCloseWithoutReadTest.java lines 122–185

    private void serverCloseWithoutClientReadIsDetected0(final boolean extraReadRequested) throws InterruptedException {
        EventLoopGroup serverGroup = null;
        EventLoopGroup clientGroup = null;
        Channel serverChannel = null;
        Channel clientChannel = null;
        try {
            final CountDownLatch latch = new CountDownLatch(1);
            final AtomicInteger bytesRead = new AtomicInteger();
            final int expectedBytes = 100;
            serverGroup = newGroup();
            clientGroup = newGroup();
            ServerBootstrap sb = new ServerBootstrap();
            sb.group(serverGroup);
            sb.channel(serverChannel());
            sb.childHandler(new ChannelInitializer<Channel>() {
                @Override
                protected void initChannel(Channel ch) {
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) {
                            ByteBuf buf = ctx.alloc().buffer(expectedBytes);
                            buf.writerIndex(buf.writerIndex() + expectedBytes);
                            ctx.writeAndFlush(buf).addListener(ChannelFutureListener.CLOSE);
                            ctx.fireChannelActive();
                        }
                    });
                }
            });

            serverChannel = sb.bind(new InetSocketAddress(0)).syncUninterruptibly().channel();

            Bootstrap cb = new Bootstrap();
            cb.group(serverGroup);
            cb.channel(clientChannel());
            // Ensure we read only one message per read() call and that we need multiple read()
            // calls to consume everything.
            cb.option(ChannelOption.AUTO_READ, false);
            cb.option(ChannelOption.MAX_MESSAGES_PER_READ, 1);
            cb.option(ChannelOption.RECVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(expectedBytes / 10));
            cb.handler(new ChannelInitializer<Channel>() {
                @Override
                protected void initChannel(Channel ch) throws Exception {
                    ch.pipeline().addLast(new TestHandler(bytesRead, extraReadRequested, latch));
                }
            });
            clientChannel = cb.connect(serverChannel.localAddress()).syncUninterruptibly().channel();

            latch.await();
            assertEquals(expectedBytes, bytesRead.get());
        } finally {
            if (serverChannel != null) {
                serverChannel.close().syncUninterruptibly();
            }
            if (clientChannel != null) {
                clientChannel.close().syncUninterruptibly();
            }
            if (serverGroup != null) {
                serverGroup.shutdownGracefully();
            }
            if (clientGroup != null) {
                clientGroup.shutdownGracefully();
            }
        }
    }