fixes requestChannel and ensure support of half-closed state (#794)

The following is the cases which MUST be supported

OUTER PUBLISHER COMPLETED - INNER PUBLISHER CAN SEND
OUTER PUBLISHER CANCELLED - INNER PUBLISHER CAN SEND
INNER PUBLISHER COMPLETED - OUTER PUBLISHER CAN SEND
INNER PUBLISHER CANCELLED - the WHOLE CHAIN IS TERMINATED

Author: OlegDokuka

rsocket-core/src/main/java/io/rsocket/core/RSocketRequester.java

@@ -560,46 +560,58 @@ private void handleStreamZero(FrameType type, ByteBuf frame) {
 
   private void handleFrame(int streamId, FrameType type, ByteBuf frame) {
     Subscriber<Payload> receiver = receivers.get(streamId);
-    if (receiver == null) {
-      handleMissingResponseProcessor(streamId, type, frame);
-    } else {
-      switch (type) {
-        case ERROR:
-          receiver.onError(Exceptions.from(streamId, frame));
-          receivers.remove(streamId);
-          break;
-        case NEXT_COMPLETE:
-          receiver.onNext(payloadDecoder.apply(frame));
-          receiver.onComplete();
-          break;
-        case CANCEL:
-          {
-            Subscription sender = senders.remove(streamId);
-            if (sender != null) {
-              sender.cancel();
-            }
-            break;
+    switch (type) {
+      case NEXT:
+        if (receiver == null) {
+          handleMissingResponseProcessor(streamId, type, frame);
+          return;
+        }
+        receiver.onNext(payloadDecoder.apply(frame));
+        break;
+      case NEXT_COMPLETE:
+        if (receiver == null) {
+          handleMissingResponseProcessor(streamId, type, frame);
+          return;
+        }
+        receiver.onNext(payloadDecoder.apply(frame));
+        receiver.onComplete();
+        break;
+      case COMPLETE:
+        if (receiver == null) {
+          handleMissingResponseProcessor(streamId, type, frame);
+          return;
+        }
+        receiver.onComplete();
+        receivers.remove(streamId);
+        break;
+      case ERROR:
+        if (receiver == null) {
+          handleMissingResponseProcessor(streamId, type, frame);
+          return;
+        }
+        receiver.onError(Exceptions.from(streamId, frame));
+        receivers.remove(streamId);
+        break;
+      case CANCEL:
+        {
+          Subscription sender = senders.remove(streamId);
+          if (sender != null) {
+            sender.cancel();
           }
-        case NEXT:
-          receiver.onNext(payloadDecoder.apply(frame));
           break;
-        case REQUEST_N:
-          {
-            Subscription sender = senders.get(streamId);
-            if (sender != null) {
-              int n = RequestNFrameFlyweight.requestN(frame);
-              sender.request(n >= Integer.MAX_VALUE ? Long.MAX_VALUE : n);
-            }
-            break;
+        }
+      case REQUEST_N:
+        {
+          Subscription sender = senders.get(streamId);
+          if (sender != null) {
+            int n = RequestNFrameFlyweight.requestN(frame);
+            sender.request(n >= Integer.MAX_VALUE ? Long.MAX_VALUE : n);
           }
-        case COMPLETE:
-          receiver.onComplete();
-          receivers.remove(streamId);
           break;
-        default:
-          throw new IllegalStateException(
-              "Client received supported frame on stream " + streamId + ": " + frame.toString());
-      }
+        }
+      default:
+        throw new IllegalStateException(
+            "Client received supported frame on stream " + streamId + ": " + frame.toString());
     }
   }
 

rsocket-core/src/main/java/io/rsocket/core/RSocketResponder.java

@@ -492,6 +492,24 @@ protected void hookOnError(Throwable throwable) {
             handleError(streamId, throwable);
           }
 
+          @Override
+          protected void hookOnCancel() {
+            // specifically for requestChannel case so when requester sends Cancel frame so the
+            // whole chain MUST be terminated
+            // Note: CancelFrame is redundant from the responder side due to spec
+            // (https://github.com/rsocket/rsocket/blob/master/Protocol.md#request-channel)
+            // Upon receiving a CANCEL, the stream is terminated on the Responder.
+            // Upon sending a CANCEL, the stream is terminated on the Requester.
+            if (requestChannel != null) {
+              channelProcessors.remove(streamId, requestChannel);
+              try {
+                requestChannel.dispose();
+              } catch (Exception e) {
+                // might be thrown back if stream is cancelled
+              }
+            }
+          }
+
           @Override
           protected void hookFinally(SignalType type) {
             sendingSubscriptions.remove(streamId);
@@ -568,6 +586,7 @@ protected void hookOnError(Throwable throwable) {
 
   private void handleCancelFrame(int streamId) {
     Subscription subscription = sendingSubscriptions.remove(streamId);
+    channelProcessors.remove(streamId);
 
     if (subscription != null) {
       subscription.cancel();