MessageAggregatorTest Class — netty Architecture

Source Code

codec-base/src/test/java/io/netty/handler/codec/MessageAggregatorTest.java lines 37–137

public class MessageAggregatorTest {
    private static final class ReadCounter extends ChannelOutboundHandlerAdapter {
        int value;

        @Override
        public void read(ChannelHandlerContext ctx) throws Exception {
            value++;
            ctx.read();
        }
    }

    abstract static class MockMessageAggregator
        extends MessageAggregator<ByteBufHolder, ByteBufHolder, ByteBufHolder, ByteBufHolder> {

        protected MockMessageAggregator() {
            super(1024);
        }

        @Override
        protected ByteBufHolder beginAggregation(ByteBufHolder start, ByteBuf content) throws Exception {
            return start.replace(content);
        }
    }

    private static ByteBufHolder message(String string) {
        return new DefaultByteBufHolder(
            Unpooled.copiedBuffer(string, CharsetUtil.US_ASCII));
    }

    @Test
    public void testReadFlowManagement() throws Exception {
        ReadCounter counter = new ReadCounter();
        ByteBufHolder first = message("first");
        ByteBufHolder chunk = message("chunk");
        ByteBufHolder last = message("last");

        MockMessageAggregator agg = spy(MockMessageAggregator.class);
        when(agg.isStartMessage(first)).thenReturn(true);
        when(agg.isContentMessage(chunk)).thenReturn(true);
        when(agg.isContentMessage(last)).thenReturn(true);
        when(agg.isLastContentMessage(last)).thenReturn(true);

        EmbeddedChannel embedded = new EmbeddedChannel(counter, agg);
        embedded.config().setAutoRead(false);

        assertFalse(embedded.writeInbound(first));
        assertFalse(embedded.writeInbound(chunk));
        assertTrue(embedded.writeInbound(last));

        assertEquals(3, counter.value); // 2 reads issued from MockMessageAggregator
                                        // 1 read issued from EmbeddedChannel constructor

        ByteBufHolder all = new DefaultByteBufHolder(Unpooled.wrappedBuffer(
            first.content().retain(), chunk.content().retain(), last.content().retain()));
        ByteBufHolder out = embedded.readInbound();

        assertEquals(all, out);
        assertTrue(all.release() && out.release());
        assertFalse(embedded.finish());
    }

    @Test
    public void testCloseWhileAggregating() throws Exception {
        ReadCounter counter = new ReadCounter();
        ByteBufHolder first = new TestMessage(Unpooled.copiedBuffer("first", CharsetUtil.US_ASCII));

        MockMessageAggregator agg = spy(MockMessageAggregator.class);
        when(agg.isStartMessage(first)).thenReturn(true);
        when(agg.isLastContentMessage(first)).thenReturn(false);

        final EmbeddedChannel embedded = new EmbeddedChannel(counter, agg);
        embedded.config().setAutoRead(false);

        assertFalse(embedded.writeInbound(first));

        assertEquals(2, counter.value);
        assertThrows(PrematureChannelClosureException.class, new Executable() {
            @Override
            public void execute() {
                embedded.finish();
            }