Minimize byte copies by using a CompositeChannelBuffer to concat the chunks. See #413

src/main/java/org/jboss/netty/handler/codec/http/HttpChunkAggregator.java

@@ -23,10 +23,12 @@ import java.util.Map.Entry;
 
 import org.jboss.netty.buffer.ChannelBuffer;
 import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.buffer.CompositeChannelBuffer;
 import org.jboss.netty.channel.ChannelHandler;
 import org.jboss.netty.channel.ChannelHandlerContext;
 import org.jboss.netty.channel.ChannelPipeline;
 import org.jboss.netty.channel.Channels;
+import org.jboss.netty.channel.LifeCycleAwareChannelHandler;
 import org.jboss.netty.channel.MessageEvent;
 import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
 import org.jboss.netty.handler.codec.frame.TooLongFrameException;
@@ -50,7 +52,7 @@ import org.jboss.netty.util.CharsetUtil;
  * @apiviz.landmark
  * @apiviz.has org.jboss.netty.handler.codec.http.HttpChunk oneway - - filters out
  */
-public class HttpChunkAggregator extends SimpleChannelUpstreamHandler {
+public class HttpChunkAggregator extends SimpleChannelUpstreamHandler implements LifeCycleAwareChannelHandler {
 
     private static final ChannelBuffer CONTINUE = ChannelBuffers.copiedBuffer(
             "HTTP/1.1 100 Continue\r\n\r\n", CharsetUtil.US_ASCII);
@@ -58,6 +60,10 @@ public class HttpChunkAggregator extends SimpleChannelUpstreamHandler {
     private final int maxContentLength;
     private HttpMessage currentMessage;
 
+    private ChannelHandlerContext ctx;
+
+    private int maxCumulationBufferComponents = 4;
+
     /**
      * Creates a new instance.
      *
@@ -75,6 +81,37 @@ public class HttpChunkAggregator extends SimpleChannelUpstreamHandler {
         this.maxContentLength = maxContentLength;
     }
 
+    /**
+     * Returns the maximum number of components in the cumulation buffer.  If the number of
+     * the components in the cumulation buffer exceeds this value, the components of the
+     * cumulation buffer are consolidated into a single component, involving memory copies.
+     * The default value of this property is {@code 4}.
+     */
+    public final int getMaxCumulationBufferComponents() {
+        return maxCumulationBufferComponents;
+    }
+
+    /**
+     * Sets the maximum number of components in the cumulation buffer.  If the number of
+     * the components in the cumulation buffer exceeds this value, the components of the
+     * cumulation buffer are consolidated into a single component, involving memory copies.
+     * The default value of this property is {@code 4} and its minimum allowed value is {@code 2}.
+     */
+    public final void setMaxCumulationBufferComponents(int maxCumulationBufferComponents) {
+        if (maxCumulationBufferComponents < 2) {
+            throw new IllegalArgumentException(
+                    "maxCumulationBufferComponents: " + maxCumulationBufferComponents +
+                    " (expected: >= 2)");
+        }
+
+        if (ctx == null) {
+            this.maxCumulationBufferComponents = maxCumulationBufferComponents;
+        } else {
+            throw new IllegalStateException(
+                    "decoder properties cannot be changed once the decoder is added to a pipeline.");
+        }
+    }
+
     @Override
     public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
             throws Exception {
@@ -103,7 +140,6 @@ public class HttpChunkAggregator extends SimpleChannelUpstreamHandler {
                     m.removeHeader(HttpHeaders.Names.TRANSFER_ENCODING);
                 }
                 m.setChunked(false);
-                m.setContent(ChannelBuffers.dynamicBuffer(e.getChannel().getConfig().getBufferFactory()));
                 this.currentMessage = m;
             } else {
                 // Not a chunked message - pass through.
@@ -132,7 +168,9 @@ public class HttpChunkAggregator extends SimpleChannelUpstreamHandler {
                         " bytes.");
             }
 
-            content.writeBytes(chunk.getContent());
+            // Append the content of the chunk
+            appendToCumulation(chunk.getContent());
+
             if (chunk.isLast()) {
                 this.currentMessage = null;
 
@@ -157,4 +195,40 @@ public class HttpChunkAggregator extends SimpleChannelUpstreamHandler {
             ctx.sendUpstream(e);
         }
     }
+
+    protected void appendToCumulation(ChannelBuffer input) {
+        ChannelBuffer cumulation = this.currentMessage.getContent();
+        if (cumulation instanceof CompositeChannelBuffer) {
+            // Make sure the resulting cumulation buffer has no more than 4 components.
+            CompositeChannelBuffer composite = (CompositeChannelBuffer) cumulation;
+            if (composite.numComponents() >= maxCumulationBufferComponents) {
+                currentMessage.setContent(ChannelBuffers.wrappedBuffer(composite.copy(), input));
+            } else {
+                List<ChannelBuffer> decomposed = composite.decompose(0, composite.readableBytes());
+                ChannelBuffer[] buffers = decomposed.toArray(new ChannelBuffer[decomposed.size() + 1]);
+                buffers[buffers.length - 1] = input;
+
+                currentMessage.setContent(ChannelBuffers.wrappedBuffer(buffers));
+            }
+        } else {
+            currentMessage.setContent(ChannelBuffers.wrappedBuffer(cumulation, input));
+        }
+
+    }
+
+    public void beforeAdd(ChannelHandlerContext ctx) throws Exception {
+        this.ctx = ctx;
+    }
+
+    public void afterAdd(ChannelHandlerContext ctx) throws Exception {
+        // noop
+    }
+
+    public void beforeRemove(ChannelHandlerContext ctx) throws Exception {
+        // noop
+    }
+
+    public void afterRemove(ChannelHandlerContext ctx) throws Exception {
+        // noop
+    }
 }

src/test/java/org/jboss/netty/handler/codec/http/HttpChunkAggregatorTest.java

@@ -0,0 +1,142 @@
+/*
+ * Copyright 2012 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 org.jboss.netty.handler.codec.http;
+
+import static org.junit.Assert.*;
+
+import java.util.List;
+
+import org.easymock.EasyMock;
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.buffer.CompositeChannelBuffer;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.handler.codec.embedder.CodecEmbedderException;
+import org.jboss.netty.handler.codec.embedder.DecoderEmbedder;
+import org.jboss.netty.handler.codec.frame.TooLongFrameException;
+import org.jboss.netty.util.CharsetUtil;
+import org.junit.Test;
+
+public class HttpChunkAggregatorTest {
+
+    @Test
+    public void testAggregate() {
+        HttpChunkAggregator aggr = new HttpChunkAggregator(1024 * 1024);
+        DecoderEmbedder<HttpMessage> embedder = new DecoderEmbedder<HttpMessage>(aggr);
+        HttpMessage message = new DefaultHttpMessage(HttpVersion.HTTP_1_1);
+        HttpHeaders.setHeader(message, "X-Test", true);
+        message.setChunked(true);
+        HttpChunk chunk1 = new DefaultHttpChunk(ChannelBuffers.copiedBuffer("test", CharsetUtil.US_ASCII));
+        HttpChunk chunk2 = new DefaultHttpChunk(ChannelBuffers.copiedBuffer("test2", CharsetUtil.US_ASCII));
+        HttpChunk chunk3 = new DefaultHttpChunk(ChannelBuffers.EMPTY_BUFFER);
+        assertFalse(embedder.offer(message));
+        assertFalse(embedder.offer(chunk1));
+        assertFalse(embedder.offer(chunk2));
+        
+        // this should trigger a messageReceived event so return true
+        assertTrue(embedder.offer(chunk3));
+        assertTrue(embedder.finish());
+        HttpMessage aggratedMessage = embedder.poll();
+        assertNotNull(aggratedMessage);
+        
+        assertEquals(chunk1.getContent().readableBytes() + chunk2.getContent().readableBytes(), HttpHeaders.getContentLength(aggratedMessage));
+        assertEquals(aggratedMessage.getHeader("X-Test"), Boolean.TRUE.toString());
+        checkContentBuffer(aggratedMessage);
+        assertNull(embedder.poll());
+
+    }
+
+    private void checkContentBuffer(HttpMessage aggregatedMessage) {
+        CompositeChannelBuffer buffer = (CompositeChannelBuffer) aggregatedMessage.getContent();
+        assertEquals(2, buffer.numComponents());
+        List<ChannelBuffer> buffers = buffer.decompose(0, buffer.capacity());
+        assertEquals(2, buffers.size());
+        for (ChannelBuffer buf: buffers) {
+            // This should be false as we decompose the buffer before to not have deep hierarchy
+            assertFalse(buf instanceof CompositeChannelBuffer);
+        }
+    }
+
+    @Test
+    public void testAggregateWithTrailer() {
+        HttpChunkAggregator aggr = new HttpChunkAggregator(1024 * 1024);
+        DecoderEmbedder<HttpMessage> embedder = new DecoderEmbedder<HttpMessage>(aggr);
+        HttpMessage message = new DefaultHttpMessage(HttpVersion.HTTP_1_1);
+        HttpHeaders.setHeader(message, "X-Test", true);
+        message.setChunked(true);
+        HttpChunk chunk1 = new DefaultHttpChunk(ChannelBuffers.copiedBuffer("test", CharsetUtil.US_ASCII));
+        HttpChunk chunk2 = new DefaultHttpChunk(ChannelBuffers.copiedBuffer("test2", CharsetUtil.US_ASCII));
+        HttpChunkTrailer trailer = new DefaultHttpChunkTrailer();
+        trailer.setHeader("X-Trailer", true);
+        
+        assertFalse(embedder.offer(message));
+        assertFalse(embedder.offer(chunk1));
+        assertFalse(embedder.offer(chunk2));
+        
+        // this should trigger a messageReceived event so return true
+        assertTrue(embedder.offer(trailer));
+        assertTrue(embedder.finish());
+        HttpMessage aggratedMessage = embedder.poll();
+        assertNotNull(aggratedMessage);
+        
+        assertEquals(chunk1.getContent().readableBytes() + chunk2.getContent().readableBytes(), HttpHeaders.getContentLength(aggratedMessage));
+        assertEquals(aggratedMessage.getHeader("X-Test"), Boolean.TRUE.toString());
+        assertEquals(aggratedMessage.getHeader("X-Trailer"), Boolean.TRUE.toString());
+        checkContentBuffer(aggratedMessage);
+
+        assertNull(embedder.poll());
+
+    }
+
+
+    @Test
+    public void testTooLongFrameException() {
+        HttpChunkAggregator aggr = new HttpChunkAggregator(4);
+        DecoderEmbedder<HttpMessage> embedder = new DecoderEmbedder<HttpMessage>(aggr);
+        HttpMessage message = new DefaultHttpMessage(HttpVersion.HTTP_1_1);
+        message.setChunked(true);
+        HttpChunk chunk1 = new DefaultHttpChunk(ChannelBuffers.copiedBuffer("test", CharsetUtil.US_ASCII));
+        HttpChunk chunk2 = new DefaultHttpChunk(ChannelBuffers.copiedBuffer("test2", CharsetUtil.US_ASCII));
+        assertFalse(embedder.offer(message));
+        assertFalse(embedder.offer(chunk1));
+        try {
+            embedder.offer(chunk2);
+            fail();
+        } catch (CodecEmbedderException e) {
+            assertTrue(e.getCause() instanceof TooLongFrameException);
+        }
+    }
+    
+    @Test(expected = IllegalArgumentException.class)
+    public void testInvalidConstructorUsage() {
+        new HttpChunkAggregator(0);
+    }
+    
+    @Test(expected = IllegalArgumentException.class)
+    public void testInvalidMaxCumulationBufferComponents() {
+        HttpChunkAggregator aggr= new HttpChunkAggregator(Integer.MAX_VALUE);
+        aggr.setMaxCumulationBufferComponents(1);
+    }
+    
+    @Test(expected = IllegalStateException.class)
+    public void testSetMaxCumulationBufferComponentsAfterInit() throws Exception {
+        HttpChunkAggregator aggr = new HttpChunkAggregator(Integer.MAX_VALUE);
+        ChannelHandlerContext ctx = EasyMock.createMock(ChannelHandlerContext.class);
+        EasyMock.replay(ctx);
+        aggr.beforeAdd(ctx);
+        aggr.setMaxCumulationBufferComponents(10);
+    }
+}