Repository: netty/netty
Issue: Feature request: Optimized versions of LengthFieldBasedFrameDecoder and FixedLengthFrameDecoder

## 


Issue comments:
@maseev I think you should try latest 4.1 branch and see how it behaves as we now also pool Slices if you use the PooledByteBufAllocator here.

@maseev maybe you want to send a PR which implements an "optimised" `LengthFieldBasedFrameDecoder` and `FixedLengthFrameDecoder` ?

I don't think your handler can be Sharable with that buffer bytebuf. 


Pull request review comments:
Maybe the check can be moved to some constructor?

No need to assign writeIndex here I guess?

Maybe I am missing something, but I assume a decoder like this one may throw every frame decoded to out via buffer.retainSlice(), etc. I could not quite follow the point here, even after reading the tests. Correct me if I am wrong. :-)

nit: mismatch of the if condition and the exception message?

I don't think so. Either way, you have to call an appropriate method which depends on an offset value. The same approach is used in LengthFieldBasedFrameDecoder. Seems legit for me.

Thanks! Will fix that very soon.

Do you mean `buffer.retain().slice()`? Well, it's not how it supposed to be working. The whole point of this decoder is to batch all incoming ByteBuf messages into one preallocated ByteBuf object. When the `channelReadComplete` method gets called we simply copy data from the buffer to another chunk buffer which would contain precisely N messages (where N >= 1) and pass it to next pipeline element. By doing so we're able to reuse this buffer, again and again, no matter how many messages we get. If I chose retain-slice approach I would have to keep an eye on the buffer reference counter because slice creates only a view of buffer sub-region, so you wouldn't be able to reuse it, if for example, the channel wasn't ready for write (channel.isWritable() returns false).

See my comment below.


