Improved HTTP message streaming in HttpMessageDecoder

netty · Jun 3, 2012 · 78301f4 · 78301f4
1 parent 38314b5
commit 78301f4
Showing 1 changed file with 58 additions and 26 deletions.
diff --git a/src/main/java/org/jboss/netty/handler/codec/http/HttpMessageDecoder.java b/src/main/java/org/jboss/netty/handler/codec/http/HttpMessageDecoder.java
@@ -107,6 +107,7 @@ public abstract class HttpMessageDecoder extends ReplayingDecoder<HttpMessageDec
     private ChannelBuffer content;
     private long chunkSize;
     private int headerSize;
+    private int contentRead;
 
     /**
      * The internal state of {@link HttpMessageDecoder}.
@@ -236,17 +237,24 @@ protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffe
             return null;
         }
         case READ_VARIABLE_LENGTH_CONTENT: {
-            if (content == null) {
-                content = ChannelBuffers.dynamicBuffer(channel.getConfig().getBufferFactory());
+            int toRead = actualReadableBytes();
+            if (toRead > maxChunkSize) {
+                toRead = maxChunkSize;
+            }
+            if (!message.isChunked()) {
+                message.setChunked(true);
+                return new Object[] {message, new DefaultHttpChunk(buffer.readBytes(toRead))};
+            } else {
+                return new DefaultHttpChunk(buffer.readBytes(toRead));
             }
-            //this will cause a replay error until the channel is closed where this will read what's left in the buffer
-            content.writeBytes(buffer.readBytes(buffer.readableBytes()));
-            return reset();
         }
         case READ_VARIABLE_LENGTH_CONTENT_AS_CHUNKS: {
             // Keep reading data as a chunk until the end of connection is reached.
-            int chunkSize = Math.min(maxChunkSize, buffer.readableBytes());
-            HttpChunk chunk = new DefaultHttpChunk(buffer.readBytes(chunkSize));
+            int toRead = actualReadableBytes();
+            if (toRead > maxChunkSize) {
+                toRead = maxChunkSize;
+            }
+            HttpChunk chunk = new DefaultHttpChunk(buffer.readBytes(toRead));
 
             if (!buffer.readable()) {
                 // Reached to the end of the connection.
@@ -259,19 +267,23 @@ protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffe
             return chunk;
         }
         case READ_FIXED_LENGTH_CONTENT: {
-            //we have a content-length so we just read the correct number of bytes
-            readFixedLengthContent(buffer);
-            return reset();
+            return readFixedLengthContent(buffer);
         }
         case READ_FIXED_LENGTH_CONTENT_AS_CHUNKS: {
-            long chunkSize = this.chunkSize;
-            HttpChunk chunk;
-            if (chunkSize > maxChunkSize) {
-                chunk = new DefaultHttpChunk(buffer.readBytes(maxChunkSize));
-                chunkSize -= maxChunkSize;
+            assert this.chunkSize <= Integer.MAX_VALUE;
+            int chunkSize = (int) this.chunkSize;
+            int readLimit = actualReadableBytes();
+            int toRead = chunkSize;
+            if (toRead > maxChunkSize) {
+                toRead = maxChunkSize;
+            }
+            if (toRead > readLimit) {
+                toRead = readLimit;
+            }
+            HttpChunk chunk = new DefaultHttpChunk(buffer.readBytes(toRead));
+            if (chunkSize > toRead) {
+                chunkSize -= toRead;
             } else {
-                assert chunkSize <= Integer.MAX_VALUE;
-                chunk = new DefaultHttpChunk(buffer.readBytes((int) chunkSize));
                 chunkSize = 0;
             }
             this.chunkSize = chunkSize;
@@ -311,14 +323,20 @@ protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffe
             return chunk;
         }
         case READ_CHUNKED_CONTENT_AS_CHUNKS: {
-            long chunkSize = this.chunkSize;
-            HttpChunk chunk;
-            if (chunkSize > maxChunkSize) {
-                chunk = new DefaultHttpChunk(buffer.readBytes(maxChunkSize));
-                chunkSize -= maxChunkSize;
+            assert this.chunkSize <= Integer.MAX_VALUE;
+            int chunkSize = (int) this.chunkSize;
+            int readLimit = actualReadableBytes();
+            int toRead = chunkSize;
+            if (toRead > maxChunkSize) {
+                toRead = maxChunkSize;
+            }
+            if (toRead > readLimit) {
+                toRead = readLimit;
+            }
+            HttpChunk chunk = new DefaultHttpChunk(buffer.readBytes(toRead));
+            if (chunkSize > toRead) {
+                chunkSize -= toRead;
             } else {
-                assert chunkSize <= Integer.MAX_VALUE;
-                chunk = new DefaultHttpChunk(buffer.readBytes((int) chunkSize));
                 chunkSize = 0;
             }
             this.chunkSize = chunkSize;
@@ -416,15 +434,29 @@ private static void skipControlCharacters(ChannelBuffer buffer) {
         }
     }
 
-    private void readFixedLengthContent(ChannelBuffer buffer) {
+    private Object readFixedLengthContent(ChannelBuffer buffer) {
+        //we have a content-length so we just read the correct number of bytes
         long length = HttpHeaders.getContentLength(message, -1);
         assert length <= Integer.MAX_VALUE;
-
+        int toRead = (int) length - contentRead;
+        if (toRead > actualReadableBytes()) {
+            toRead = actualReadableBytes();
+        }
+        contentRead = contentRead + toRead;
+        if (length < contentRead) {
+            if (!message.isChunked()) {
+                message.setChunked(true);
+                return new Object[] {message, new DefaultHttpChunk(buffer.readBytes(toRead))};
+            } else {
+                return new DefaultHttpChunk(buffer.readBytes(toRead));
+            }
+        }
         if (content == null) {
             content = buffer.readBytes((int) length);
         } else {
             content.writeBytes(buffer.readBytes((int) length));
         }
+        return reset();
     }
 
     private State readHeaders(ChannelBuffer buffer) throws TooLongFrameException {