AbstractCoalescingBufferQueueTest Class — netty Architecture

Source Code

transport/src/test/java/io/netty/channel/AbstractCoalescingBufferQueueTest.java lines 35–154

public class AbstractCoalescingBufferQueueTest {

    // See https://github.com/netty/netty/issues/10286
    @Test
    public void testDecrementAllWhenWriteAndRemoveAll() {
        testDecrementAll(true);
    }

    // See https://github.com/netty/netty/issues/10286
    @Test
    public void testDecrementAllWhenReleaseAndFailAll() {
        testDecrementAll(false);
    }

    private static void testDecrementAll(boolean write) {
        EmbeddedChannel channel = new EmbeddedChannel(new ChannelOutboundHandlerAdapter() {
            @Override
            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
                ReferenceCountUtil.release(msg);
                promise.setSuccess();
            }
        }, new ChannelHandlerAdapter() { });
        final AbstractCoalescingBufferQueue queue = new AbstractCoalescingBufferQueue(channel, 128) {
            @Override
            protected ByteBuf compose(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf next) {
                return composeIntoComposite(alloc, cumulation, next);
            }

            @Override
            protected ByteBuf removeEmptyValue() {
                return Unpooled.EMPTY_BUFFER;
            }
        };

        final byte[] bytes = new byte[128];
        queue.add(Unpooled.wrappedBuffer(bytes), future -> {
            queue.add(Unpooled.wrappedBuffer(bytes));
            assertEquals(bytes.length, queue.readableBytes());
        });

        assertEquals(bytes.length, queue.readableBytes());

        ChannelHandlerContext ctx = channel.pipeline().lastContext();
        if (write) {
            queue.writeAndRemoveAll(ctx);
        } else {
            queue.releaseAndFailAll(ctx, new ClosedChannelException());
        }
        ByteBuf buffer = queue.remove(channel.alloc(), 128, channel.newPromise());
        assertFalse(buffer.isReadable());
        buffer.release();

        assertTrue(queue.isEmpty());
        assertEquals(0, queue.readableBytes());

        assertFalse(channel.finish());
    }

    @Test
    public void testKeepStateConsistentOnError() {
        final IllegalReferenceCountException exception = new IllegalReferenceCountException();
        final EmbeddedChannel channel = new EmbeddedChannel();
        final AbstractCoalescingBufferQueue queue = new AbstractCoalescingBufferQueue(channel, 128) {
            @Override
            protected ByteBuf compose(ByteBufAllocator alloc, ByteBuf cumulation, ByteBuf next) {
                // Simulate throwing an IllegalReferenceCountException.
                throw exception;
            }

            @Override
            protected ByteBuf composeFirst(ByteBufAllocator allocator, ByteBuf first, int bufferSize) {
                // Simulate throwing an IllegalReferenceCountException.
                throw exception;
            }

            @Override
            protected ByteBuf removeEmptyValue() {
                return Unpooled.EMPTY_BUFFER;
            }
        };