fixes onClose behaviour to not error on shutdown

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

Author: OlegDokuka

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

@@ -64,7 +64,6 @@
 import reactor.core.publisher.BaseSubscriber;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.MonoProcessor;
 import reactor.core.publisher.Operators;
 import reactor.core.publisher.SignalType;
 import reactor.core.publisher.UnicastProcessor;
@@ -110,7 +109,6 @@ class RSocketRequester implements RSocket {
   private final RequesterLeaseHandler leaseHandler;
   private final ByteBufAllocator allocator;
   private final KeepAliveFramesAcceptor keepAliveFramesAcceptor;
-  private final MonoProcessor<Void> onClose;
   private final Scheduler serialScheduler;
 
   RSocketRequester(
@@ -131,16 +129,12 @@ class RSocketRequester implements RSocket {
     this.leaseHandler = leaseHandler;
     this.senders = new SynchronizedIntObjectHashMap<>();
     this.receivers = new SynchronizedIntObjectHashMap<>();
-    this.onClose = MonoProcessor.create();
     this.serialScheduler = serialScheduler;
 
     // DO NOT Change the order here. The Send processor must be subscribed to before receiving
     this.sendProcessor = new UnboundedProcessor<>();
 
-    connection
-        .onClose()
-        .or(onClose)
-        .subscribe(null, this::tryTerminateOnConnectionError, this::tryTerminateOnConnectionClose);
+    connection.onClose().subscribe(null, this::tryTerminateOnConnectionError, this::tryShutdown);
     connection.send(sendProcessor).subscribe(null, this::handleSendProcessorError);
 
     connection.receive().subscribe(this::handleIncomingFrames, e -> {});
@@ -193,12 +187,12 @@ public void dispose() {
 
   @Override
   public boolean isDisposed() {
-    return onClose.isDisposed();
+    return connection.isDisposed();
   }
 
   @Override
   public Mono<Void> onClose() {
-    return onClose;
+    return connection.onClose();
   }
 
   private Mono<Void> handleFireAndForget(Payload payload) {
@@ -709,10 +703,6 @@ private void tryTerminateOnConnectionError(Throwable e) {
     tryTerminate(() -> e);
   }
 
-  private void tryTerminateOnConnectionClose() {
-    tryTerminate(() -> CLOSED_CHANNEL_EXCEPTION);
-  }
-
   private void tryTerminateOnZeroError(ByteBuf errorFrame) {
     tryTerminate(() -> Exceptions.from(0, errorFrame));
   }
@@ -726,6 +716,14 @@ private void tryTerminate(Supplier<Throwable> errorSupplier) {
     }
   }
 
+  private void tryShutdown() {
+    if (terminationError == null) {
+      if (TERMINATION_ERROR.compareAndSet(this, null, CLOSED_CHANNEL_EXCEPTION)) {
+        terminate(CLOSED_CHANNEL_EXCEPTION);
+      }
+    }
+  }
+
   private void terminate(Throwable e) {
     connection.dispose();
     leaseHandler.dispose();
@@ -761,7 +759,6 @@ private void terminate(Throwable e) {
     senders.clear();
     receivers.clear();
     sendProcessor.dispose();
-    onClose.onError(e);
   }
 
   private void handleSendProcessorError(Throwable t) {

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

@@ -76,7 +76,6 @@ class RSocketResponder implements RSocket {
   private final PayloadDecoder payloadDecoder;
   private final ResponderLeaseHandler leaseHandler;
   private final Disposable leaseHandlerDisposable;
-  private final MonoProcessor<Void> onClose;
 
   private volatile Throwable terminationError;
   private static final AtomicReferenceFieldUpdater<RSocketResponder, Throwable> TERMINATION_ERROR =
@@ -111,7 +110,6 @@ class RSocketResponder implements RSocket {
     this.leaseHandler = leaseHandler;
     this.sendingSubscriptions = new SynchronizedIntObjectHashMap<>();
     this.channelProcessors = new SynchronizedIntObjectHashMap<>();
-    this.onClose = MonoProcessor.create();
 
     // DO NOT Change the order here. The Send processor must be subscribed to before receiving
     // connections
@@ -124,7 +122,6 @@ class RSocketResponder implements RSocket {
 
     this.connection
         .onClose()
-        .or(onClose)
         .subscribe(null, this::tryTerminateOnConnectionError, this::tryTerminateOnConnectionClose);
   }
 
@@ -257,12 +254,12 @@ public void dispose() {
 
   @Override
   public boolean isDisposed() {
-    return onClose.isDisposed();
+    return connection.isDisposed();
   }
 
   @Override
   public Mono<Void> onClose() {
-    return onClose;
+    return connection.onClose();
   }
 
   private void cleanup(Throwable e) {

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

@@ -41,6 +41,8 @@
 import org.junit.runner.Description;
 import org.junit.runners.model.Statement;
 import org.reactivestreams.Publisher;
+import reactor.core.Disposable;
+import reactor.core.Disposables;
 import reactor.core.publisher.DirectProcessor;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -51,6 +53,34 @@ public class RSocketTest {
 
   @Rule public final SocketRule rule = new SocketRule();
 
+  @Test
+  public void rsocketDisposalShouldEndupWithNoErrorsOnClose() {
+    RSocket requestHandlingRSocket =
+        new RSocket() {
+          final Disposable disposable = Disposables.single();
+
+          @Override
+          public void dispose() {
+            disposable.dispose();
+          }
+
+          @Override
+          public boolean isDisposed() {
+            return disposable.isDisposed();
+          }
+        };
+    rule.setRequestAcceptor(requestHandlingRSocket);
+    rule.crs
+        .onClose()
+        .as(StepVerifier::create)
+        .expectSubscription()
+        .then(rule.crs::dispose)
+        .expectComplete()
+        .verify(Duration.ofMillis(100));
+
+    Assertions.assertThat(requestHandlingRSocket.isDisposed()).isTrue();
+  }
+
   @Test(timeout = 2_000)
   public void testRequestReplyNoError() {
     StepVerifier.create(rule.crs.requestResponse(DefaultPayload.create("hello")))
@@ -413,6 +443,9 @@ protected void init() {
       LocalDuplexConnection clientConnection =
           new LocalDuplexConnection("client", allocator, serverProcessor, clientProcessor);
 
+      clientConnection.onClose().doFinally(__ -> serverConnection.dispose()).subscribe();
+      serverConnection.onClose().doFinally(__ -> clientConnection.dispose()).subscribe();
+
       requestAcceptor =
           null != requestAcceptor
               ? requestAcceptor