SocketBufReleaseTest Class — netty Architecture

Source Code

testsuite/src/main/java/io/netty/testsuite/transport/socket/SocketBufReleaseTest.java lines 37–114

public class SocketBufReleaseTest extends AbstractSocketTest {

    private static final EventExecutor executor =
            new DefaultEventExecutorGroup(1, new DefaultThreadFactory(SocketBufReleaseTest.class, true)).next();

    @Test
    public void testBufRelease(TestInfo testInfo) throws Throwable {
        run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
            @Override
            public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
                testBufRelease(serverBootstrap, bootstrap);
            }
        });
    }

    public void testBufRelease(ServerBootstrap sb, Bootstrap cb) throws Throwable {
        BufWriterHandler serverHandler = new BufWriterHandler();
        BufWriterHandler clientHandler = new BufWriterHandler();

        sb.childHandler(serverHandler);
        cb.handler(clientHandler);

        Channel sc = sb.bind().sync().channel();
        Channel cc = cb.connect(sc.localAddress()).sync().channel();

        // Ensure the server socket accepted the client connection *and* initialized pipeline successfully.
        serverHandler.channelFuture.sync();

        // and then close all sockets.
        sc.close().sync();
        cc.close().sync();

        serverHandler.check();
        clientHandler.check();

        serverHandler.release();
        clientHandler.release();
    }

    private static class BufWriterHandler extends SimpleChannelInboundHandler<Object> {

        private final Random random = new Random();
        private final CountDownLatch latch = new CountDownLatch(1);
        private ByteBuf buf;
        private final Promise<Channel> channelFuture = new DefaultPromise<Channel>(executor);

        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
            channelFuture.setSuccess(ctx.channel());
        }

        @Override
        public void channelActive(final ChannelHandlerContext ctx) throws Exception {
            byte[] data = new byte[1024];
            random.nextBytes(data);

            buf = ctx.alloc().buffer();
            // call retain on it so it can't be put back on the pool
            buf.writeBytes(data).retain();

            ctx.channel().writeAndFlush(buf).addListener(future -> latch.countDown());
        }

        @Override
        public void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
            // discard
        }

        public void check() throws InterruptedException {
            latch.await();
            assertEquals(1, buf.refCnt());
        }

        void release() {
            buf.release();
        }
    }
}