wip

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

Author: OlegDokuka

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

@@ -85,13 +85,9 @@ default double availability() {
   }
 
   @Override
-  default void dispose() {
+  default void dispose() {}
 
-  }
-
-  default void disposeGracefully() {
-
-  }
+  default void disposeGracefully() {}
 
   @Override
   default boolean isDisposed() {

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

@@ -47,6 +47,7 @@
 import java.util.function.Supplier;
 import reactor.core.Disposable;
 import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
 import reactor.util.annotation.Nullable;
 import reactor.util.function.Tuples;
 import reactor.util.retry.Retry;
@@ -655,6 +656,9 @@ public Mono<RSocket> connect(Supplier<ClientTransport> transportSupplier) {
                                     requesterLeaseTracker = null;
                                   }
 
+                                  Sinks.Empty<Void> requesterSink = Sinks.empty();
+                                  Sinks.Empty<Void> responderSink = Sinks.empty();
+
                                   RSocket rSocketRequester =
                                       new RSocketRequester(
                                           multiplexer.asClientConnection(),
@@ -667,7 +671,9 @@ public Mono<RSocket> connect(Supplier<ClientTransport> transportSupplier) {
                                           (int) keepAliveMaxLifeTime.toMillis(),
                                           keepAliveHandler,
                                           interceptors::initRequesterRequestInterceptor,
-                                          requesterLeaseTracker);
+                                          requesterLeaseTracker,
+                                          requesterSink,
+                                          requesterSink.asMono().and(responderSink.asMono()));
 
                                   RSocket wrappedRSocketRequester =
                                       interceptors.initRequester(rSocketRequester);

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

@@ -66,8 +66,8 @@ class RSocketRequester extends RequesterResponderSupport implements RSocket {
           RSocketRequester.class, Throwable.class, "terminationError");
 
   @Nullable private final RequesterLeaseTracker requesterLeaseTracker;
+  private Mono<Void> onAllClose;
   private final KeepAliveFramesAcceptor keepAliveFramesAcceptor;
-  private final Sinks.Empty<Void> onClose;
 
   RSocketRequester(
       DuplexConnection connection,
@@ -80,18 +80,21 @@ class RSocketRequester extends RequesterResponderSupport implements RSocket {
       int keepAliveAckTimeout,
       @Nullable KeepAliveHandler keepAliveHandler,
       Function<RSocket, RequestInterceptor> requestInterceptorFunction,
-      @Nullable RequesterLeaseTracker requesterLeaseTracker) {
+      @Nullable RequesterLeaseTracker requesterLeaseTracker,
+      Sinks.Empty<Void> onClose,
+      Mono<Void> onAllClose) {
     super(
         mtu,
         maxFrameLength,
         maxInboundPayloadSize,
         payloadDecoder,
         connection,
         streamIdSupplier,
-        requestInterceptorFunction);
+        requestInterceptorFunction,
+        onClose);
 
     this.requesterLeaseTracker = requesterLeaseTracker;
-    this.onClose = Sinks.empty();
+    this.onAllClose = onAllClose;
 
     // DO NOT Change the order here. The Send processor must be subscribed to before receiving
     connection.onClose().subscribe(null, this::tryTerminateOnConnectionError, this::tryShutdown);
@@ -188,13 +191,12 @@ public double availability() {
 
   @Override
   public void dispose() {
-    tryShutdown();
+    getDuplexConnection().sendErrorAndClose(new ConnectionErrorException("Disposed"));
   }
 
   @Override
   public void disposeGracefully() {
     super.terminate();
-    super.onAllStreamsTerminated()
   }
 
   @Override
@@ -204,7 +206,7 @@ public boolean isDisposed() {
 
   @Override
   public Mono<Void> onClose() {
-    return onClose.asMono();
+    return onAllClose;
   }
 
   private void handleIncomingFrames(ByteBuf frame) {
@@ -340,7 +342,6 @@ private void terminate(Throwable e) {
     if (keepAliveFramesAcceptor != null) {
       keepAliveFramesAcceptor.dispose();
     }
-    getDuplexConnection().dispose();
     final RequestInterceptor requestInterceptor = getRequestInterceptor();
     if (requestInterceptor != null) {
       requestInterceptor.dispose();

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

@@ -10,8 +10,6 @@
 import io.rsocket.plugins.RequestInterceptor;
 import java.util.Objects;
 import java.util.function.Function;
-
-import reactor.core.publisher.Mono;
 import reactor.core.publisher.Sinks;
 import reactor.util.annotation.Nullable;
 
@@ -23,7 +21,7 @@ class RequesterResponderSupport {
   private final PayloadDecoder payloadDecoder;
   private final ByteBufAllocator allocator;
   private final DuplexConnection connection;
-  private final Sinks.Empty<Void> onAllStreamsTerminatedSink;
+  private final Sinks.Empty<Void> onClose;
   @Nullable private final RequestInterceptor requestInterceptor;
 
   @Nullable final StreamIdSupplier streamIdSupplier;
@@ -39,7 +37,8 @@ public RequesterResponderSupport(
       PayloadDecoder payloadDecoder,
       DuplexConnection connection,
       @Nullable StreamIdSupplier streamIdSupplier,
-      Function<RSocket, ? extends RequestInterceptor> requestInterceptorFunction) {
+      Function<RSocket, ? extends RequestInterceptor> requestInterceptorFunction,
+      Sinks.Empty<Void> onClose) {
 
     this.activeStreams = new IntObjectHashMap<>();
     this.mtu = mtu;
@@ -49,7 +48,7 @@ public RequesterResponderSupport(
     this.allocator = connection.alloc();
     this.streamIdSupplier = streamIdSupplier;
     this.connection = connection;
-    this.onAllStreamsTerminatedSink = Sinks.empty();
+    this.onClose = onClose;
     this.requestInterceptor = requestInterceptorFunction.apply((RSocket) this);
   }
 
@@ -187,7 +186,7 @@ public boolean remove(int streamId, FrameHandler frameHandler) {
     }
 
     if (terminated) {
-      onAllStreamsTerminatedSink.tryEmitEmpty();
+      onClose.tryEmitEmpty();
     }
     return true;
   }
@@ -206,11 +205,7 @@ public void terminate() {
     }
 
     if (terminated) {
-      onAllStreamsTerminatedSink.tryEmitEmpty();
+      onClose.tryEmitEmpty();
     }
   }
-
-  public Mono<Void> onAllStreamsTerminated() {
-    return onAllStreamsTerminatedSink.asMono();
-  }
 }