Empty Payload Response treated as Complete frame (#206)

#### Problem

`FrameHeaderFlyweight` resolves the frame type as `NEXT_COMPLETE` only if the payload is not empty. Otherwise, the frame is treated as `COMPLETE`.
This is a problem for request-response as this means the response publisher will complete without emitting a response, if the response does not have any data. Such a condition is a failure as request-response should have exaclty one response.

 #### Modification

 Eliminated the check for data length and always emit `NEXT_COMPLETE` if the complete flag is set.
 According to the [protocol](https://github.com/ReactiveSocket/reactivesocket/blob/master/Protocol.md#response-frame):

 ```
 A Response is generally referred to as a NEXT.
 ```

 So, this behavior would not be a violation of the protocol but will produce empty `onNext` before `onComplete`

 #### Result

 Better behavior with empty responses.

Author: NiteshKant

reactivesocket-core/src/main/java/io/reactivesocket/frame/FrameHeaderFlyweight.java

@@ -148,8 +148,8 @@ public static int encode(
         final int frameLength = computeFrameHeaderLength(frameType, metadata.remaining(), data.remaining());
 
         final FrameType outFrameType;
-
         switch (frameType) {
+            case NEXT_COMPLETE:
             case COMPLETE:
                 outFrameType = FrameType.RESPONSE;
                 flags |= FLAGS_RESPONSE_C;
@@ -162,10 +162,10 @@ public static int encode(
                 break;
         }
 
-        int length = FrameHeaderFlyweight.encodeFrameHeader(mutableDirectBuffer, offset, frameLength, flags, outFrameType, streamId);
+        int length = encodeFrameHeader(mutableDirectBuffer, offset, frameLength, flags, outFrameType, streamId);
 
-        length += FrameHeaderFlyweight.encodeMetadata(mutableDirectBuffer, offset, offset + length, metadata);
-        length += FrameHeaderFlyweight.encodeData(mutableDirectBuffer, offset + length, data);
+        length += encodeMetadata(mutableDirectBuffer, offset, offset + length, metadata);
+        length += encodeData(mutableDirectBuffer, offset + length, data);
 
         return length;
     }
@@ -179,13 +179,10 @@ public static FrameType frameType(final DirectBuffer directBuffer, final int off
 
         if (FrameType.RESPONSE == result) {
             final int flags = flags(directBuffer, offset);
-            final int dataLength = dataLength(directBuffer, offset, 0);
 
             boolean complete = FLAGS_RESPONSE_C == (flags & FLAGS_RESPONSE_C);
-            if (complete && 0 < dataLength) {
+            if (complete) {
                 result = FrameType.NEXT_COMPLETE;
-            } else if (complete) {
-                result = FrameType.COMPLETE;
             } else {
                 result = FrameType.NEXT;
             }
@@ -233,7 +230,7 @@ private static int frameLength(final DirectBuffer directBuffer, final int offset
     }
 
     private static int computeMetadataLength(final int metadataPayloadLength) {
-        return metadataPayloadLength + ((0 == metadataPayloadLength) ? 0 : BitUtil.SIZE_OF_INT);
+        return metadataPayloadLength + (0 == metadataPayloadLength? 0 : BitUtil.SIZE_OF_INT);
     }
 
     private static int metadataFieldLength(final DirectBuffer directBuffer, final int offset) {

reactivesocket-core/src/main/java/io/reactivesocket/internal/RemoteSender.java

@@ -142,7 +142,8 @@ public void onSubscribe(Subscription s) {
     @Override
     public void onNext(Frame frame) {
         // No flow-control check
-        assert frame.getType() != FrameType.ERROR && frame.getType() != FrameType.COMPLETE;
+        FrameType frameType = frame.getType();
+        assert frameType != FrameType.ERROR && !isCompleteFrame(frameType);
         synchronized (this) {
             outstanding--;
         }
@@ -229,10 +230,14 @@ private boolean trySendTerminalFrame(Frame frame, Throwable optionalError) {
 
     private void unsafeSendTerminalFrameToTransport(Frame terminalFrame, Throwable optionalError) {
         transportSubscription.safeOnNext(terminalFrame);
-        if (terminalFrame.getType() == FrameType.COMPLETE) {
+        if (terminalFrame.getType() == FrameType.COMPLETE || terminalFrame.getType() == FrameType.NEXT_COMPLETE) {
             transportSubscription.safeOnComplete();
         } else {
             transportSubscription.safeOnError(optionalError);
         }
     }
+
+    private static boolean isCompleteFrame(FrameType frameType) {
+        return frameType == FrameType.COMPLETE || frameType == FrameType.NEXT_COMPLETE;
+    }
 }

reactivesocket-core/src/test/java/io/reactivesocket/ServerReactiveSocketTest.java

@@ -29,9 +29,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.empty;
-import static org.hamcrest.Matchers.hasSize;
-import static org.hamcrest.Matchers.is;
+import static org.hamcrest.Matchers.*;
 
 public class ServerReactiveSocketTest {
 
@@ -60,7 +58,8 @@ public void testHandleResponseFrameNoError() throws Exception {
         assertThat("Unexpected error.", rule.errors, is(empty()));
         TestSubscriber<Frame> sendSub = sendSubscribers.iterator().next();
         sendSub.request(2);
-        assertThat("Unexpected frame sent.", rule.connection.awaitSend().getType(), is(FrameType.COMPLETE));
+        assertThat("Unexpected frame sent.", rule.connection.awaitSend().getType(),
+                   anyOf(is(FrameType.COMPLETE), is(FrameType.NEXT_COMPLETE)));
     }
 
     @Test(timeout = 2000)

reactivesocket-core/src/test/java/io/reactivesocket/internal/RemoteSenderTest.java

@@ -77,7 +77,7 @@ public void testOnComplete() throws Exception {
         receiverSub.assertValue(new Predicate<Frame>() {
             @Override
             public boolean test(Frame frame) throws Exception {
-                return frame.getType() == FrameType.COMPLETE;
+                return frame.getType() == FrameType.COMPLETE || frame.getType() == FrameType.NEXT_COMPLETE;
             }
         });
 
@@ -138,7 +138,7 @@ public void testOnCompleteWithBuffer() throws Exception {
         receiverSub.assertValue(new Predicate<Frame>() {
             @Override
             public boolean test(Frame frame) throws Exception {
-                return frame.getType() == FrameType.COMPLETE;
+                return frame.getType() == FrameType.COMPLETE || frame.getType() == FrameType.NEXT_COMPLETE;
             }
         });
         receiverSub.assertNoErrors();

reactivesocket-examples/src/main/java/io/reactivesocket/examples/transport/tcp/requestresponse/HelloWorldClient.java

@@ -55,10 +55,10 @@ public Publisher<Payload> requestResponse(Payload p) {
                                                               .connect())
                                     .blockingFirst();
 
-        Flowable.fromPublisher(socket.requestResponse(new PayloadImpl("Hello")))
+        Flowable.fromPublisher(socket.requestResponse(PayloadImpl.EMPTY))
                 .map(payload -> payload.getData())
                 .map(ByteBufferUtil::toUtf8String)
-                .doOnNext(System.out::println)
+                .doOnNext(x -> System.out.println("===>>>> " + x))
                 .concatWith(Flowable.fromPublisher(socket.close()).cast(String.class))
                 .blockingFirst();
     }