QPID-1025: updated fix for empty payload issue, this change removes state transitions that don't consume input bytes

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

2 files changed, 34 insertions, 15 deletions

diff --git a/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java b/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java
index ad13c45575..0ff3455624 100644
--- a/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java
+++ b/java/client/src/test/java/org/apache/qpid/test/unit/topic/TopicSessionTest.java
@@ -180,7 +180,7 @@ public class TopicSessionTest extends QpidTestCase
         con.start();
         TextMessage tm = session1.createTextMessage("Hello");
         publisher.publish(tm);
-        tm = (TextMessage) consumer1.receive(200000L);
+        tm = (TextMessage) consumer1.receive(10000L);
         assertNotNull(tm);
         String msgText = tm.getText();
         assertEquals("Hello", msgText);
@@ -188,7 +188,7 @@ public class TopicSessionTest extends QpidTestCase
         msgText = tm.getText();
         assertNull(msgText);
         publisher.publish(tm);
-        tm = (TextMessage) consumer1.receive(20000000L);
+        tm = (TextMessage) consumer1.receive(10000L);
         assertNotNull(tm);
         msgText = tm.getText();
         assertNull(msgText);
diff --git a/java/common/src/main/java/org/apache/qpidity/transport/network/InputHandler.java b/java/common/src/main/java/org/apache/qpidity/transport/network/InputHandler.java
index 6d8a98cdfe..d1c03348b4 100644
--- a/java/common/src/main/java/org/apache/qpidity/transport/network/InputHandler.java
+++ b/java/common/src/main/java/org/apache/qpidity/transport/network/InputHandler.java
@@ -63,9 +63,7 @@ public class InputHandler implements Receiver<ByteBuffer>
         FRAME_HDR_RSVD3,
         FRAME_HDR_RSVD4,
         FRAME_HDR_RSVD5,
-        FRAME_PAYLOAD,
         FRAME_FRAGMENT,
-        FRAME_END,
         ERROR;
     }
 
@@ -113,7 +111,7 @@ public class InputHandler implements Receiver<ByteBuffer>
 
     public void received(ByteBuffer buf)
     {
-        while (buf.hasRemaining() || state == FRAME_PAYLOAD)
+        while (buf.hasRemaining())
         {
             state = next(buf);
         }
@@ -188,8 +186,11 @@ public class InputHandler implements Receiver<ByteBuffer>
         case FRAME_HDR_RSVD4:
             return expect(buf, 0, FRAME_HDR_RSVD5);
         case FRAME_HDR_RSVD5:
-            return expect(buf, 0, FRAME_PAYLOAD);
-        case FRAME_PAYLOAD:
+            if (!expect(buf, 0))
+            {
+                return ERROR;
+            }
+
             frame = new Frame(flags, type, track, channel);
             if (size > buf.remaining()) {
                 frame.addFragment(buf.slice());
@@ -201,7 +202,7 @@ public class InputHandler implements Receiver<ByteBuffer>
                 buf.position(buf.position() + size);
                 frame.addFragment(payload);
                 frame();
-                return FRAME_END;
+                return FRAME_HDR;
             }
         case FRAME_FRAGMENT:
             int delta = size - frame.getSize();
@@ -215,10 +216,8 @@ public class InputHandler implements Receiver<ByteBuffer>
                 buf.position(buf.position() + delta);
                 frame.addFragment(fragment);
                 frame();
-                return FRAME_END;
+                return FRAME_HDR;
             }
-        case FRAME_END:
-            return FRAME_HDR;
         default:
             throw new IllegalStateException();
         }
@@ -236,15 +235,35 @@ public class InputHandler implements Receiver<ByteBuffer>
 
     private State expect(ByteBuffer buf, byte expected, State next)
     {
-        byte b = buf.get();
-        if (b == expected) {
+        if (expect(buf, expected))
+        {
             return next;
-        } else {
-            error("expecting '%x', got '%x'", expected, b);
+        }
+        else
+        {
             return ERROR;
         }
     }
 
+    private boolean expect(ByteBuffer buf, int expected)
+    {
+        return expect(buf, (byte) expected);
+    }
+
+    private boolean expect(ByteBuffer buf, byte expected)
+    {
+        byte b = buf.get();
+        if (b == expected)
+        {
+            return true;
+        }
+        else
+        {
+            error("expecting '%x', got '%x'", expected, b);
+            return false;
+        }
+    }
+
     public void exception(Throwable t)
     {
         receiver.exception(t);