Repository: netty/netty
Issue: HttpObjectAggregator w/ isStartMessage: Bug or Feature?

Let's say you want to be very selective when and if to aggregate HttpRequests and Responses. Say you want to only aggregate responses with a content-type of application/json. Naturally I went with this:

```java
public class MyHttpObjectAggregator extends HttpObjectAggregator {
  
  public MyHttpObjectAggregator(int maxContentLength) {
    super(maxContentLength);
  }

  @Override
  protected boolean isStartMessage(HttpObject msg) throws Exception {
    if (msg instanceof HttpResponse) {
      HttpResponse response = (HttpResponse) msg;
      HttpHeaders headers = response.headers();
      
      String contentType = headers.get(HttpHeaderNames.CONTENT_TYPE);
      if (AsciiString.contentEqualsIgnoreCase(contentType, HttpHeaderValues.APPLICATION_JSON)) {
        return true;
      }
    }
    
    return false;
  }
}
```


... but it will not work due to the following two conditions

1) isContentMessage returns true for any HttpContent

https://github.com/netty/netty/blob/00afb19d7a37de21b35ce4f6cb3fa7f74809f2ab/codec/src/main/java/io/netty/handler/codec/MessageAggregator.java#L99

2) Subsequently we enter decode(), isContentMessage returns again true but `currentMessage` is null and early exits. Leaving the aggregator in a bad state.

https://github.com/netty/netty/blob/00afb19d7a37de21b35ce4f6cb3fa7f74809f2ab/codec/src/main/java/io/netty/handler/codec/MessageAggregator.java#L261-L265

My work around is to introduce a toggle flag in my aggregator that looks like this:

```java
public class MyHttpObjectAggregator extends HttpObjectAggregator {
  
  private boolean aggregating;
  
  public MyHttpObjectAggregator(int maxContentLength) {
    super(maxContentLength);
  }

  @Override
  protected boolean isStartMessage(HttpObject msg) throws Exception {
    if (msg instanceof HttpResponse) {
      HttpResponse response = (HttpResponse) msg;
      HttpHeaders headers = response.headers();
      
      String contentType = headers.get(HttpHeaderNames.CONTENT_TYPE);
      if (AsciiString.contentEqualsIgnoreCase(contentType, HttpHeaderValues.APPLICATION_JSON)) {
        aggregating = true;
        return true;
        
      } else {
        aggregating = false;
        return false;
      }
    }
    
    return false;
  }

  @Override
  protected boolean isContentMessage(HttpObject msg) throws Exception {
    return aggregating && super.isContentMessage(msg);
  }
}
```

I'm wondering if this is intended behavior or a bug?

### Expected behavior

### Actual behavior

### Steps to reproduce

### Minimal yet complete reproducer code (or URL to code)

### Netty version

### JVM version (e.g. `java -version`)

### OS version (e.g. `uname -a`)


Issue comments:
@rkapsi hmmm good question. For me it sounds more like a bug. WDYT ?
@normanmaurer same, I'll create a PR

Pull request review comments:
nit: remove empty line
assert return value
release message 
assert return value 
ensure to release stuff 
call `channel.finish()` and assert return value
call `channel.finish()` and assert return value
assert return value
release message
assert return value

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 2cdb880c993..ca43f1d797a 100644
--- a/codec/src/main/java/io/netty/handler/codec/MessageAggregator.java
+++ b/codec/src/main/java/io/netty/handler/codec/MessageAggregator.java
@@ -61,6 +61,8 @@ public abstract class MessageAggregator<I, S, C extends ByteBufHolder, O extends
     private ChannelHandlerContext ctx;
     private ChannelFutureListener continueResponseWriteListener;
 
