SocketReadPendingTest Class — netty Architecture

Source Code

testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketReadPendingTest.java lines 44–213

public class SocketReadPendingTest extends AbstractSocketTest {
    @Test
    @Timeout(value = 60000, unit = TimeUnit.MILLISECONDS)
    public void testReadPendingIsResetAfterEachRead(TestInfo testInfo) throws Throwable {
        run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
            @Override
            public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
                testReadPendingIsResetAfterEachRead(serverBootstrap, bootstrap);
            }
        });
    }

    public void testReadPendingIsResetAfterEachRead(ServerBootstrap sb, Bootstrap cb) throws Throwable {
        Channel serverChannel = null;
        Channel clientChannel = null;
        try {
            ReadPendingInitializer serverInitializer = new ReadPendingInitializer();
            ReadPendingInitializer clientInitializer = new ReadPendingInitializer();
            sb.option(ChannelOption.SO_BACKLOG, 1024)
              .option(ChannelOption.AUTO_READ, true)
              .childOption(ChannelOption.AUTO_READ, false)
              // We intend to do 2 reads per read loop wakeup
              .childOption(ChannelOption.RECVBUF_ALLOCATOR, new TestNumReadsRecvByteBufAllocator(2))
              .childHandler(serverInitializer);

            serverChannel = sb.bind().syncUninterruptibly().channel();

            cb.option(ChannelOption.AUTO_READ, false)
              // We intend to do 2 reads per read loop wakeup
              .option(ChannelOption.RECVBUF_ALLOCATOR, new TestNumReadsRecvByteBufAllocator(2))
              .handler(clientInitializer);
            clientChannel = cb.connect(serverChannel.localAddress()).syncUninterruptibly().channel();

            // 4 bytes means 2 read loops for TestNumReadsRecvByteBufAllocator
            clientChannel.writeAndFlush(randomBufferType(clientChannel.alloc(), new byte[4], 0, 4));

            // 4 bytes means 2 read loops for TestNumReadsRecvByteBufAllocator
            assertTrue(serverInitializer.channelInitLatch.await(5, TimeUnit.SECONDS));
            serverInitializer.channel.writeAndFlush(
                    randomBufferType(serverInitializer.channel.alloc(), new byte[4], 0 , 4));

            serverInitializer.channel.read();
            serverInitializer.readPendingHandler.assertAllRead();

            clientChannel.read();
            clientInitializer.readPendingHandler.assertAllRead();
        } finally {
            if (serverChannel != null) {
                serverChannel.close().syncUninterruptibly();
            }
            if (clientChannel != null) {
                clientChannel.close().syncUninterruptibly();
            }
        }
    }

    private static class ReadPendingInitializer extends ChannelInitializer<Channel> {
        final ReadPendingReadHandler readPendingHandler = new ReadPendingReadHandler();
        final CountDownLatch channelInitLatch = new CountDownLatch(1);
        volatile Channel channel;

        @Override
        protected void initChannel(Channel ch) throws Exception {
            channel = ch;
            ch.pipeline().addLast(readPendingHandler);
            channelInitLatch.countDown();
        }
    }

    private static final class ReadPendingReadHandler extends ChannelInboundHandlerAdapter {
        private final AtomicInteger count = new AtomicInteger();
        private final CountDownLatch latch = new CountDownLatch(1);
        private final CountDownLatch latch2 = new CountDownLatch(2);

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ReferenceCountUtil.release(msg);
            if (count.incrementAndGet() == 1) {
                // Call read the first time, to ensure it is not reset the second time.
                ctx.read();
            }