HttpObjectAggregator doesn't check content-length header

Motivation:
The HttpObjectAggregator always responds with a 100-continue response. It should check the Content-Length header to see if the content length is OK, and if not responds with a 417.

Modifications:
- HttpObjectAggregator checks the Content-Length header in the case of a 100-continue.

Result:
HttpObjectAggregator responds with 417 if content is known to be too big.

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

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2015 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.http;
+
+/**
+ * A user event designed to communicate that a expectation has failed and there should be no expectation that a
+ * body will follow.
+ */
+public final class HttpExpectationFailedEvent {
+    public static final HttpExpectationFailedEvent INSTANCE = new HttpExpectationFailedEvent();
+    private HttpExpectationFailedEvent() { }
+}

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

@@ -135,6 +135,16 @@ public final class HttpHeaderUtil {
         return defaultValue;
     }
 
+    /**
+     * Get an {@code int} representation of {@link #getContentLength(HttpMessage, long)}.
+     * @return the content length or {@code defaultValue} if this message does
+     *         not have the {@code "Content-Length"} header or its value is not
+     *         a number. Not to exceed the boundaries of integer.
+     */
+    public static int getContentLength(HttpMessage message, int defaultValue) {
+        return (int) Math.min(Integer.MAX_VALUE, HttpHeaderUtil.getContentLength(message, (long) defaultValue));
+    }
+
     /**
      * Returns the content length of the specified web socket message.  If the
      * specified message is not a web socket message, {@code -1} is returned.

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

@@ -29,6 +29,9 @@ import io.netty.handler.codec.TooLongFrameException;
 import io.netty.util.internal.logging.InternalLogger;
 import io.netty.util.internal.logging.InternalLoggerFactory;
 
+import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH;
+import static io.netty.handler.codec.http.HttpHeaderUtil.getContentLength;
+
 /**
  * A {@link ChannelHandler} that aggregates an {@link HttpMessage}
  * and its following {@link HttpContent}s into a single {@link FullHttpRequest}
@@ -50,28 +53,43 @@ import io.netty.util.internal.logging.InternalLoggerFactory;
  */
 public class HttpObjectAggregator
         extends MessageAggregator<HttpObject, HttpMessage, HttpContent, FullHttpMessage> {
-
     private static final InternalLogger logger = InternalLoggerFactory.getInstance(HttpObjectAggregator.class);
-    private static final FullHttpResponse CONTINUE = new DefaultFullHttpResponse(
+    private static final FullHttpResponse CONTINUE =
-            HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE, Unpooled.EMPTY_BUFFER);
+            new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE, Unpooled.EMPTY_BUFFER);
+    private static final FullHttpResponse EXPECTATION_FAILED = new DefaultFullHttpResponse(
+            HttpVersion.HTTP_1_1, HttpResponseStatus.EXPECTATION_FAILED, Unpooled.EMPTY_BUFFER);
     private static final FullHttpResponse TOO_LARGE = new DefaultFullHttpResponse(
             HttpVersion.HTTP_1_1, HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE, Unpooled.EMPTY_BUFFER);
 
     static {
-        TOO_LARGE.headers().set(HttpHeaderNames.CONTENT_LENGTH, 0);
+        EXPECTATION_FAILED.headers().set(CONTENT_LENGTH, 0);
+        TOO_LARGE.headers().set(CONTENT_LENGTH, 0);
+    }
+
+    private final boolean closeOnExpectationFailed;
+
+    /**
+     * Creates a new instance.
+     * @param maxContentLength the maximum length of the aggregated content in bytes.
+     * If the length of the aggregated content exceeds this value,
+     * {@link #handleOversizedMessage(ChannelHandlerContext, HttpMessage)} will be called.
+     */
+    public HttpObjectAggregator(int maxContentLength) {
+        this(maxContentLength, false);
     }
 
     /**
      * Creates a new instance.
-     *
+     * @param maxContentLength the maximum length of the aggregated content in bytes.
-     * @param maxContentLength
-     *        the maximum length of the aggregated content in bytes.
      * If the length of the aggregated content exceeds this value,
-     *        {@link #handleOversizedMessage(ChannelHandlerContext, HttpMessage)}
+     * {@link #handleOversizedMessage(ChannelHandlerContext, HttpMessage)} will be called.
-     *        will be called.
+     * @param closeOnExpectationFailed If a 100-continue response is detected but the content length is too large
+     * then {@code true} means close the connection. otherwise the connection will remain open and data will be
+     * consumed and discarded until the next request is received.
      */