+    private boolean aggregating;
+
     /**
      * Creates a new instance.
      *
@@ -96,7 +98,20 @@ public boolean acceptInboundMessage(Object msg) throws Exception {
         @SuppressWarnings("unchecked")
         I in = (I) msg;
 
-        return (isContentMessage(in) || isStartMessage(in)) && !isAggregated(in);
+        if (isAggregated(in)) {
+            return false;
+        }
+
+        // NOTE: It's tempting to make this check only if aggregating is false. There are however
+        // side conditions in decode(...) in respect to large messages.
+        if (isStartMessage(in)) {
+            aggregating = true;
+            return true;
+        } else if (aggregating && isContentMessage(in)) {
+            return true;
+        }
+
+        return false;
     }
 
     /**
@@ -192,6 +207,8 @@ protected final ChannelHandlerContext ctx() {
 
     @Override
     protected void decode(final ChannelHandlerContext ctx, I msg, List<Object> out) throws Exception {
+        assert aggregating;
+
         if (isStartMessage(msg)) {
             handlingOversizedMessage = false;
             if (currentMessage != null) {
@@ -246,7 +263,7 @@ public void operationComplete(ChannelFuture future) throws Exception {
                 } else {
                     aggregated = beginAggregation(m, EMPTY_BUFFER);
                 }
-                finishAggregation(aggregated);
+                finishAggregation0(aggregated);
                 out.add(aggregated);
                 return;
             }
@@ -301,7 +318,7 @@ public void operationComplete(ChannelFuture future) throws Exception {
             }
 
             if (last) {
-                finishAggregation(currentMessage);
+                finishAggregation0(currentMessage);
 
                 // All done
                 out.add(currentMessage);
@@ -371,6 +388,11 @@ protected abstract Object newContinueResponse(S start, int maxContentLength, Cha
      */
     protected void aggregate(O aggregated, C content) throws Exception { }
 
+    private void finishAggregation0(O aggregated) throws Exception {
+        aggregating = false;
+        finishAggregation(aggregated);
+    }
+
     /**
      * Invoked when the specified {@code aggregated} message is about to be passed to the next handler in the pipeline.
      */
@@ -378,6 +400,7 @@ protected void finishAggregation(O aggregated) throws Exception { }
 
     private void invokeHandleOversizedMessage(ChannelHandlerContext ctx, S oversized) throws Exception {
         handlingOversizedMessage = true;
+        aggregating = false;
         currentMessage = null;
         try {
             handleOversizedMessage(ctx, oversized);
@@ -441,6 +464,7 @@ private void releaseCurrentMessage() {
             currentMessage.release();
             currentMessage = null;
             handlingOversizedMessage = false;
+            aggregating = false;
         }
     }
 }


Test patch:
diff --git a/codec-http/src/test/java/io/netty/handler/codec/http/HttpObjectAggregatorTest.java b/codec-http/src/test/java/io/netty/handler/codec/http/HttpObjectAggregatorTest.java
index 7dc0ac5a519..503c93339f0 100644
--- a/codec-http/src/test/java/io/netty/handler/codec/http/HttpObjectAggregatorTest.java
+++ b/codec-http/src/test/java/io/netty/handler/codec/http/HttpObjectAggregatorTest.java
@@ -23,7 +23,10 @@
 import io.netty.handler.codec.DecoderResult;
 import io.netty.handler.codec.DecoderResultProvider;
 import io.netty.handler.codec.TooLongFrameException;
+import io.netty.util.AsciiString;
 import io.netty.util.CharsetUtil;
+import io.netty.util.ReferenceCountUtil;
+
 import org.junit.Test;
 import org.mockito.Mockito;
 
