CompositeBufferGatheringWriteTest Class — netty Architecture

Source Code

testsuite/src/main/java/io/netty/testsuite/transport/socket/CompositeBufferGatheringWriteTest.java lines 43–309

public class CompositeBufferGatheringWriteTest extends AbstractSocketTest {
    private static final int EXPECTED_BYTES = 20;

    @Test
    @Timeout(value = 10000, unit = TimeUnit.MILLISECONDS)
    public void testSingleCompositeBufferWrite(TestInfo testInfo) throws Throwable {
        run(testInfo, new Runner<ServerBootstrap, Bootstrap>() {
            @Override
            public void run(ServerBootstrap serverBootstrap, Bootstrap bootstrap) throws Throwable {
                testSingleCompositeBufferWrite(serverBootstrap, bootstrap);
            }
        });
    }

    public void testSingleCompositeBufferWrite(ServerBootstrap sb, Bootstrap cb) throws Throwable {
        Channel serverChannel = null;
        Channel clientChannel = null;
        try {
            final CountDownLatch latch = new CountDownLatch(1);
            final AtomicReference<Object> clientReceived = new AtomicReference<Object>();
            sb.childHandler(new ChannelInitializer<Channel>() {
                @Override
                protected void initChannel(Channel ch) throws Exception {
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
                            ctx.writeAndFlush(newCompositeBuffer(ctx.alloc()))
                                    .addListener(ChannelFutureListener.CLOSE);
                        }
                    });
                }
            });
            cb.handler(new ChannelInitializer<Channel>() {
                @Override
                protected void initChannel(Channel ch) throws Exception {
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        private ByteBuf aggregator;
                        @Override
                        public void handlerAdded(ChannelHandlerContext ctx) {
                            aggregator = ctx.alloc().buffer(EXPECTED_BYTES);
                        }

                        @Override
                        public void channelRead(ChannelHandlerContext ctx, Object msg) {
                            try {
                                if (msg instanceof ByteBuf) {
                                    aggregator.writeBytes((ByteBuf) msg);
                                }
                            } finally {
                                ReferenceCountUtil.release(msg);
                            }
                        }

                        @Override
                        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                            // IOException is fine as it will also close the channel and may just be a connection reset.
                            if (!(cause instanceof IOException)) {
                                clientReceived.set(cause);
                                latch.countDown();
                            }
                        }

                        @Override
                        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                            if (clientReceived.compareAndSet(null, aggregator)) {
                                try {
                                    assertEquals(EXPECTED_BYTES, aggregator.readableBytes());
                                } catch (Throwable cause) {
                                    aggregator.release();
                                    aggregator = null;
                                    clientReceived.set(cause);
                                } finally {
                                    latch.countDown();
                                }
                            }
                        }
                    });
                }
            });

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