-    public HttpObjectAggregator(int maxContentLength) {
+    public HttpObjectAggregator(int maxContentLength, boolean closeOnExpectationFailed) {
         super(maxContentLength);
+        this.closeOnExpectationFailed = closeOnExpectationFailed;
     }
 
     @Override
@@ -95,22 +113,32 @@ public class HttpObjectAggregator
     }
 
     @Override
-    protected boolean hasContentLength(HttpMessage start) throws Exception {
+    protected boolean isContentLengthInvalid(HttpMessage start, int maxContentLength) {
-        return HttpHeaderUtil.isContentLengthSet(start);
+        return getContentLength(start, -1) > maxContentLength;
     }
 
     @Override
-    protected long contentLength(HttpMessage start) throws Exception {
+    protected Object newContinueResponse(HttpMessage start, int maxContentLength, ChannelPipeline pipeline) {
-        return HttpHeaderUtil.getContentLength(start);
-    }
-
-    @Override
-    protected Object newContinueResponse(HttpMessage start) throws Exception {
         if (HttpHeaderUtil.is100ContinueExpected(start)) {
-            return CONTINUE;
+            if (getContentLength(start, -1) <= maxContentLength) {
-        } else {
+                return CONTINUE.duplicate().retain();
+            }
+
+            pipeline.fireUserEventTriggered(HttpExpectationFailedEvent.INSTANCE);
+            return EXPECTATION_FAILED.duplicate().retain();
+        }
         return null;
     }
+
+    @Override
+    protected boolean closeAfterContinueResponse(Object msg) {
+        return closeOnExpectationFailed && ignoreContentAfterContinueResponse(msg);
+    }
+
+    @Override
+    protected boolean ignoreContentAfterContinueResponse(Object msg) {
+        return msg instanceof HttpResponse &&
+               ((HttpResponse) msg).status().code() == HttpResponseStatus.EXPECTATION_FAILED.code();
     }
 
     @Override
@@ -157,7 +185,8 @@ public class HttpObjectAggregator
     protected void handleOversizedMessage(final ChannelHandlerContext ctx, HttpMessage oversized) throws Exception {
         if (oversized instanceof HttpRequest) {
             // send back a 413 and close the connection
-            ChannelFuture future = ctx.writeAndFlush(TOO_LARGE).addListener(new ChannelFutureListener() {
+            ChannelFuture future = ctx.writeAndFlush(TOO_LARGE.duplicate().retain()).addListener(
+                    new ChannelFutureListener() {
                 @Override
                 public void operationComplete(ChannelFuture future) throws Exception {
                     if (!future.isSuccess()) {

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

@@ -431,6 +431,16 @@ public abstract class HttpObjectDecoder extends ByteToMessageDecoder {
             // This method is responsible for ending requests in some situations and must be called
             // when the input has been shutdown.
             super.channelInactive(ctx);
+        } else if (evt instanceof HttpExpectationFailedEvent) {
+            switch (currentState) {
+            case READ_FIXED_LENGTH_CONTENT:
+            case READ_VARIABLE_LENGTH_CONTENT:
+            case READ_CHUNK_SIZE:
+                reset();
+                break;
+            default:
+                break;
+            }
         }
         super.userEventTriggered(ctx, evt);
     }

codec-http/src/main/java/io/netty/handler/codec/http/websocketx/WebSocketFrameAggregator.java

@@ -16,6 +16,7 @@
 package io.netty.handler.codec.http.websocketx;
 
 import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelPipeline;
 import io.netty.handler.codec.MessageAggregator;
 import io.netty.handler.codec.TooLongFrameException;
 
@@ -63,18 +64,23 @@ public class WebSocketFrameAggregator
     }
 
     @Override
-    protected boolean hasContentLength(WebSocketFrame start) throws Exception {
+    protected boolean isContentLengthInvalid(WebSocketFrame start, int maxContentLength) {
         return false;
     }
 
     @Override
-    protected long contentLength(WebSocketFrame start) throws Exception {
+    protected Object newContinueResponse(WebSocketFrame start, int maxContentLength, ChannelPipeline pipeline) {
+        return null;
+    }
+
+    @Override
+    protected boolean closeAfterContinueResponse(Object msg) throws Exception {
         throw new UnsupportedOperationException();
     }
 
     @Override
-    protected Object newContinueResponse(WebSocketFrame start) throws Exception {
+    protected boolean ignoreContentAfterContinueResponse(Object msg) throws Exception {
-        return null;
+        throw new UnsupportedOperationException();
     }
 
     @Override

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

@@ -29,9 +29,16 @@ import org.junit.Test;
 import java.nio.channels.ClosedChannelException;
 import java.util.List;
 
-import static io.netty.util.ReferenceCountUtil.*;
+import static io.netty.util.ReferenceCountUtil.releaseLater;
-import static org.hamcrest.CoreMatchers.*;
+import static org.hamcrest.CoreMatchers.instanceOf;
-import static org.junit.Assert.*;
+import static org.hamcrest.CoreMatchers.is;
+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.assertThat;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 
 public class HttpObjectAggregatorTest {
 
@@ -146,129 +153,6 @@ public class HttpObjectAggregatorTest {
         checkOversizedRequest(message);
     }
 
-    @Test
-    public void testOversizedRequestWith100Continue() {
-        EmbeddedChannel embedder = new EmbeddedChannel(new HttpObjectAggregator(8));
-
-        // send an oversized request with 100 continue
-        HttpRequest message = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.PUT, "http://localhost");
-        HttpHeaderUtil.set100ContinueExpected(message, true);
-        HttpHeaderUtil.setContentLength(message, 16);
-
-        HttpContent chunk1 = releaseLater(new DefaultHttpContent(Unpooled.copiedBuffer("some", CharsetUtil.US_ASCII)));
-        HttpContent chunk2 = releaseLater(new DefaultHttpContent(Unpooled.copiedBuffer("test", CharsetUtil.US_ASCII)));
-        HttpContent chunk3 = LastHttpContent.EMPTY_LAST_CONTENT;
-
-        // Send a request with 100-continue + large Content-Length header value.
-        assertFalse(embedder.writeInbound(message));
-
-        // The agregator should respond with '413 Request Entity Too Large.'
-        FullHttpResponse response = embedder.readOutbound();
-        assertEquals(HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE, response.status());
-        assertEquals("0", response.headers().get(HttpHeaderNames.CONTENT_LENGTH));
-
-        // An ill-behaving client could continue to send data without a respect, and such data should be discarded.
-        assertFalse(embedder.writeInbound(chunk1));
-
-        // The aggregator should not close the connection because keep-alive is on.
-        assertTrue(embedder.isOpen());
-
-        // Now send a valid request.
-        HttpRequest message2 = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.PUT, "http://localhost");
-
-        assertFalse(embedder.writeInbound(message2));
-        assertFalse(embedder.writeInbound(chunk2));
-        assertTrue(embedder.writeInbound(chunk3));
-
-        FullHttpRequest fullMsg = embedder.readInbound();
-        assertNotNull(fullMsg);
-
-        assertEquals(
-                chunk2.content().readableBytes() + chunk3.content().readableBytes(),
-                HttpHeaderUtil.getContentLength(fullMsg));
-
-        assertEquals(HttpHeaderUtil.getContentLength(fullMsg), fullMsg.content().readableBytes());
-
-        fullMsg.release();
-        assertFalse(embedder.finish());
-    }
-
-    @Test
-    public void testOversizedRequestWith100ContinueAndDecoder() {
-        EmbeddedChannel embedder = new EmbeddedChannel(new HttpRequestDecoder(), new HttpObjectAggregator(4));
-        embedder.writeInbound(Unpooled.copiedBuffer(
-                "PUT /upload HTTP/1.1\r\n" +
-                        "Expect: 100-continue\r\n" +
-                        "Content-Length: 100\r\n\r\n", CharsetUtil.US_ASCII));
-
-        assertNull(embedder.readInbound());
-
-        FullHttpResponse response = embedder.readOutbound();
-        assertEquals(HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE, response.status());
-        assertEquals("0", response.headers().get(HttpHeaderNames.CONTENT_LENGTH));
-
-        // Keep-alive is on by default in HTTP/1.1, so the connection should be still alive.
-        assertTrue(embedder.isOpen());
-
-        // The decoder should be reset by the aggregator at this point and be able to decode the next request.
-        embedder.writeInbound(Unpooled.copiedBuffer("GET /max-upload-size HTTP/1.1\r\n\r\n", CharsetUtil.US_ASCII));
-
-        FullHttpRequest request = embedder.readInbound();
-        assertThat(request.method(), is(HttpMethod.GET));
-        assertThat(request.uri(), is("/max-upload-size"));
-        assertThat(request.content().readableBytes(), is(0));
-        request.release();
-
-        assertFalse(embedder.finish());
-    }
-
-    @Test
-    public void testRequestAfterOversized100ContinueAndDecoder() {
-        EmbeddedChannel embedder = new EmbeddedChannel(new HttpRequestDecoder(), new HttpObjectAggregator(15));
-
-        // Write first request with Expect: 100-continue
-        HttpRequest message = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.PUT, "http://localhost");
-        HttpHeaderUtil.set100ContinueExpected(message, true);
-        HttpHeaderUtil.setContentLength(message, 16);
-
-        HttpContent chunk1 = releaseLater(new DefaultHttpContent(Unpooled.copiedBuffer("some", CharsetUtil.US_ASCII)));
-        HttpContent chunk2 = releaseLater(new DefaultHttpContent(Unpooled.copiedBuffer("test", CharsetUtil.US_ASCII)));
-        HttpContent chunk3 = LastHttpContent.EMPTY_LAST_CONTENT;
-
-        // Send a request with 100-continue + large Content-Length header value.
-        assertFalse(embedder.writeInbound(message));
-
-        // The agregator should respond with '413 Request Entity Too Large.'
-        FullHttpResponse response = embedder.readOutbound();
-        assertEquals(HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE, response.status());
-        assertEquals("0", response.headers().get(HttpHeaderNames.CONTENT_LENGTH));
-
-        // An ill-behaving client could continue to send data without a respect, and such data should be discarded.
-        assertFalse(embedder.writeInbound(chunk1));
-
-        // The aggregator should not close the connection because keep-alive is on.
-        assertTrue(embedder.isOpen());
-
-        // Now send a valid request.
-        HttpRequest message2 = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.PUT, "http://localhost");
-
-        assertFalse(embedder.writeInbound(message2));
-        assertFalse(embedder.writeInbound(chunk2));
-        assertTrue(embedder.writeInbound(chunk3));
-
-        FullHttpRequest fullMsg = embedder.readInbound();
-        assertNotNull(fullMsg);
-
-        assertEquals(
-                chunk2.content().readableBytes() + chunk3.content().readableBytes(),
-                HttpHeaderUtil.getContentLength(fullMsg));
-
-        assertEquals(HttpHeaderUtil.getContentLength(fullMsg), fullMsg.content().readableBytes());
-
-        fullMsg.release();
-        assertFalse(embedder.finish());
-    }
-
     private static void checkOversizedRequest(HttpRequest message) {
         EmbeddedChannel embedder = new EmbeddedChannel(new HttpObjectAggregator(4));
 
@@ -387,4 +271,146 @@ public class HttpObjectAggregatorTest {
         assertNull(ch.readInbound());
         ch.finish();
     }
+
+    @Test
+    public void testOversizedRequestWith100Continue() {
+        EmbeddedChannel embedder = new EmbeddedChannel(new HttpObjectAggregator(8));
+
+        // Send an oversized request with 100 continue.
+        HttpRequest message = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.PUT, "http://localhost");
+        HttpHeaderUtil.set100ContinueExpected(message, true);
+        HttpHeaderUtil.setContentLength(message, 16);
+
+        HttpContent chunk1 = releaseLater(new DefaultHttpContent(Unpooled.copiedBuffer("some", CharsetUtil.US_ASCII)));
+        HttpContent chunk2 = releaseLater(new DefaultHttpContent(Unpooled.copiedBuffer("test", CharsetUtil.US_ASCII)));
+        HttpContent chunk3 = LastHttpContent.EMPTY_LAST_CONTENT;
+
+        // Send a request with 100-continue + large Content-Length header value.
+        assertFalse(embedder.writeInbound(message));
+
+        // The aggregator should respond with '417.'
+        FullHttpResponse response = (FullHttpResponse) embedder.readOutbound();
+        assertEquals(HttpResponseStatus.EXPECTATION_FAILED, response.status());
+        assertEquals("0", response.headers().get(HttpHeaderNames.CONTENT_LENGTH));
+
+        // An ill-behaving client could continue to send data without a respect, and such data should be discarded.
+        assertFalse(embedder.writeInbound(chunk1));
+
+        // The aggregator should not close the connection because keep-alive is on.
+        assertTrue(embedder.isOpen());
+
+        // Now send a valid request.
+        HttpRequest message2 = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.PUT, "http://localhost");
+
+        assertFalse(embedder.writeInbound(message2));
+        assertFalse(embedder.writeInbound(chunk2));
+        assertTrue(embedder.writeInbound(chunk3));
+
+        FullHttpRequest fullMsg = (FullHttpRequest) embedder.readInbound();
+        assertNotNull(fullMsg);
+
+        assertEquals(
+                chunk2.content().readableBytes() + chunk3.content().readableBytes(),
+                HttpHeaderUtil.getContentLength(fullMsg));
+
+        assertEquals(HttpHeaderUtil.getContentLength(fullMsg), fullMsg.content().readableBytes());
+
+        fullMsg.release();
+        assertFalse(embedder.finish());
+    }
+
+    @Test
+    public void testOversizedRequestWith100ContinueAndDecoder() {
+        EmbeddedChannel embedder = new EmbeddedChannel(new HttpRequestDecoder(), new HttpObjectAggregator(4));
+        embedder.writeInbound(Unpooled.copiedBuffer(
+                "PUT /upload HTTP/1.1\r\n" +
+                        "Expect: 100-continue\r\n" +
+                        "Content-Length: 100\r\n\r\n", CharsetUtil.US_ASCII));
+
+        assertNull(embedder.readInbound());
+
+        FullHttpResponse response = (FullHttpResponse) embedder.readOutbound();
+        assertEquals(HttpResponseStatus.EXPECTATION_FAILED, response.status());
+        assertEquals("0", response.headers().get(HttpHeaderNames.CONTENT_LENGTH));
+
+        // Keep-alive is on by default in HTTP/1.1, so the connection should be still alive.
+        assertTrue(embedder.isOpen());
+
+        // The decoder should be reset by the aggregator at this point and be able to decode the next request.
+        embedder.writeInbound(Unpooled.copiedBuffer("GET /max-upload-size HTTP/1.1\r\n\r\n", CharsetUtil.US_ASCII));
+
+        FullHttpRequest request = (FullHttpRequest) embedder.readInbound();
+        assertThat(request.method(), is(HttpMethod.GET));
+        assertThat(request.uri(), is("/max-upload-size"));
+        assertThat(request.content().readableBytes(), is(0));
+        request.release();
+
+        assertFalse(embedder.finish());
+    }
+
+    @Test
+    public void testOversizedRequestWith100ContinueAndDecoderCloseConnection() {
+        EmbeddedChannel embedder = new EmbeddedChannel(new HttpRequestDecoder(), new HttpObjectAggregator(4, true));
+        embedder.writeInbound(Unpooled.copiedBuffer(
+                "PUT /upload HTTP/1.1\r\n" +
+                        "Expect: 100-continue\r\n" +
+                        "Content-Length: 100\r\n\r\n", CharsetUtil.US_ASCII));
+
+        assertNull(embedder.readInbound());
+
+        FullHttpResponse response = (FullHttpResponse) embedder.readOutbound();
+        assertEquals(HttpResponseStatus.EXPECTATION_FAILED, response.status());
+        assertEquals("0", response.headers().get(HttpHeaderNames.CONTENT_LENGTH));
+
+        // We are forcing the connection closed if an expectation is exceeded.
+        assertFalse(embedder.isOpen());
+        assertFalse(embedder.finish());
+    }
+
+    @Test
+    public void testRequestAfterOversized100ContinueAndDecoder() {
+        EmbeddedChannel embedder = new EmbeddedChannel(new HttpRequestDecoder(), new HttpObjectAggregator(15));
+
+        // Write first request with Expect: 100-continue.
+        HttpRequest message = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.PUT, "http://localhost");
+        HttpHeaderUtil.set100ContinueExpected(message, true);
+        HttpHeaderUtil.setContentLength(message, 16);
+
+        HttpContent chunk1 = releaseLater(new DefaultHttpContent(Unpooled.copiedBuffer("some", CharsetUtil.US_ASCII)));
+        HttpContent chunk2 = releaseLater(new DefaultHttpContent(Unpooled.copiedBuffer("test", CharsetUtil.US_ASCII)));
+        HttpContent chunk3 = LastHttpContent.EMPTY_LAST_CONTENT;
+
+        // Send a request with 100-continue + large Content-Length header value.
+        assertFalse(embedder.writeInbound(message));
+
+        // The aggregator should respond with '417'.
+        FullHttpResponse response = (FullHttpResponse) embedder.readOutbound();
+        assertEquals(HttpResponseStatus.EXPECTATION_FAILED, response.status());
+        assertEquals("0", response.headers().get(HttpHeaderNames.CONTENT_LENGTH));
+
+        // An ill-behaving client could continue to send data without a respect, and such data should be discarded.
+        assertFalse(embedder.writeInbound(chunk1));
+
+        // The aggregator should not close the connection because keep-alive is on.
+        assertTrue(embedder.isOpen());
+
+        // Now send a valid request.
+        HttpRequest message2 = new DefaultHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.PUT, "http://localhost");
+
+        assertFalse(embedder.writeInbound(message2));
+        assertFalse(embedder.writeInbound(chunk2));
+        assertTrue(embedder.writeInbound(chunk3));
+
+        FullHttpRequest fullMsg = (FullHttpRequest) embedder.readInbound();
+        assertNotNull(fullMsg);
+
+        assertEquals(
+                chunk2.content().readableBytes() + chunk3.content().readableBytes(),
+                HttpHeaderUtil.getContentLength(fullMsg));
+
+        assertEquals(HttpHeaderUtil.getContentLength(fullMsg), fullMsg.content().readableBytes());
+
+        fullMsg.release();
+        assertFalse(embedder.finish());
+    }
 }

codec-memcache/src/main/java/io/netty/handler/codec/memcache/AbstractMemcacheObjectAggregator.java

@@ -18,6 +18,8 @@ package io.netty.handler.codec.memcache;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelPipeline;
 import io.netty.handler.codec.MessageAggregator;
+import io.netty.handler.codec.http.HttpResponse;
+import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.memcache.binary.BinaryMemcacheRequestDecoder;
 import io.netty.handler.codec.memcache.binary.BinaryMemcacheResponseEncoder;
 
@@ -64,17 +66,22 @@ public abstract class AbstractMemcacheObjectAggregator<H extends MemcacheMessage
     }
 
     @Override
-    protected boolean hasContentLength(H start) throws Exception {
+    protected boolean isContentLengthInvalid(H start, int maxContentLength) {
         return false;
     }
 
     @Override
-    protected long contentLength(H start) throws Exception {
+    protected Object newContinueResponse(H start, int maxContentLength, ChannelPipeline pipeline) {
+        return null;
+    }
+
+    @Override
+    protected boolean closeAfterContinueResponse(Object msg) throws Exception {
         throw new UnsupportedOperationException();
     }
 
     @Override
-    protected Object newContinueResponse(H start) throws Exception {
+    protected boolean ignoreContentAfterContinueResponse(Object msg) throws Exception {
-        return null;
+        throw new UnsupportedOperationException();
     }
 }

codec-stomp/src/main/java/io/netty/handler/codec/stomp/StompSubframeAggregator.java

@@ -63,20 +63,26 @@ public class StompSubframeAggregator
     }
 
     @Override
-    protected boolean hasContentLength(StompHeadersSubframe start) throws Exception {
+    protected boolean isContentLengthInvalid(StompHeadersSubframe start, int maxContentLength) {
-        return start.headers().contains(StompHeaders.CONTENT_LENGTH);
+        return (int) Math.min(Integer.MAX_VALUE, start.headers().getLong(StompHeaders.CONTENT_LENGTH, -1)) >
+                     maxContentLength;
     }
 
     @Override
-    protected long contentLength(StompHeadersSubframe start) throws Exception {
+    protected Object newContinueResponse(StompHeadersSubframe start, int maxContentLength, ChannelPipeline pipeline) {
-        return start.headers().getLong(StompHeaders.CONTENT_LENGTH, 0);
-    }
-
-    @Override
-    protected Object newContinueResponse(StompHeadersSubframe start) throws Exception {
         return null;
     }
 
+    @Override
+    protected boolean closeAfterContinueResponse(Object msg) throws Exception {
+        throw new UnsupportedOperationException();
+    }
+
+    @Override
+    protected boolean ignoreContentAfterContinueResponse(Object msg) throws Exception {
+        throw new UnsupportedOperationException();
+    }
+
     @Override
     protected StompFrame beginAggregation(StompHeadersSubframe start, ByteBuf content) throws Exception {
         StompFrame ret = new DefaultStompFrame(start.command(), content);

codec/src/main/java/io/netty/handler/codec/MessageAggregator.java

@@ -23,6 +23,7 @@ import io.netty.channel.ChannelFuture;
 import io.netty.channel.ChannelFutureListener;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPipeline;
 import io.netty.util.ReferenceCountUtil;
 
 import java.util.List;
@@ -197,17 +198,9 @@ public abstract class MessageAggregator<I, S, C extends ByteBufHolder, O extends
             @SuppressWarnings("unchecked")
             S m = (S) msg;
 
-            // if content length is set, preemptively close if it's too large
-            if (hasContentLength(m)) {
-                if (contentLength(m) > maxContentLength) {
-                    // handle oversized message
-                    invokeHandleOversizedMessage(ctx, m);
-                    return;
-                }
-            }
-
             // Send the continue response if necessary (e.g. 'Expect: 100-continue' header)
-            Object continueResponse = newContinueResponse(m);
+            // Check before content length. Failing an expectation may result in a different response being sent.
+            Object continueResponse = newContinueResponse(m, maxContentLength, ctx.pipeline());
             if (continueResponse != null) {
                 // Cache the write listener for reuse.
                 ChannelFutureListener listener = continueResponseWriteListener;
@@ -221,7 +214,24 @@ public abstract class MessageAggregator<I, S, C extends ByteBufHolder, O extends
                         }
                     };
                 }
-                ctx.writeAndFlush(continueResponse).addListener(listener);
+
+                // Make sure to call this before writing, otherwise reference counts may be invalid.
+                boolean closeAfterWrite = closeAfterContinueResponse(continueResponse);
+                handlingOversizedMessage = ignoreContentAfterContinueResponse(continueResponse);
+
+                final ChannelFuture future = ctx.writeAndFlush(continueResponse).addListener(listener);
+
+                if (closeAfterWrite) {
+                    future.addListener(ChannelFutureListener.CLOSE);
+                    return;
+                }
+                if (handlingOversizedMessage) {
+                    return;
+                }
+            } else if (isContentLengthInvalid(m, maxContentLength)) {
+                // if content length is set, preemptively close if it's too large
+                invokeHandleOversizedMessage(ctx, m);
+                return;
             }
 
             if (m instanceof DecoderResultProvider && !((DecoderResultProvider) m).decoderResult().isSuccess()) {
@@ -316,16 +326,14 @@ public abstract class MessageAggregator<I, S, C extends ByteBufHolder, O extends
     }
 
     /**
-     * Returns {@code true} if and only if the specified start message already contains the information about the
+     * Determine if the message {@code start}'s content length is known, and if it greater than
-     * length of the whole content.
+     * {@code maxContentLength}.
+     * @param start The message which may indicate the content length.
+     * @param maxContentLength The maximum allowed content length.
+     * @return {@code true} if the message {@code start}'s content length is known, and if it greater than
+     * {@code maxContentLength}. {@code false} otherwise.
      */
-    protected abstract boolean hasContentLength(S start) throws Exception;
+    protected abstract boolean isContentLengthInvalid(S start, int maxContentLength) throws Exception;
-
-    /**
-     * Retrieves the length of the whole content from the specified start message. This method is invoked only when
-     * {@link #hasContentLength(Object)} returned {@code true}.
-     */
-    protected abstract long contentLength(S start) throws Exception;
 
     /**
      * Returns the 'continue response' for the specified start message if necessary. For example, this method is
@@ -333,7 +341,26 @@ public abstract class MessageAggregator<I, S, C extends ByteBufHolder, O extends
      *
      * @return the 'continue response', or {@code null} if there's no message to send
      */
-    protected abstract Object newContinueResponse(S start) throws Exception;
+    protected abstract Object newContinueResponse(S start, int maxContentLength, ChannelPipeline pipeline)
+            throws Exception;
+
+    /**
+     * Determine if the channel should be closed after the result of {@link #newContinueResponse(Object)} is written.
+     * @param The return value from {@link #newContinueResponse(Object)}.
+     * @return {@code true} if the channel should be closed after the result of {@link #newContinueResponse(Object)}
+     * is written. {@code false} otherwise.
+     */
+    protected abstract boolean closeAfterContinueResponse(Object msg) throws Exception;
+
+    /**
+     * Determine if all objects for the current request/response should be ignored or not.
+     * Messages will stop being ignored the next time {@link #isContentMessage(Object)} returns {@code true}.
+     *
+     * @param The return value from {@link #newContinueResponse(Object)}.
+     * @return {@code true} if all objects for the current request/response should be ignored or not.
+     * {@code false} otherwise.
+     */
+    protected abstract boolean ignoreContentAfterContinueResponse(Object msg) throws Exception;
 
     /**
      * Creates a new aggregated message from the specified start message and the specified content.  If the start