@@ -40,6 +43,7 @@
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.junit.Assert.assertSame;
 
 public class HttpObjectAggregatorTest {
 
@@ -517,4 +521,123 @@ public void testReplaceAggregatedResponse() {
         aggregatedRep.release();
         replacedRep.release();
     }
+
+    @Test
+    public void testSelectiveRequestAggregation() {
+        HttpObjectAggregator myPostAggregator = new HttpObjectAggregator(1024 * 1024) {
+            @Override
+            protected boolean isStartMessage(HttpObject msg) throws Exception {
+                if (msg instanceof HttpRequest) {
+                    HttpRequest request = (HttpRequest) msg;
+                    HttpMethod method = request.method();
+
+                    if (method.equals(HttpMethod.POST)) {
+                        return true;
+                    }
+                }
+
+                return false;
+            }
+        };
+
+        EmbeddedChannel channel = new EmbeddedChannel(myPostAggregator);
+
+        try {
+            // Aggregate: POST
+            HttpRequest request1 = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, "/");
+            HttpContent content1 = new DefaultHttpContent(Unpooled.copiedBuffer("Hello, World!", CharsetUtil.UTF_8));
+            request1.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN);
+
+            assertTrue(channel.writeInbound(request1, content1, LastHttpContent.EMPTY_LAST_CONTENT));
+
+            // Getting an aggregated response out
+            Object msg1 = channel.readInbound();
+            try {
+                assertTrue(msg1 instanceof FullHttpRequest);
+            } finally {
+                ReferenceCountUtil.release(msg1);
+            }
+
+            // Don't aggregate: non-POST
+            HttpRequest request2 = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.PUT, "/");
+            HttpContent content2 = new DefaultHttpContent(Unpooled.copiedBuffer("Hello, World!", CharsetUtil.UTF_8));
+            request2.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN);
+
+            try {
+                assertTrue(channel.writeInbound(request2, content2, LastHttpContent.EMPTY_LAST_CONTENT));
+
+                // Getting the same response objects out
+                assertSame(request2, channel.readInbound());
+                assertSame(content2, channel.readInbound());
+                assertSame(LastHttpContent.EMPTY_LAST_CONTENT, channel.readInbound());
+            } finally {
+              ReferenceCountUtil.release(request2);
+              ReferenceCountUtil.release(content2);
+            }
+
+            assertFalse(channel.finish());
+        } finally {
+          channel.close();
+        }
+    }
+
+    @Test
+    public void testSelectiveResponseAggregation() {
+        HttpObjectAggregator myTextAggregator = new HttpObjectAggregator(1024 * 1024) {
+            @Override
+            protected boolean isStartMessage(HttpObject msg) throws Exception {
+                if (msg instanceof HttpResponse) {
+                    HttpResponse response = (HttpResponse) msg;
+                    HttpHeaders headers = response.headers();
+
+                    String contentType = headers.get(HttpHeaderNames.CONTENT_TYPE);
+                    if (AsciiString.contentEqualsIgnoreCase(contentType, HttpHeaderValues.TEXT_PLAIN)) {
+                        return true;
+                    }
+                }
+
+                return false;
+            }
+        };
+
+        EmbeddedChannel channel = new EmbeddedChannel(myTextAggregator);
+
+        try {
+            // Aggregate: text/plain
+            HttpResponse response1 = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
+            HttpContent content1 = new DefaultHttpContent(Unpooled.copiedBuffer("Hello, World!", CharsetUtil.UTF_8));
+            response1.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.TEXT_PLAIN);
+
+            assertTrue(channel.writeInbound(response1, content1, LastHttpContent.EMPTY_LAST_CONTENT));
+
+            // Getting an aggregated response out
+            Object msg1 = channel.readInbound();
+            try {
+                assertTrue(msg1 instanceof FullHttpResponse);
+            } finally {
+                ReferenceCountUtil.release(msg1);
+            }
+
+            // Don't aggregate: application/json
+            HttpResponse response2 = new DefaultHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
+            HttpContent content2 = new DefaultHttpContent(Unpooled.copiedBuffer("{key: 'value'}", CharsetUtil.UTF_8));
+            response2.headers().set(HttpHeaderNames.CONTENT_TYPE, HttpHeaderValues.APPLICATION_JSON);
+
+            try {
+                assertTrue(channel.writeInbound(response2, content2, LastHttpContent.EMPTY_LAST_CONTENT));
+
+                // Getting the same response objects out
+                assertSame(response2, channel.readInbound());
+                assertSame(content2, channel.readInbound());
+                assertSame(LastHttpContent.EMPTY_LAST_CONTENT, channel.readInbound());
+            } finally {
+                ReferenceCountUtil.release(response2);
+                ReferenceCountUtil.release(content2);
+            }
+
+            assertFalse(channel.finish());
+        } finally {
+          channel.close();
+        }
+    }
 }