SocketHalfClosedTest Class — netty Architecture

Source Code

testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketHalfClosedTest.java lines 56–865

@Timeout(value = 20000, unit = MILLISECONDS)
public class SocketHalfClosedTest extends AbstractSocketTest {

    protected int maxReadCompleteWithNoDataAfterInputShutdown() {
        return 2; // nio needs read flag to detect full closure.
    }

    @Test
    public void testAllDataReadEventTriggeredAfterHalfClosure(TestInfo testInfo) throws Throwable {
        run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
            @Override
            public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
                if (bootstrap.config().group() instanceof OioEventLoopGroup) {
                    logger.debug("Ignoring test for incompatible OIO event system");
                    return;
                } else if (bootstrap.config().group() instanceof IoEventLoopGroup) {
                    IoEventLoopGroup group = (IoEventLoopGroup) bootstrap.config().group();
                    if (group.isIoType(NioIoHandler.class)) {
                        logger.debug("Ignoring test for incompatible NioHandler");
                        return;
                    }
                }
                allDataReadEventTriggeredAfterHalfClosure(serverBootstrap, bootstrap);
            }
        });
    }

    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();
                        }