Repository: netty/netty
Issue: MessageAggregator doesn't manage read flow (auto-read is disabled)
Labels: defect

### Expected behavior

This was discovered in Finagle's [HTTP proxy support handler](https://github.com/twitter/finagle/blob/develop/finagle-netty4/src/main/scala/com/twitter/finagle/netty4/proxy/HttpProxyConnectHandler.scala#L103-L106).

Calling `ctx.read()` (when auto-read is disabled) in any handler placed right after the [`MessageAggregator`](https://github.com/netty/netty/blob/4.1/codec/src/main/java/io/netty/handler/codec/MessageAggregator.java) (or `HttpObjectAggregator`) should guarantee at least one inbound message to be fired in a feasible future.

### Actual behavior
`MessageAggregator` doesn't manage the read-flow when auto-read is disabled so the connection just stalls if there are several things that have to be aggregated. I searched for `read()` in `MessageAggregator` and didn't find anything.

I think we should keep issuing read-requests from `MessageAggregator` if (1) a previous handler issued a read-request and (2) message aggregation is still in progress. Please, let me know what do you think about that. As usual, I'm happy to prepare a PR.

### Environment:

Netty 4.1.9-Final


Issue comments:
@vkostyukov yes this is a bug. Can you please submit a pr ?
@vkostyukov maybe let me also think about if we can fix this in a more generic way so it "just works" for all handlers.
@vkostyukov so yes just do a PR for now. I may be able to come up with a more general solution but not sure yet and it may take some time. 
Thanks @normanmaurer! I think fixing it in `MessageAggregator` will be generic enough since it will also fix `HttpObjectMessageAggretator`, `WebSocketFrameAggregator`, `AbstractMemcacheObjectAggregator`, `RedisBulkStringAggregator`, etc.

Working on a fix and a test.
So I was playing with the implementation today and it turns out it's not that easy to do.

Because `MesasgeAggregator` is an inbound handler, there is no way to intercept a read request from one of the next handlers, which I planned to use as a predicate whether or not start managing the read-flow (think about it this way: there is no need to keep reading if nobody asked us to). We could probably just do it (no matter if the read request was issued or not) and it will be more-or-less fine. I'd personally prefer this "implicit" auto-read=true (which we could document) for the pipelines with installed message aggregator rather than stalled connections, but I can totally see why people may not like this magical behavior change.

@normanmaurer, @Scottmitch What do you think about that?
Another option would be just changing `MessageAggregator` to be a duplex channel handler but that's a public API change.
@vkostyukov - Can we use a similar strategy as [ByteToMessageDecoder](https://github.com/netty/netty/blob/4.1/codec/src/main/java/io/netty/handler/codec/ByteToMessageDecoder.java#L304)? `MessageAggregator` would essentially track if anything was added to `out` in `channelRead`, and check/reset that state in `channelReadComplete`.
Oh I like it @Scottmitch! Thans a lot! Going to try this out.
Fixed by https://github.com/netty/netty/pull/6604 .. @vkostyukov thanks!

Pull request review comments:
please fix formatting 
use `Unpooled.copiedBuffer(..., CharsetUtil.US_ASCII);`
fix formatting
use `embedded.readInbound()` and so train the read messages. Also assert its what you expect and ensure you release everything
private final
nit: `Unpooled.copiedBuffer(string, CharsetUtil.US_ASCII);`

No need to first convert to byte[] and then do the copy
Sure! Just updated the PR.

Patch that resolved the issue:
diff --git a/codec/src/main/java/io/netty/handler/codec/MessageAggregator.java b/codec/src/main/java/io/netty/handler/codec/MessageAggregator.java
index 38e5bd7575a..2bede688bd0 100644
--- a/codec/src/main/java/io/netty/handler/codec/MessageAggregator.java
+++ b/codec/src/main/java/io/netty/handler/codec/MessageAggregator.java
@@ -399,6 +399,17 @@ protected void handleOversizedMessage(ChannelHandlerContext ctx, S oversized) th
                 new TooLongFrameException("content length exceeded " + maxContentLength() + " bytes."));
     }
 
+    @Override
+    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
+        // We might need keep reading the channel until the full message is aggregated.
+        //
+        // See https://github.com/netty/netty/issues/6583
+        if (currentMessage != null && !ctx.channel().config().isAutoRead()) {
+            ctx.read();
+        }
+        ctx.fireChannelReadComplete();
+    }
+
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
         try {


Test patch:
diff --git a/codec/src/test/java/io/netty/handler/codec/MessageAggregatorTest.java b/codec/src/test/java/io/netty/handler/codec/MessageAggregatorTest.java
new file mode 100644
index 00000000000..d3c4f5f6c97
--- /dev/null
+++ b/codec/src/test/java/io/netty/handler/codec/MessageAggregatorTest.java
@@ -0,0 +1,94 @@
+/*
+ * Copyright 2017 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 org.junit.Test;
+
+import static org.junit.Assert.*;
+import static org.mockito.Mockito.*;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufHolder;
+import io.netty.buffer.DefaultByteBufHolder;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.embedded.EmbeddedChannel;
+import io.netty.util.CharsetUtil;
+
+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));
+    }
+
+    @SuppressWarnings("unchecked")
+    @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());
+    }
+}