Patch that resolved the issue:
diff --git a/codec/src/main/java/io/netty/handler/codec/BulkFixedLengthFrameDecoder.java b/codec/src/main/java/io/netty/handler/codec/BulkFixedLengthFrameDecoder.java
new file mode 100644
index 00000000000..4c3cfb8866d
--- /dev/null
+++ b/codec/src/main/java/io/netty/handler/codec/BulkFixedLengthFrameDecoder.java
@@ -0,0 +1,73 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.netty.handler.codec;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+
+import java.util.List;
+
+/**
+ * A decoder that mostly mimics {@link FixedLengthFrameDecoder} semantics. The only difference is this decoder batches
+ * incoming {@link ByteBuf} messages and then decodes them at once, passing a big {@link ByteBuf} which can possibly
+ * contain N messages (where N >= 1) through the pipeline at once.
+ */
+public class BulkFixedLengthFrameDecoder extends BulkFrameDecoder {
+
+    final int frameLength;
+
+    /**
+     * Creates a new instance.
+     *
+     * @param frameLength the length of the frame
+     */
+    public BulkFixedLengthFrameDecoder(int frameLength) {
+        this(0, frameLength);
+    }
+
+    /**
+     * Creates a new instance.
+     *
+     * @param startBufferSize the start buffer size which is used for batching incoming {@link ByteBuf} messages. Size
+     * of this buffer must be greater or equal to zero. If the size of this buffer is zero then the default size will be
+     * used (= 256 bytes).
+     * @param frameLength the length of the frame
+     */
+    public BulkFixedLengthFrameDecoder(int startBufferSize, int frameLength) {
+        super(startBufferSize);
+
+        if (frameLength <= 0) {
+            throw new IllegalArgumentException("frameLength must be a positive integer: " + frameLength);
+        }
+
+        this.frameLength = frameLength;
+    }
+
+    @Override
+    protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) {
+        final int readableBytes = buffer.readableBytes();
+
+        if (readableBytes < frameLength) {
+            return;
+        }
+
+        final int chunkLength = buffer.readableBytes() / frameLength * frameLength;
+        final ByteBuf chunk = ctx.alloc().buffer(chunkLength, chunkLength).writeBytes(buffer, chunkLength);
+
+        out.add(chunk);
+    }
+}
diff --git a/codec/src/main/java/io/netty/handler/codec/BulkFrameDecoder.java b/codec/src/main/java/io/netty/handler/codec/BulkFrameDecoder.java
new file mode 100644
index 00000000000..cfc65229d8a
--- /dev/null
+++ b/codec/src/main/java/io/netty/handler/codec/BulkFrameDecoder.java
@@ -0,0 +1,114 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.netty.handler.codec;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelInboundHandlerAdapter;
+
+import java.util.List;
+
+public abstract class BulkFrameDecoder extends ChannelInboundHandlerAdapter {
+
+    private final int startBufferSize;
+
+    private ByteBuf buffer;
+
+    /**
+     * @param startBufferSize the start buffer size which is used for batching incoming {@link ByteBuf} messages. Size
+     * of this buffer must be greater or equal to zero. If the size of this buffer is zero then the default size will be
+     * used (= 256 bytes).
+     */
+    protected BulkFrameDecoder(int startBufferSize) {
+        if (startBufferSize < 0) {
+            throw new IllegalArgumentException("startBufferSize must be greater or equal to zero: " + startBufferSize);
+        }
+
+        this.startBufferSize = startBufferSize;
+
+        CodecUtil.ensureNotSharable(this);
+    }
+
+    protected abstract void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out);
+
+    @Override
+    public void channelActive(ChannelHandlerContext ctx) throws Exception {
+        buffer = internalBuffer(ctx, startBufferSize);
+
+        ctx.fireChannelActive();
+    }
+
+    @Override
+    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
+        releaseInternalBuffer();
+        ctx.fireChannelInactive();
+    }
+
+    @Override
+    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
+        releaseInternalBuffer();
+    }
+
+    @Override
+    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
+        if (msg instanceof ByteBuf) {
+            final ByteBuf frame = (ByteBuf) msg;
+
+            try {
+                buffer.writeBytes(frame);
+            } finally {
+                frame.release();
+            }
+        } else {
+            ctx.fireChannelRead(msg);
+        }
+    }
+
+    @Override
+    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+        CodecOutputList out = CodecOutputList.newInstance();
+
+        try {
+            decode(ctx, buffer, out);
+        } finally {
+            buffer.discardReadBytes();
+            fireChannelRead(ctx, out);
+            out.recycle();
+        }
+    }
+
+    private static void fireChannelRead(ChannelHandlerContext ctx, List<Object> msgs) {
+        for (int i = 0, size = msgs.size(); i < size; ++i) {
+            ctx.fireChannelRead(msgs.get(i));
+        }
+    }
+
+    private static ByteBuf internalBuffer(ChannelHandlerContext ctx, int bufferSize) {
+        if (bufferSize == 0) {
+            return ctx.alloc().buffer();
+        }
+
+        return ctx.alloc().buffer(bufferSize);
+    }
+
+    private void releaseInternalBuffer() {
+        if (buffer != null) {
+            buffer.release();
+            buffer = null;
+        }
+    }
+}
diff --git a/codec/src/main/java/io/netty/handler/codec/BulkLengthFieldBasedFrameDecoder.java b/codec/src/main/java/io/netty/handler/codec/BulkLengthFieldBasedFrameDecoder.java
new file mode 100644
index 00000000000..5ec47fa65f4
--- /dev/null
+++ b/codec/src/main/java/io/netty/handler/codec/BulkLengthFieldBasedFrameDecoder.java
@@ -0,0 +1,160 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.handler.codec;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+
+import java.util.List;
+
+/**
+ * A decoder that mostly mimics {@link LengthFieldBasedFrameDecoder} semantics. The only difference is this decoder
+ * batches incoming {@link ByteBuf} messages and then decodes them at once, passing a big {@link ByteBuf} which can
+ * possibly contain N messages (where N >= 1) through the pipeline at once.
+ */
+public class BulkLengthFieldBasedFrameDecoder extends BulkFrameDecoder {
+
+    private final int maxFrameLength;
+    private final int lengthFieldOffset;
+    private final int lengthFieldLength;
+    private final int lengthFieldEndOffset;
+
+    /**
+     * Creates a new instance.
+     *
+     * @param maxFrameLength the maximum length of the frame.  If the length of the frame is greater than this value,
+     * {@link TooLongFrameException} will be thrown.
+     * @param lengthFieldOffset the offset of the length field
+     * @param lengthFieldLength the length of the length field
+     */
+    public BulkLengthFieldBasedFrameDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength) {
+        this(0, maxFrameLength, lengthFieldOffset, lengthFieldLength);
+    }
+
+    /**
+     * Creates a new instance.
+     *
+     * @param startBufferSize the start buffer size which is used for batching incoming {@link ByteBuf} messages. Size
+     * of this buffer must be greater or equal to zero. If the size of this buffer is zero then the default size will be
+     * used (= 256 bytes).
+     * @param maxFrameLength the maximum length of the frame.  If the length of the frame is greater than this value,
+     * {@link TooLongFrameException} will be thrown.
+     * @param lengthFieldOffset the offset of the length field
+     * @param lengthFieldLength the length of the length field
+     */
+    public BulkLengthFieldBasedFrameDecoder(int startBufferSize, int maxFrameLength, int lengthFieldOffset,
+                                            int lengthFieldLength) {
+        super(startBufferSize);
+
+        if (maxFrameLength <= 0) {
+            throw new IllegalArgumentException(
+                    "maxFrameLength must be a positive integer: " +
+                    maxFrameLength);
+        }
+
+        if (lengthFieldOffset < 0) {
+            throw new IllegalArgumentException(
+                    "lengthFieldOffset must be a non-negative integer: " +
+                    lengthFieldOffset);
+        }
+
+        if (lengthFieldOffset > maxFrameLength - lengthFieldLength) {
+            throw new IllegalArgumentException(
+                    "maxFrameLength (" + maxFrameLength + ") " +
+                    "must be equal to or greater than " +
+                    "lengthFieldOffset (" + lengthFieldOffset + ") + " +
+                    "lengthFieldLength (" + lengthFieldLength + ").");
+        }
+
+        this.maxFrameLength = maxFrameLength;
+        this.lengthFieldOffset = lengthFieldOffset;
+        this.lengthFieldLength = lengthFieldLength;
+        lengthFieldEndOffset = lengthFieldOffset + lengthFieldLength;
+    }
+
+    @Override
+    protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) {
+        int readableBytes = buffer.readableBytes();
+        int rdIdx = buffer.readerIndex();
+
+        while (readableBytes >= lengthFieldEndOffset) {
+            int actualLengthFieldOffset = rdIdx + lengthFieldOffset;
+            long frameLength = getUnadjustedFrameLength(buffer, actualLengthFieldOffset, lengthFieldLength);
+
+            if (frameLength < 0) {
+                throw new CorruptedFrameException(
+                        "negative pre-adjustment length field: " + frameLength);
+            }
+
+            frameLength += lengthFieldEndOffset;
+
+            if (frameLength > maxFrameLength) {
+                throw new TooLongFrameException(
+                        "Adjusted frame length exceeds " + maxFrameLength +
+                        ": " + frameLength + " - discarded");
+            }
+
+            int frameLengthInt = (int) frameLength;
+
+            if (readableBytes < frameLengthInt) {
+                break;
+            }
+
+            readableBytes -= frameLengthInt;
+            rdIdx += frameLengthInt;
+        }
+
+        if (rdIdx == 0) {
+            return;
+        }
+
+        final int previousWriterIndex = buffer.writerIndex();
+
+        buffer.writerIndex(rdIdx);
+
+        final ByteBuf chunk = ctx.alloc().buffer(rdIdx, rdIdx).writeBytes(buffer, rdIdx);
+
+        buffer.writerIndex(previousWriterIndex);
+        buffer.discardReadBytes();
+
+        out.add(chunk);
+    }
+
+    protected long getUnadjustedFrameLength(ByteBuf buf, int offset, int length) {
+        long frameLength;
+        switch (length) {
+        case 1:
+            frameLength = buf.getUnsignedByte(offset);
+            break;
+        case 2:
+            frameLength = buf.getUnsignedShort(offset);
+            break;
+        case 3:
+            frameLength = buf.getUnsignedMedium(offset);
+            break;
+        case 4:
+            frameLength = buf.getUnsignedInt(offset);
+            break;
+        case 8:
+            frameLength = buf.getLong(offset);
+            break;
+        default:
+            throw new DecoderException(
+                    "unsupported lengthFieldLength: " + lengthFieldLength + " (expected: 1, 2, 3, 4, or 8)");
+        }
+        return frameLength;
+    }
+}


