fixes issue with requestChannel never completes if server cancels (#592) (#594)

Signed-off-by: Oleh Dokuka <[email protected]>

Author: OlegDokuka

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

@@ -471,7 +471,6 @@ private void handleFrame(int streamId, FrameType type, Frame frame) {
         case CANCEL:
         {
           LimitableRequestPublisher sender = senders.remove(streamId);
-          receivers.remove(streamId);
           if (sender != null) {
             sender.cancel();
           }

rsocket-core/src/test/java/io/rsocket/RSocketClientTest.java

@@ -40,6 +40,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.stream.Collectors;
+import org.assertj.core.api.Assertions;
 import org.junit.Rule;
 import org.junit.Test;
 import org.reactivestreams.Publisher;
@@ -49,6 +50,7 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.MonoProcessor;
+import reactor.core.publisher.UnicastProcessor;
 
 public class RSocketClientTest {
 
@@ -195,6 +197,24 @@ public void testChannelRequestCancellation() {
         .blockFirst();
   }
 
+  @Test
+  public void testChannelRequestServerSideCancellation() {
+    MonoProcessor<Payload> cancelled = MonoProcessor.create();
+    UnicastProcessor<Payload> request = UnicastProcessor.create();
+    request.onNext(EmptyPayload.INSTANCE);
+    rule.socket.requestChannel(request).subscribe(cancelled);
+    int streamId = rule.getStreamIdForRequestType(REQUEST_CHANNEL);
+    rule.connection.addToReceivedBuffer(Frame.Cancel.from(streamId));
+    rule.connection.addToReceivedBuffer(Frame.PayloadFrame.from(streamId, NEXT_COMPLETE));
+    Flux.first(
+            cancelled,
+            Flux.error(new IllegalStateException("Channel request not cancelled"))
+                .delaySubscription(Duration.ofSeconds(1)))
+        .blockFirst();
+
+    Assertions.assertThat(request.isDisposed()).isTrue();
+  }
+
   public int sendRequestResponse(Publisher<Payload> response) {
     Subscriber<Payload> sub = TestSubscriber.create();
     response.subscribe(sub);