QPID-6189 : [Java Common] fix stack overflow bug when frame is broken into thousands of reads/writes

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1634539 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 54 insertions, 0 deletions

diff --git a/qpid/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java b/qpid/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java
index 9d98168687..9bafc30ebc 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java
@@ -47,6 +47,7 @@ import org.apache.qpid.protocol.AMQConstant;
  */
 public abstract class AMQDecoder<T extends MethodProcessor>
 {
+    private static final int MAX_BUFFERS_LIMIT = 10;
     private final T _methodProcessor;
 
     /** Holds the protocol initiation decoder. */
@@ -298,6 +299,25 @@ public abstract class AMQDecoder<T extends MethodProcessor>
                         _remainingBufs.add(new ByteArrayInputStream(remaining));
                     }
                 }
+
+                if(_remainingBufs.size() > MAX_BUFFERS_LIMIT)
+                {
+                    int totalSize = 0;
+                    for(ByteArrayInputStream stream : _remainingBufs)
+                    {
+                        totalSize += stream.available();
+                    }
+
+                    byte[] completeBuffer = new byte[totalSize];
+                    int pos = 0;
+                    for(ByteArrayInputStream stream : _remainingBufs)
+                    {
+                        pos += stream.read(completeBuffer, pos, stream.available());
+                    }
+
+                    _remainingBufs.clear();
+                    _remainingBufs.add(new ByteArrayInputStream(completeBuffer));
+                }
             }
         }
     }
diff --git a/qpid/java/common/src/test/java/org/apache/qpid/codec/AMQDecoderTest.java b/qpid/java/common/src/test/java/org/apache/qpid/codec/AMQDecoderTest.java
index 51f3ce1113..e597192766 100644
--- a/qpid/java/common/src/test/java/org/apache/qpid/codec/AMQDecoderTest.java
+++ b/qpid/java/common/src/test/java/org/apache/qpid/codec/AMQDecoderTest.java
@@ -25,17 +25,22 @@ import java.io.ByteArrayOutputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.Arrays;
 import java.util.List;
+import java.util.Random;
 
 import junit.framework.TestCase;
 
+import org.apache.qpid.framing.AMQBody;
 import org.apache.qpid.framing.AMQDataBlock;
 import org.apache.qpid.framing.AMQFrame;
 import org.apache.qpid.framing.AMQFrameDecodingException;
 import org.apache.qpid.framing.AMQProtocolVersionException;
+import org.apache.qpid.framing.ContentBody;
 import org.apache.qpid.framing.FrameCreatingMethodProcessor;
 import org.apache.qpid.framing.HeartbeatBody;
 import org.apache.qpid.framing.ProtocolVersion;
+import org.apache.qpid.util.BytesDataOutput;
 
 public class AMQDecoderTest extends TestCase
 {
@@ -72,6 +77,35 @@ public class AMQDecoderTest extends TestCase
             fail("decode was not a frame");
         }
     }
+
+
+    public void testDecodeWithManyBuffers() throws AMQProtocolVersionException, AMQFrameDecodingException, IOException
+    {
+        Random random = new Random();
+        final byte[] payload = new byte[2048];
+        random.nextBytes(payload);
+        final AMQBody body = new ContentBody(payload);
+        AMQFrame frame = new AMQFrame(1, body);
+        byte[] outputBuf = new byte[4096];
+        BytesDataOutput dataOutput = new BytesDataOutput(outputBuf);
+        frame.writePayload(dataOutput);
+        for(int i = 0 ; i < dataOutput.length(); i++)
+        {
+            _decoder.decodeBuffer(ByteBuffer.wrap(outputBuf, i, 1));
+
+        }
+        List<AMQDataBlock> frames = _methodProcessor.getProcessedMethods();
+        if (frames.get(0) instanceof AMQFrame)
+        {
+            assertEquals(ContentBody.TYPE, ((AMQFrame) frames.get(0)).getBodyFrame().getFrameType());
+            ContentBody decodedBody = (ContentBody) ((AMQFrame) frames.get(0)).getBodyFrame();
+            assertTrue("Body was corrupted", Arrays.equals(payload, decodedBody.getPayload()));
+        }
+        else
+        {
+            fail("decode was not a frame");
+        }
+    }
     
     public void testPartialFrameDecode() throws AMQProtocolVersionException, AMQFrameDecodingException, IOException
     {