[#1880] Use ByteBufAllocator when read bytes into new chunks

buffer/src/main/java/io/netty/buffer/ByteBufUtil.java

@@ -257,6 +257,23 @@ public final class ByteBufUtil {
         return Long.reverseBytes(value);
     }
 
+    /**
+     * Read the given amount of bytes into a new {@link ByteBuf} that is allocated from the {@link ByteBufAllocator}.
+     */
+    public static ByteBuf readBytes(ByteBufAllocator alloc, ByteBuf buffer, int length) {
+        boolean release = true;
+        ByteBuf dst = alloc.buffer(length);
+        try {
+            buffer.readBytes(dst);
+            release = false;
+            return dst;
+        } finally {
+            if (release) {
+                dst.release();
+            }
+        }
+    }
+
     private static int firstIndexOf(ByteBuf buffer, int fromIndex, int toIndex, byte value) {
         fromIndex = Math.max(fromIndex, 0);
         if (fromIndex >= toIndex || buffer.capacity() == 0) {

codec-http/src/main/java/io/netty/handler/codec/http/HttpObjectDecoder.java

@@ -25,6 +25,8 @@ import io.netty.handler.codec.TooLongFrameException;
 
 import java.util.List;
 
+import static io.netty.buffer.ByteBufUtil.readBytes;
+
 /**
  * Decodes {@link ByteBuf}s into {@link HttpMessage}s and
  * {@link HttpContent}s.
@@ -265,7 +267,7 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<HttpObjectDecod
                 toRead = maxChunkSize;
             }
             out.add(message);
-            out.add(new DefaultHttpContent(buffer.readBytes(toRead)));
+            out.add(new DefaultHttpContent(readBytes(ctx.alloc(), buffer, toRead)));
             return;
         }
         case READ_VARIABLE_LENGTH_CONTENT_AS_CHUNKS: {
@@ -274,7 +276,7 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<HttpObjectDecod
             if (toRead > maxChunkSize) {
                 toRead = maxChunkSize;
             }
-            ByteBuf content = buffer.readBytes(toRead);
+            ByteBuf content = readBytes(ctx.alloc(), buffer, toRead);
             if (!buffer.isReadable()) {
                 reset();
                 out.add(new DefaultLastHttpContent(content));
@@ -284,7 +286,7 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<HttpObjectDecod
             return;
         }
         case READ_FIXED_LENGTH_CONTENT: {
-            readFixedLengthContent(buffer, out);
+            readFixedLengthContent(ctx, buffer, out);
             return;
         }
         case READ_FIXED_LENGTH_CONTENT_AS_CHUNKS: {
@@ -308,7 +310,7 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<HttpObjectDecod
             if (toRead > chunkSize) {
                 toRead = (int) chunkSize;
             }
-            ByteBuf content = buffer.readBytes(toRead);
+            ByteBuf content = readBytes(ctx.alloc(), buffer, toRead);
             if (chunkSize > toRead) {
                 chunkSize -= toRead;
             } else {
@@ -348,7 +350,7 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<HttpObjectDecod
         }
         case READ_CHUNKED_CONTENT: {
             assert chunkSize <= Integer.MAX_VALUE;
-            HttpContent chunk = new DefaultHttpContent(buffer.readBytes((int) chunkSize));
+            HttpContent chunk = new DefaultHttpContent(readBytes(ctx.alloc(), buffer, (int) chunkSize));
             checkpoint(State.READ_CHUNK_DELIMITER);
             out.add(chunk);
             return;
@@ -375,7 +377,7 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<HttpObjectDecod
             if (toRead > readLimit) {
                 toRead = readLimit;
             }
-            HttpContent chunk = new DefaultHttpContent(buffer.readBytes(toRead));
+            HttpContent chunk = new DefaultHttpContent(readBytes(ctx.alloc(), buffer, toRead));
             if (chunkSize > toRead) {
                 chunkSize -= toRead;
             } else {
@@ -553,7 +555,7 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<HttpObjectDecod
         }
     }
 
-    private void readFixedLengthContent(ByteBuf buffer, List<Object> out) {
+    private void readFixedLengthContent(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) {
         //we have a content-length so we just read the correct number of bytes
         long length = HttpHeaders.getContentLength(message, -1);
         assert length <= Integer.MAX_VALUE;
@@ -564,11 +566,11 @@ public abstract class HttpObjectDecoder extends ReplayingDecoder<HttpObjectDecod
         contentRead += toRead;
         if (length < contentRead) {
             out.add(message);
-            out.add(new DefaultHttpContent(buffer.readBytes(toRead)));
+            out.add(new DefaultHttpContent(readBytes(ctx.alloc(), buffer, toRead)));
             return;
         }
         if (content == null) {
-            content = buffer.readBytes((int) length);
+            content = readBytes(ctx.alloc(), buffer, (int) length);
         } else {
             content.writeBytes(buffer, (int) length);
         }

codec-http/src/test/java/io/netty/handler/codec/http/HttpContentCompressorTest.java

@@ -82,10 +82,12 @@ public class HttpContentCompressorTest {
         chunk = (HttpContent) ch.readOutbound();
         assertThat(chunk, is(instanceOf(HttpContent.class)));
         assertThat(chunk.content().isReadable(), is(true));
+        chunk.release();
 
         chunk = (HttpContent) ch.readOutbound();
         assertThat(chunk, is(instanceOf(LastHttpContent.class)));
         assertThat(chunk.content().isReadable(), is(false));
+        chunk.release();
 
         assertThat(ch.readOutbound(), is(nullValue()));
     }

codec-http/src/test/java/io/netty/handler/codec/http/HttpContentEncoderTest.java

@@ -67,6 +67,8 @@ public class HttpContentEncoderTest {
         chunk = (HttpContent) ch.readOutbound();
         assertThat(chunk.content().isReadable(), is(false));
         assertThat(chunk, is(instanceOf(LastHttpContent.class)));
+        chunk.release();
+
         assertThat(ch.readOutbound(), is(nullValue()));
     }
 
@@ -88,8 +90,12 @@ public class HttpContentEncoderTest {
         HttpContent chunk;
         chunk = (HttpContent) ch.readOutbound();
         assertThat(chunk.content().toString(CharsetUtil.US_ASCII), is("3"));
+        chunk.release();
+
         chunk = (HttpContent) ch.readOutbound();
         assertThat(chunk.content().toString(CharsetUtil.US_ASCII), is("2"));
+        chunk.release();
+
         chunk = (HttpContent) ch.readOutbound();
         assertThat(chunk.content().toString(CharsetUtil.US_ASCII), is("1"));
 
@@ -97,6 +103,7 @@ public class HttpContentEncoderTest {
         chunk = (HttpContent) ch.readOutbound();
         assertThat(chunk.content().isReadable(), is(false));
         assertThat(chunk, is(instanceOf(LastHttpContent.class)));
+        chunk.release();
 
         assertThat(ch.readOutbound(), is(nullValue()));
     }
@@ -121,16 +128,23 @@ public class HttpContentEncoderTest {
         HttpContent chunk;
         chunk = (HttpContent) ch.readOutbound();
         assertThat(chunk.content().toString(CharsetUtil.US_ASCII), is("3"));
+        chunk.release();
+
         chunk = (HttpContent) ch.readOutbound();
         assertThat(chunk.content().toString(CharsetUtil.US_ASCII), is("2"));
+        chunk.release();
+
         chunk = (HttpContent) ch.readOutbound();
         assertThat(chunk.content().toString(CharsetUtil.US_ASCII), is("1"));
-
         assertThat(chunk, is(instanceOf(HttpContent.class)));
+        chunk.release();
+
         chunk = (HttpContent) ch.readOutbound();
         assertThat(chunk.content().isReadable(), is(false));
         assertThat(chunk, is(instanceOf(LastHttpContent.class)));
         assertEquals("Netty", ((LastHttpContent) chunk).trailingHeaders().get("X-Test"));
+        chunk.release();
+
         assertThat(ch.readOutbound(), is(nullValue()));
     }
 
@@ -148,8 +162,12 @@ public class HttpContentEncoderTest {
         HttpContent c = (HttpContent) ch.readOutbound();
         assertThat(c.content().readableBytes(), is(2));
         assertThat(c.content().toString(CharsetUtil.US_ASCII), is("42"));
+        c.release();
+
         LastHttpContent last = (LastHttpContent) ch.readOutbound();
         assertThat(last.content().readableBytes(), is(0));
+        last.release();
+
         assertThat(ch.readOutbound(), is(nullValue()));
     }
 
@@ -169,10 +187,13 @@ public class HttpContentEncoderTest {
         HttpContent chunk = (HttpContent) ch.readOutbound();
         assertThat(chunk.content().toString(CharsetUtil.US_ASCII), is("0"));
         assertThat(chunk, is(instanceOf(HttpContent.class)));
+        chunk.release();
 
         chunk = (HttpContent) ch.readOutbound();
         assertThat(chunk.content().isReadable(), is(false));
         assertThat(chunk, is(instanceOf(LastHttpContent.class)));
+        chunk.release();
+
         assertThat(ch.readOutbound(), is(nullValue()));
     }
 
@@ -198,6 +219,7 @@ public class HttpContentEncoderTest {
         assertThat(res.headers().get(Names.CONTENT_ENCODING), is(nullValue()));
         assertThat(res.content().readableBytes(), is(0));
         assertThat(res.content().toString(CharsetUtil.US_ASCII), is(""));
+        res.release();
 
         assertThat(ch.readOutbound(), is(nullValue()));
     }

codec-http/src/test/java/io/netty/handler/codec/http/HttpObjectAggregatorTest.java

@@ -68,6 +68,7 @@ public class HttpObjectAggregatorTest {
             // This should be false as we decompose the buffer before to not have deep hierarchy
             assertFalse(buf instanceof CompositeByteBuf);
         }
+        aggregatedMessage.release();
     }
 
     @Test

codec-http/src/test/java/io/netty/handler/codec/http/HttpResponseDecoderTest.java

@@ -38,6 +38,7 @@ public class HttpResponseDecoderTest {
 
         LastHttpContent content = (LastHttpContent) ch.readInbound();
         assertThat(content.content().isReadable(), is(false));
+        content.release();
 
         assertThat(ch.readInbound(), is(nullValue()));
     }
@@ -55,11 +56,13 @@ public class HttpResponseDecoderTest {
         ch.writeInbound(Unpooled.wrappedBuffer(new byte[1024]));
         HttpContent content = (HttpContent) ch.readInbound();
         assertThat(content.content().readableBytes(), is(1024));
+        content.release();
 
         assertThat(ch.finish(), is(true));
 
         LastHttpContent lastContent = (LastHttpContent) ch.readInbound();
         assertThat(lastContent.content().isReadable(), is(false));
+        lastContent.release();
 
         assertThat(ch.readInbound(), is(nullValue()));
     }

codec-http/src/test/java/io/netty/handler/codec/http/HttpServerCodecTest.java

@@ -57,6 +57,7 @@ public class HttpServerCodecTest {
             empty = false;
             totalBytesPolled += httpChunk.content().readableBytes();
             Assert.assertFalse(httpChunk instanceof LastHttpContent);
+            httpChunk.release();
         }
         Assert.assertFalse(empty);
         Assert.assertEquals(offeredContentLength, totalBytesPolled);