Test patch:
diff --git a/codec/src/test/java/io/netty/handler/codec/BulkFixedLengthFrameDecoderTest.java b/codec/src/test/java/io/netty/handler/codec/BulkFixedLengthFrameDecoderTest.java
new file mode 100644
index 00000000000..d6ba85f2066
--- /dev/null
+++ b/codec/src/test/java/io/netty/handler/codec/BulkFixedLengthFrameDecoderTest.java
@@ -0,0 +1,108 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package io.netty.handler.codec;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.embedded.EmbeddedChannel;
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.*;
+import static org.junit.Assert.*;
+
+public class BulkFixedLengthFrameDecoderTest {
+
+    @Test(expected = IllegalArgumentException.class)
+    public void decoderWithZeroFrameLengthShouldThrowException() {
+        new BulkFixedLengthFrameDecoder(0);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void decoderWithNegativeFrameLengthShouldThrowException() {
+        new BulkFixedLengthFrameDecoder(-1);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void decoderWithNegativeStartBufferSizeShouldThrowException() {
+        new BulkFixedLengthFrameDecoder(-1, 1);
+    }
+
+    @Test
+    public void writingNotAByteBufShouldPassThePipeline() {
+        final int frameLength = 10;
+        final EmbeddedChannel channel = new EmbeddedChannel(new BulkFixedLengthFrameDecoder(frameLength));
+        final Object marker = new Object();
+
+        channel.writeInbound(marker);
+        final Object receivedObject = channel.readInbound();
+
+        assertThat(receivedObject, is(equalTo(marker)));
+
+        channel.finish();
+    }
+
+    @Test
+    public void writingNotEnoughBytesReadsNothing() {
+        final int frameLength = 10;
+        final EmbeddedChannel channel = new EmbeddedChannel(new BulkFixedLengthFrameDecoder(frameLength));
+        final ByteBuf chunk = Unpooled.copiedBuffer(new byte[5]);
+
+        channel.writeInbound(chunk);
+        final Object receivedObject = channel.readInbound();
+        assertThat(receivedObject, is(equalTo(null)));
+
+        channel.finish();
+    }
+
+    @Test
+    public void writingOneAndAHalfMessageShouldRetrieveOnlyOneMessage() {
+        final int frameLength = 10;
+        final int startBufferSize = 5;
+        final int oneAndAHalfMessageLength = frameLength + frameLength / 2;
+        final EmbeddedChannel channel =
+                new EmbeddedChannel(new BulkFixedLengthFrameDecoder(startBufferSize, frameLength));
+        final ByteBuf chunk = Unpooled.copiedBuffer(new byte[oneAndAHalfMessageLength]);
+
+        channel.writeInbound(chunk);
+        final ByteBuf receivedChunk = channel.readInbound();
+        assertThat(receivedChunk.readableBytes(), is(equalTo(frameLength)));
+        receivedChunk.release();
+
+        channel.finish();
+    }
+
+    @Test
+    public void leftBytesShouldAccumulate() {
+        final int frameLength = 10;
+        final int startBufferSize = 5;
+        final EmbeddedChannel channel =
+                new EmbeddedChannel(new BulkFixedLengthFrameDecoder(startBufferSize, frameLength));
+
+        channel.writeInbound(Unpooled.copiedBuffer(new byte[15]));
+        ByteBuf receivedChunk = channel.readInbound();
+        assertThat(receivedChunk.readableBytes(), is(equalTo(frameLength)));
+        receivedChunk.release();
+
+        channel.writeInbound(Unpooled.copiedBuffer(new byte[10]));
+        receivedChunk = channel.readInbound();
+        assertThat(receivedChunk.readableBytes(), is(equalTo(frameLength)));
+        receivedChunk.release();
+
+        receivedChunk = channel.readInbound();
+        assertThat(receivedChunk, is(equalTo(null)));
+        channel.finish();
+    }
+}
diff --git a/codec/src/test/java/io/netty/handler/codec/BulkLengthFieldBasedFrameDecoderTest.java b/codec/src/test/java/io/netty/handler/codec/BulkLengthFieldBasedFrameDecoderTest.java
new file mode 100644
index 00000000000..16b231e98e5
--- /dev/null
+++ b/codec/src/test/java/io/netty/handler/codec/BulkLengthFieldBasedFrameDecoderTest.java
@@ -0,0 +1,112 @@
+/*
+ * Copyright 2016 The Netty Project
+ *
+ * The Netty Project licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package io.netty.handler.codec;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.embedded.EmbeddedChannel;
+import org.junit.Test;
+
+import static org.hamcrest.CoreMatchers.*;
+import static org.junit.Assert.*;
+
+public class BulkLengthFieldBasedFrameDecoderTest {
+
+    @Test(expected = IllegalArgumentException.class)
+    public void decoderWithZeroMaxFrameLengthShouldThrowException() {
+        new BulkLengthFieldBasedFrameDecoder(0, 1, 1);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void decoderWithNegativeMaxFrameLengthShouldThrowException() {
+        new BulkLengthFieldBasedFrameDecoder(-1, 1, 1);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void decoderWithNegativeLengthFieldOffsetShouldThrowException() {
+        new BulkLengthFieldBasedFrameDecoder(1, -1, 1);
+    }
+
+    @Test(expected = IllegalArgumentException.class)
+    public void decoderWithTooBigLengthFieldOffsetShouldThrowException() {
+        new BulkLengthFieldBasedFrameDecoder(10, 8, 8);
+    }
+
+    @Test
+    public void writingNotAByteBufShouldPassThePipeline() {
+        final EmbeddedChannel channel = new EmbeddedChannel(new BulkLengthFieldBasedFrameDecoder(2, 0, 1));
+        final Object marker = new Object();
+
+        channel.writeInbound(marker);
+        final Object receivedObject = channel.readInbound();
+
+        assertThat(receivedObject, is(equalTo(marker)));
+
+        channel.finish();
+    }
+
+    @Test
+    public void writingNotEnoughBytesReadsNothing() {
+        final EmbeddedChannel channel = new EmbeddedChannel(new BulkLengthFieldBasedFrameDecoder(6, 2, 1));
+        final ByteBuf chunk = Unpooled.copiedBuffer(new byte[] { 1, 1, 3, 0, 0 });
+
+        channel.writeInbound(chunk);
+        final Object receivedObject = channel.readInbound();
+        assertThat(receivedObject, is(equalTo(null)));
+
+        channel.finish();
+    }
+
+    @Test
+    public void writingOneAndAHalfMessageShouldRetrieveOnlyOneMessage() {
+        final EmbeddedChannel channel =
+                new EmbeddedChannel(new BulkLengthFieldBasedFrameDecoder(10, 2, 1));
+        final byte[] completeFrame = { 1, 1, 3, 0, 0, 0 };
+        final byte[] halfFrame = { 1, 1, 2, 0 };
+        final ByteBuf chunk = Unpooled.copiedBuffer(completeFrame, halfFrame);
+
+        channel.writeInbound(chunk);
+        final ByteBuf receivedChunk = channel.readInbound();
+        assertThat(receivedChunk.readableBytes(), is(equalTo(completeFrame.length)));
+        receivedChunk.release();
+
+        channel.finish();
+    }
+
+    @Test
+    public void leftBytesShouldAccumulate() {
+        final EmbeddedChannel channel =
+                new EmbeddedChannel(new BulkLengthFieldBasedFrameDecoder(10, 2, 1));
+        final byte[] completeFrame = { 1, 1, 3, 0, 0, 0 };
+        final byte[] halfFrame = { 1, 1, 4, 0 };
+        final byte[] anotherHalfFrame = { 0, 0, 0 };
+        ByteBuf chunk = Unpooled.copiedBuffer(completeFrame, halfFrame);
+
+        channel.writeInbound(chunk);
+        ByteBuf receivedChunk = channel.readInbound();
+        assertThat(receivedChunk.readableBytes(), is(equalTo(completeFrame.length)));
+        receivedChunk.release();
+
+        chunk = Unpooled.copiedBuffer(anotherHalfFrame);
+        channel.writeInbound(chunk);
+        receivedChunk = channel.readInbound();
+        assertThat(receivedChunk.readableBytes(), is(equalTo(halfFrame.length + anotherHalfFrame.length)));
+        receivedChunk.release();
+
+        channel.finish();
+    }
+}