QPID-2025: Add a AMQProtocolEngine from the de-MINAfied AMQMinaProtocolSession. Remove various now-unused classes and update references. Add tests for AMQDecoder. Net -1500 lines, +25% performance on transient messaging. Nice.

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/java-network-refactor@810110 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 228 insertions, 2 deletions

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
new file mode 100644
index 0000000000..46c812e265
--- /dev/null
+++ b/qpid/java/common/src/test/java/org/apache/qpid/codec/AMQDecoderTest.java
@@ -0,0 +1,130 @@
+package org.apache.qpid.codec;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+
+import junit.framework.TestCase;
+
+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.HeartbeatBody;
+
+public class AMQDecoderTest extends TestCase
+{
+
+    private AMQCodecFactory _factory;
+    private AMQDecoder _decoder;
+
+
+    public void setUp()
+    {
+        _factory = new AMQCodecFactory(false, null);
+        _decoder = _factory.getDecoder();
+    }
+   
+    
+    public void testSingleFrameDecode() throws AMQProtocolVersionException, AMQFrameDecodingException
+    {
+        ByteBuffer msg = HeartbeatBody.FRAME.toNioByteBuffer();
+        ArrayList<AMQDataBlock> frames = _decoder.decodeBuffer(msg);
+        if (frames.get(0) instanceof AMQFrame)
+        {
+            assertEquals(HeartbeatBody.FRAME.getBodyFrame().getFrameType(), ((AMQFrame) frames.get(0)).getBodyFrame().getFrameType());
+        }
+        else
+        {
+            fail("decode was not a frame");
+        }
+    }
+    
+    public void testPartialFrameDecode() throws AMQProtocolVersionException, AMQFrameDecodingException
+    {
+        ByteBuffer msg = HeartbeatBody.FRAME.toNioByteBuffer();
+        ByteBuffer msgA = msg.slice();
+        int msgbPos = msg.remaining() / 2;
+        int msgaLimit = msg.remaining() - msgbPos;
+        msgA.limit(msgaLimit);
+        msg.position(msgbPos);
+        ByteBuffer msgB = msg.slice();
+        ArrayList<AMQDataBlock> frames = _decoder.decodeBuffer(msgA);
+        assertEquals(0, frames.size());
+        frames = _decoder.decodeBuffer(msgB);
+        assertEquals(1, frames.size());
+        if (frames.get(0) instanceof AMQFrame)
+        {
+            assertEquals(HeartbeatBody.FRAME.getBodyFrame().getFrameType(), ((AMQFrame) frames.get(0)).getBodyFrame().getFrameType());
+        }
+        else
+        {
+            fail("decode was not a frame");
+        }
+    }
+    
+    public void testMultipleFrameDecode() throws AMQProtocolVersionException, AMQFrameDecodingException
+    {
+        ByteBuffer msgA = HeartbeatBody.FRAME.toNioByteBuffer();
+        ByteBuffer msgB = HeartbeatBody.FRAME.toNioByteBuffer();
+        ByteBuffer msg = ByteBuffer.allocate(msgA.remaining() + msgB.remaining());
+        msg.put(msgA);
+        msg.put(msgB);
+        msg.flip();
+        ArrayList<AMQDataBlock> frames = _decoder.decodeBuffer(msg);
+        assertEquals(2, frames.size());
+        for (AMQDataBlock frame : frames)
+        {
+            if (frame instanceof AMQFrame)
+            {
+                assertEquals(HeartbeatBody.FRAME.getBodyFrame().getFrameType(), ((AMQFrame) frame).getBodyFrame().getFrameType());
+            }
+            else
+            {
+                fail("decode was not a frame");
+            }
+        }
+    }
+    
+    public void testMultiplePartialFrameDecode() throws AMQProtocolVersionException, AMQFrameDecodingException
+    {
+        ByteBuffer msgA = HeartbeatBody.FRAME.toNioByteBuffer();
+        ByteBuffer msgB = HeartbeatBody.FRAME.toNioByteBuffer();
+        ByteBuffer msgC = HeartbeatBody.FRAME.toNioByteBuffer();
+        
+        ByteBuffer sliceA = ByteBuffer.allocate(msgA.remaining() + msgB.remaining() / 2);
+        sliceA.put(msgA);
+        int limit = msgB.limit();
+        int pos = msgB.remaining() / 2;
+        msgB.limit(pos);
+        sliceA.put(msgB);
+        sliceA.flip();
+        msgB.limit(limit);
+        msgB.position(pos);
+        
+        ByteBuffer sliceB = ByteBuffer.allocate(msgB.remaining() + pos);
+        sliceB.put(msgB);
+        msgC.limit(pos);
+        sliceB.put(msgC);
+        sliceB.flip();
+        msgC.limit(limit);
+        
+        ArrayList<AMQDataBlock> frames = _decoder.decodeBuffer(sliceA);
+        assertEquals(1, frames.size());
+        frames = _decoder.decodeBuffer(sliceB);
+        assertEquals(1, frames.size());
+        frames = _decoder.decodeBuffer(msgC);
+        assertEquals(1, frames.size());
+        for (AMQDataBlock frame : frames)
+        {
+            if (frame instanceof AMQFrame)
+            {
+                assertEquals(HeartbeatBody.FRAME.getBodyFrame().getFrameType(), ((AMQFrame) frame).getBodyFrame().getFrameType());
+            }
+            else
+            {
+                fail("decode was not a frame");
+            }
+        }
+    }
+    
+}
diff --git a/qpid/java/common/src/test/java/org/apache/qpid/codec/MockAMQVersionAwareProtocolSession.java b/qpid/java/common/src/test/java/org/apache/qpid/codec/MockAMQVersionAwareProtocolSession.java
new file mode 100644
index 0000000000..bd7fb68d93
--- /dev/null
+++ b/qpid/java/common/src/test/java/org/apache/qpid/codec/MockAMQVersionAwareProtocolSession.java
@@ -0,0 +1,95 @@
+package org.apache.qpid.codec;
+
+import java.nio.ByteBuffer;
+
+import org.apache.qpid.AMQException;
+import org.apache.qpid.framing.AMQDataBlock;
+import org.apache.qpid.framing.AMQMethodBody;
+import org.apache.qpid.framing.ContentBody;
+import org.apache.qpid.framing.ContentHeaderBody;
+import org.apache.qpid.framing.HeartbeatBody;
+import org.apache.qpid.framing.MethodRegistry;
+import org.apache.qpid.framing.ProtocolVersion;
+import org.apache.qpid.protocol.AMQVersionAwareProtocolSession;
+import org.apache.qpid.transport.Sender;
+
+public class MockAMQVersionAwareProtocolSession implements AMQVersionAwareProtocolSession
+{
+
+    @Override
+    public void contentBodyReceived(int channelId, ContentBody body) throws AMQException
+    {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void contentHeaderReceived(int channelId, ContentHeaderBody body) throws AMQException
+    {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public MethodRegistry getMethodRegistry()
+    {
+        return MethodRegistry.getMethodRegistry(ProtocolVersion.v0_9);
+    }
+
+    @Override
+    public void heartbeatBodyReceived(int channelId, HeartbeatBody body) throws AMQException
+    {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void init()
+    {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void methodFrameReceived(int channelId, AMQMethodBody body) throws AMQException
+    {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void setSender(Sender<ByteBuffer> sender)
+    {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public void writeFrame(AMQDataBlock frame)
+    {
+        // TODO Auto-generated method stub
+
+    }
+
+    @Override
+    public byte getProtocolMajorVersion()
+    {
+        // TODO Auto-generated method stub
+        return 0;
+    }
+
+    @Override
+    public byte getProtocolMinorVersion()
+    {
+        // TODO Auto-generated method stub
+        return 0;
+    }
+
+    @Override
+    public ProtocolVersion getProtocolVersion()
+    {
+        // TODO Auto-generated method stub
+        return null;
+    }
+
+}
diff --git a/qpid/java/common/src/test/java/org/apache/qpid/transport/network/mina/MINANetworkDriverTest.java b/qpid/java/common/src/test/java/org/apache/qpid/transport/network/mina/MINANetworkDriverTest.java
index 7901f6a99d..6024875cf5 100644
--- a/qpid/java/common/src/test/java/org/apache/qpid/transport/network/mina/MINANetworkDriverTest.java
+++ b/qpid/java/common/src/test/java/org/apache/qpid/transport/network/mina/MINANetworkDriverTest.java
@@ -299,7 +299,7 @@ public class MINANetworkDriverTest extends TestCase
         _countingEngine.setNewLatch(TEST_DATA.getBytes().length);
         _client.send(ByteBuffer.wrap(TEST_DATA.getBytes()));
         _countingEngine.getExceptionLatch().await(2, TimeUnit.SECONDS);
-        assertEquals("Exception should not been thrown", 0, 
+        assertEquals("Exception should have been thrown", 0, 
                 _countingEngine.getExceptionLatch().getCount());
     } 
     
@@ -321,11 +321,12 @@ public class MINANetworkDriverTest extends TestCase
     {
         EchoProtocolEngine _engine = null;
         
-        public ProtocolEngine newProtocolEngine()
+        public ProtocolEngine newProtocolEngine(NetworkDriver driver)
         {
             if (_engine == null)
             {
                 _engine = new EchoProtocolEngine();
+                _engine.setNetworkDriver(driver);
             }
             return getEngine();
         }