Fix exception if Response has content (#11093)

Motivation:

If compression is enabled and the HttpResponse also
implements HttpContent (but not LastHttpContent) then
the buffer will be freed to eagerly.

Modification:

I retain the buffer the same way that is done for the LastHttpContent case.

Note that there is another suspicious looking call a few lines above (if beginEncode returns null). I am not sure if this should also be retained.

Result:

Fixes #11092

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

@@ -193,7 +193,7 @@ public abstract class HttpContentEncoder extends MessageToMessageCodec<HttpReque
                     res.headers().remove(HttpHeaderNames.CONTENT_LENGTH);
                     res.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
 
-                    out.add(res);
+                    out.add(ReferenceCountUtil.retain(res));
                     state = State.AWAIT_CONTENT;
                     if (!(msg instanceof HttpContent)) {
                         // only break out the switch statement if we have not content to process

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

@@ -17,6 +17,7 @@ package io.netty.handler.codec.http;
 
 import io.netty.bootstrap.Bootstrap;
 import io.netty.bootstrap.ServerBootstrap;
+import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.Channel;
@@ -48,7 +49,13 @@ import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.not;
 import static org.hamcrest.CoreMatchers.nullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 public class HttpContentCompressorTest {
 
@@ -168,6 +175,47 @@ public class HttpContentCompressorTest {
         assertThat(ch.readOutbound(), is(nullValue()));
     }
 
+    @Test
+    public void testChunkedContentWithAssembledResponse() throws Exception {
+        EmbeddedChannel ch = new EmbeddedChannel(new HttpContentCompressor());
+        ch.writeInbound(newRequest());
+
+        HttpResponse res = new AssembledHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,
+                Unpooled.copiedBuffer("Hell", CharsetUtil.US_ASCII));
+        res.headers().set(HttpHeaderNames.TRANSFER_ENCODING, HttpHeaderValues.CHUNKED);
+        ch.writeOutbound(res);
+
+        assertAssembledEncodedResponse(ch);
+
+        ch.writeOutbound(new DefaultHttpContent(Unpooled.copiedBuffer("o, w", CharsetUtil.US_ASCII)));
+        ch.writeOutbound(new DefaultLastHttpContent(Unpooled.copiedBuffer("orld", CharsetUtil.US_ASCII)));
+
+        HttpContent chunk;
+        chunk = ch.readOutbound();
+        assertThat(ByteBufUtil.hexDump(chunk.content()), is("1f8b0800000000000000f248cdc901000000ffff"));
+        chunk.release();
+
+        chunk = ch.readOutbound();
+        assertThat(ByteBufUtil.hexDump(chunk.content()), is("cad7512807000000ffff"));
+        chunk.release();
+
+        chunk = ch.readOutbound();
+        assertThat(ByteBufUtil.hexDump(chunk.content()), is("ca2fca4901000000ffff"));
+        chunk.release();
+
+        chunk = ch.readOutbound();
+        assertThat(ByteBufUtil.hexDump(chunk.content()), is("0300c2a99ae70c000000"));
+        assertThat(chunk, is(instanceOf(HttpContent.class)));
+        chunk.release();
+
+        chunk = ch.readOutbound();
+        assertThat(chunk.content().isReadable(), is(false));
+        assertThat(chunk, is(instanceOf(LastHttpContent.class)));
+        chunk.release();
+
+        assertThat(ch.readOutbound(), is(nullValue()));
+    }
+
     @Test
     public void testChunkedContentWithTrailingHeader() throws Exception {
         EmbeddedChannel ch = new EmbeddedChannel(new HttpContentCompressor());
@@ -668,4 +716,92 @@ public class HttpContentCompressorTest {
         assertThat(res.headers().get(HttpHeaderNames.CONTENT_LENGTH), is(nullValue()));
         assertThat(res.headers().get(HttpHeaderNames.CONTENT_ENCODING), is("gzip"));
     }
+    private static void assertAssembledEncodedResponse(EmbeddedChannel ch) {
+        Object o = ch.readOutbound();
+        assertThat(o, is(instanceOf(AssembledHttpResponse.class)));
+
+        AssembledHttpResponse res = (AssembledHttpResponse) o;
+        try {
+            assertThat(res, is(instanceOf(HttpContent.class)));
+            assertThat(res.headers().get(HttpHeaderNames.TRANSFER_ENCODING), is("chunked"));
+            assertThat(res.headers().get(HttpHeaderNames.CONTENT_LENGTH), is(nullValue()));
+            assertThat(res.headers().get(HttpHeaderNames.CONTENT_ENCODING), is("gzip"));
+        } finally {
+            res.release();
+        }
+    }
+
+    static class AssembledHttpResponse extends DefaultHttpResponse implements HttpContent {
+
+        private final ByteBuf content;
+
+        AssembledHttpResponse(HttpVersion version, HttpResponseStatus status, ByteBuf content) {
+            super(version, status);
+            this.content = content;
+        }
+
+        @Override
+        public HttpContent copy() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public HttpContent duplicate() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public HttpContent retainedDuplicate() {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public HttpContent replace(ByteBuf content) {
+            throw new UnsupportedOperationException();
+        }
+
+        @Override
+        public AssembledHttpResponse retain() {
+            content.retain();
+            return this;
+        }
+
+        @Override
+        public AssembledHttpResponse retain(int increment) {
+            content.retain(increment);
+            return this;
+        }
+
+        @Override
+        public ByteBuf content() {
+            return content;
+        }
+
+        @Override
+        public int refCnt() {
+            return content.refCnt();
+        }
+
+        @Override
+        public boolean release() {
+            return content.release();
+        }
+
+        @Override
+        public boolean release(int decrement) {
+            return content.release(decrement);
+        }
+
+        @Override
+        public AssembledHttpResponse touch() {
+            content.touch();
+            return this;
+        }
+
+        @Override
+        public AssembledHttpResponse touch(Object hint) {
+            content.touch(hint);
+            return this;
+        }
+    }
 }