bug fix and NEXT_COMPLETE support.

Author: tmontgomery

src/main/java/io/reactivesocket/FrameFlyweight.java

@@ -142,12 +142,17 @@ public FrameType frameType(final ByteBuffer byteBuffer)
     {
         frameBuffer.wrap(byteBuffer);
         FrameType result = FrameType.from(frameBuffer.getShort(TYPE_FIELD_OFFSET, ByteOrder.BIG_ENDIAN));
+        final int dataLength = dataLength(byteBuffer, 0);
 
         if (FrameType.RESPONSE == result)
         {
             final int flags = flags(byteBuffer);
 
-            if (FLAGS_C == (flags & FLAGS_C))
+            if (FLAGS_C == (flags & FLAGS_C) && 0 < dataLength)
+            {
+                result = FrameType.NEXT_COMPLETE;
+            }
+            else if (FLAGS_C == (flags & FLAGS_C))
             {
                 result = FrameType.COMPLETE;
             }
@@ -202,7 +207,7 @@ public static ByteBuffer slice(final ByteBuffer byteBuffer, final int position,
         final ByteBuffer result = byteBuffer.slice();
 
         byteBuffer.limit(savedLimit).position(savedPosition);
-        return byteBuffer.slice();
+        return result;
     }
 
     private void ensureByteArrayCapacity(final int length)

src/main/java/io/reactivesocket/FrameType.java

@@ -33,7 +33,8 @@ public enum FrameType
      ERROR(0x21),
      // synthetic types from Responder for use by the rest of the machinery
      NEXT(0x22),
-     COMPLETE(0x23);
+     COMPLETE(0x23),
+     NEXT_COMPLETE(0x24);
 
     private static FrameType[] typesById;
 

src/test/java/io/reactivesocket/FrameTest.java

@@ -45,7 +45,7 @@ public void testWrapMessage() {
 
         f.wrap(2, FrameType.COMPLETE, "done");
         assertEquals("done", f.getData());
-        assertEquals(FrameType.COMPLETE, f.getType());
+        assertEquals(FrameType.NEXT_COMPLETE, f.getType());
         assertEquals(2, f.getStreamId());
     }
 
@@ -58,7 +58,7 @@ public void testWrapBytes() {
         f.wrap(b);
 
         assertEquals("another", f.getData());
-        assertEquals(FrameType.COMPLETE, f.getType());
+        assertEquals(FrameType.NEXT_COMPLETE, f.getType());
         assertEquals(20, f.getStreamId());
     }
 }