wip

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

Author: OlegDokuka

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

@@ -177,7 +177,7 @@ public boolean remove(int streamId, FrameHandler frameHandler) {
         return false;
       }
       activeStreams.remove(streamId);
-      if (activeStreams.size() == 0) {
+      if (this.terminating && activeStreams.size() == 0) {
         terminated = true;
         this.terminated = true;
       } else {

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

@@ -491,10 +491,6 @@ public Mono<Void> onClose() {
 
     onCloseDelayer.tryEmitEmpty();
 
-    onCloseSubscriber.assertNotTerminated();
-
-    rule.otherClosedSink.tryEmitEmpty();
-
     onCloseSubscriber.assertTerminated().assertComplete();
 
     Assertions.assertThat(rule.socket.isDisposed()).isTrue();
@@ -516,10 +512,6 @@ public void shouldResolveOnStartSource() {
 
     rule.client.onClose().subscribe(assertSubscriber1);
 
-    assertSubscriber1.assertNotTerminated();
-
-    rule.otherClosedSink.tryEmitEmpty();
-
     assertSubscriber1.assertTerminated().assertComplete();
 
     Assertions.assertThat(rule.socket.isDisposed()).isTrue();
@@ -579,8 +571,6 @@ public void shouldBeRestartedIfSourceWasClosed() {
 
     rule.client.dispose();
 
-    rule.otherClosedSink.tryEmitEmpty();
-
     terminateSubscriber.assertTerminated().assertComplete();
 
     Assertions.assertThat(rule.client.connect()).isFalse();
@@ -636,8 +626,6 @@ public void shouldStartOriginalSourceOnceIfRacing() {
 
       rule.client.dispose();
 
-      rule.otherClosedSink.tryEmitEmpty();
-
       Assertions.assertThat(rule.client.isDisposed()).isTrue();
       Assertions.assertThat(rule.socket.isDisposed()).isTrue();
 
@@ -656,10 +644,8 @@ public static class ClientSocketRule extends AbstractSocketRule<RSocket> {
     protected Sinks.One<RSocket> producer;
 
     protected Sinks.Empty<Void> onGracefulShutdownStartedSink;
-    protected Sinks.Empty<Void> otherGracefulShutdownSink;
     protected Sinks.Empty<Void> thisGracefulShutdownSink;
     protected Sinks.Empty<Void> thisClosedSink;
-    protected Sinks.Empty<Void> otherClosedSink;
 
     @Override
     protected void doInit() {
@@ -679,10 +665,8 @@ protected void doInit() {
     @Override
     protected RSocket newRSocket() {
       this.onGracefulShutdownStartedSink = Sinks.empty();
-      this.otherGracefulShutdownSink = Sinks.empty();
       this.thisGracefulShutdownSink = Sinks.empty();
       this.thisClosedSink = Sinks.empty();
-      this.otherClosedSink = Sinks.empty();
       return new RSocketRequester(
           connection,
           PayloadDecoder.ZERO_COPY,
@@ -698,8 +682,8 @@ protected RSocket newRSocket() {
           onGracefulShutdownStartedSink,
           thisGracefulShutdownSink,
           thisClosedSink,
-          otherGracefulShutdownSink.asMono().and(thisGracefulShutdownSink.asMono()),
-          otherClosedSink.asMono().and(thisClosedSink.asMono()));
+          thisGracefulShutdownSink.asMono(),
+          thisClosedSink.asMono());
     }
   }
 }