allDataReadEventTriggeredAfterHalfClosure() — netty Function Reference

Source Code

testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketHalfClosedTest.java lines 83–199

    private void allDataReadEventTriggeredAfterHalfClosure(ServerBootstrap sb, Bootstrap cb) throws Throwable {
        final int totalServerBytesWritten = 1;
        final CountDownLatch clientReadAllDataLatch = new CountDownLatch(1);
        final CountDownLatch clientHalfClosedLatch = new CountDownLatch(1);
        final CountDownLatch clientHalfClosedAllBytesRead = new CountDownLatch(1);
        final AtomicInteger clientReadCompletes = new AtomicInteger();
        final AtomicInteger clientZeroDataReadCompletes = new AtomicInteger();
        Channel serverChannel = null;
        Channel clientChannel = null;
        AtomicReference<Channel> serverChildChannel = new AtomicReference<>();
        try {
            cb.option(ChannelOption.ALLOW_HALF_CLOSURE, true)
                    .option(ChannelOption.AUTO_CLOSE, false)
                    .option(ChannelOption.AUTO_READ, false);

            sb.option(ChannelOption.ALLOW_HALF_CLOSURE, true)
                    .option(ChannelOption.AUTO_CLOSE, false)
                    .childOption(ChannelOption.TCP_NODELAY, true);

            sb.childHandler(new ChannelInitializer<Channel>() {
                @Override
                protected void initChannel(Channel ch) throws Exception {
                    serverChildChannel.set(ch);
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            ByteBuf buf = ctx.alloc().buffer(totalServerBytesWritten);
                            buf.writerIndex(buf.capacity());
                            ctx.writeAndFlush(buf);
                        }

                        @Override
                        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
                            ctx.close();
                        }
                    });
                }
            });

            // client.
            cb.handler(new ChannelInitializer<Channel>() {
                @Override
                protected void initChannel(Channel ch) {
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        private int bytesRead;
                        private int bytesSinceReadComplete;

                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) {
                            ByteBuf buf = (ByteBuf) msg;
                            bytesRead += buf.readableBytes();
                            bytesSinceReadComplete += buf.readableBytes();
                            buf.release();
                        }

                        @Override
                        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
                            if (evt == ChannelInputShutdownEvent.INSTANCE) {
                                clientHalfClosedLatch.countDown();
                            } else if (evt == ChannelInputShutdownReadComplete.INSTANCE) {
                                clientHalfClosedAllBytesRead.countDown();
                                ctx.close();
                            }
                        }

                        @Override
                        public void channelReadComplete(ChannelHandlerContext ctx) {
                            if (bytesSinceReadComplete == 0) {
                                clientZeroDataReadCompletes.incrementAndGet();
                            } else {
                                bytesSinceReadComplete = 0;
                            }
                            clientReadCompletes.incrementAndGet();
                            if (bytesRead == totalServerBytesWritten) {
                                // Bounce this through the event loop to make sure it happens after we're done
                                // with the read operation.
                                ch.eventLoop().execute(new Runnable() {
                                    @Override
                                    public void run() {
                                        clientReadAllDataLatch.countDown();
                                    }