Requester Channel: Cancel request on response termination (#517)

* Requester Channel: cancel request on response termination

* fix some tests which assume port 8080 is available

Author: mostroverkhov

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

@@ -367,7 +367,10 @@ private Flux<Payload> handleChannel(Flux<Payload> request) {
                   .doFinally(
                       s -> {
                         receivers.remove(streamId);
-                        senders.remove(streamId);
+                        LimitableRequestPublisher sender = senders.remove(streamId);
+                        if (sender != null) {
+                          sender.cancel();
+                        }
                       });
             }));
   }

rsocket-core/src/main/java/io/rsocket/fragmentation/FragmentationDuplexConnection.java

@@ -68,7 +68,8 @@ public FragmentationDuplexConnection(DuplexConnection delegate, int maxFragmentS
    *
    * @param byteBufAllocator the {@link ByteBufAllocator} to use
    * @param delegate the {@link DuplexConnection} to decorate
-   * @param maxFragmentSize the maximum fragment size. A value of 0 indicates that frames should not be fragmented.
+   * @param maxFragmentSize the maximum fragment size. A value of 0 indicates that frames should not
+   *     be fragmented.
    * @throws NullPointerException if {@code byteBufAllocator} or {@code delegate} are {@code null}
    * @throws IllegalArgumentException if {@code maxFragmentSize} is not {@code positive}
    */

rsocket-core/src/main/java/io/rsocket/internal/SwitchTransform.java

@@ -43,8 +43,7 @@ public void subscribe(CoreSubscriber<? super R> actual) {
     source.subscribe(new SwitchTransformSubscriber<>(actual, transformer));
   }
 
-  static final class SwitchTransformSubscriber<T, R>
-      implements CoreSubscriber<T> {
+  static final class SwitchTransformSubscriber<T, R> implements CoreSubscriber<T> {
     @SuppressWarnings("rawtypes")
     static final AtomicIntegerFieldUpdater<SwitchTransformSubscriber> ONCE =
         AtomicIntegerFieldUpdater.newUpdater(SwitchTransformSubscriber.class, "once");

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

@@ -48,6 +48,7 @@
 import reactor.core.publisher.BaseSubscriber;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
+import reactor.core.publisher.MonoProcessor;
 
 public class RSocketClientTest {
 
@@ -182,6 +183,18 @@ public void testLazyRequestResponse() {
     assertThat("Stream ID reused.", streamId2, not(equalTo(streamId)));
   }
 
+  @Test
+  public void testChannelRequestCancellation() {
+    MonoProcessor<Void> cancelled = MonoProcessor.create();
+    Flux<Payload> request = Flux.<Payload>never().doOnCancel(cancelled::onComplete);
+    rule.socket.requestChannel(request).subscribe().dispose();
+    Flux.first(
+            cancelled,
+            Flux.error(new IllegalStateException("Channel request not cancelled"))
+                .delaySubscription(Duration.ofSeconds(1)))
+        .blockFirst();
+  }
+
   public int sendRequestResponse(Publisher<Payload> response) {
     Subscriber<Payload> sub = TestSubscriber.create();
     response.subscribe(sub);

rsocket-transport-netty/src/test/java/io/rsocket/transport/netty/server/WebsocketRouteTransportTest.java

@@ -75,7 +75,7 @@ void newHandlerNullAcceptor() {
   @Test
   void start() {
     WebsocketRouteTransport serverTransport =
-        new WebsocketRouteTransport(HttpServer.create(), routes -> {}, "/test-path");
+        new WebsocketRouteTransport(HttpServer.create(0), routes -> {}, "/test-path");
 
     serverTransport
         .start(duplexConnection -> Mono.empty())
@@ -90,7 +90,7 @@ void startNullAcceptor() {
     assertThatNullPointerException()
         .isThrownBy(
             () ->
-                new WebsocketRouteTransport(HttpServer.create(), routes -> {}, "/test-path")
+                new WebsocketRouteTransport(HttpServer.create(0), routes -> {}, "/test-path")
                     .start(null))
         .withMessage("acceptor must not be null");
   }