introduces disposable channels

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

Author: OlegDokuka

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

@@ -27,13 +27,16 @@
 import io.netty.util.IllegalReferenceCountException;
 import io.rsocket.DuplexConnection;
 import io.rsocket.Payload;
+import io.rsocket.exceptions.CanceledException;
 import io.rsocket.frame.CancelFrameCodec;
 import io.rsocket.frame.FrameType;
 import io.rsocket.frame.decoder.PayloadDecoder;
 import io.rsocket.plugins.RequestInterceptor;
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
+import reactor.core.Disposable;
+import reactor.core.Disposables;
 import reactor.core.Exceptions;
 import reactor.core.Scannable;
 import reactor.core.publisher.Mono;
@@ -42,7 +45,7 @@
 import reactor.util.annotation.Nullable;
 
 final class RequestResponseRequesterMono extends Mono<Payload>
-    implements RequesterFrameHandler, LeasePermitHandler, Subscription, Scannable {
+    implements RequesterFrameHandler, LeasePermitHandler, Subscription, Scannable, Disposable {
 
   final ByteBufAllocator allocator;
   final Payload payload;
@@ -255,6 +258,32 @@ public final void cancel() {
     }
   }
 
+  @Override
+  public void dispose() {
+    long previousState = markTerminated(STATE, this);
+    if (isTerminated(previousState)) {
+      return;
+    }
+
+    if (isFirstFrameSent(previousState)) {
+      final int streamId = this.streamId;
+      this.requesterResponderSupport.remove(streamId, this);
+
+      ReassemblyUtils.synchronizedRelease(this, previousState);
+
+      this.connection.sendFrame(streamId, CancelFrameCodec.encode(this.allocator, streamId));
+
+      final RequestInterceptor requestInterceptor = this.requestInterceptor;
+      if (requestInterceptor != null) {
+        requestInterceptor.onCancel(streamId, FrameType.REQUEST_RESPONSE);
+      }
+    } else if (!isReadyToSendFirstFrame(previousState)) {
+      this.payload.release();
+    }
+
+    this.actual.onError(new CanceledException("Forcefully disposed"));
+  }
+
   @Override
   public final void handlePayload(Payload value) {
     if (this.done) {

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

@@ -38,13 +38,14 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import reactor.core.CoreSubscriber;
+import reactor.core.Disposable;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Operators;
 import reactor.util.annotation.Nullable;
 import reactor.util.context.Context;
 
 final class RequestResponseResponderSubscriber
-    implements ResponderFrameHandler, CoreSubscriber<Payload> {
+    implements ResponderFrameHandler, CoreSubscriber<Payload>, Disposable {
 
   static final Logger logger = LoggerFactory.getLogger(RequestResponseResponderSubscriber.class);
 

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

@@ -9,6 +9,9 @@
 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;
 
 class RequesterResponderSupport {
@@ -19,12 +22,14 @@ class RequesterResponderSupport {
   private final PayloadDecoder payloadDecoder;
   private final ByteBufAllocator allocator;
   private final DuplexConnection connection;
+  private final Sinks.Empty<Void> onAllStreamsTerminatedSink;
   @Nullable private final RequestInterceptor requestInterceptor;
 
   @Nullable final StreamIdSupplier streamIdSupplier;
   final IntObjectMap<FrameHandler> activeStreams;
 
   boolean terminating;
+  boolean terminated;
 
   public RequesterResponderSupport(
       int mtu,
@@ -43,6 +48,7 @@ public RequesterResponderSupport(
     this.allocator = connection.alloc();
     this.streamIdSupplier = streamIdSupplier;
     this.connection = connection;
+    this.onAllStreamsTerminatedSink = Sinks.empty();
     this.requestInterceptor = requestInterceptorFunction.apply((RSocket) this);
   }
 
@@ -122,8 +128,8 @@ public int addAndGetNextStreamId(FrameHandler frameHandler) {
   }
 
   public synchronized boolean add(int streamId, FrameHandler frameHandler) {
-    if (this.terminating) {
-      throw
+    if (this.terminated) {
+         throwing
     }
     final IntObjectMap<FrameHandler> activeStreams = this.activeStreams;
     // copy of Map.putIfAbsent(key, value) without `streamId` boxing
@@ -165,7 +171,27 @@ public synchronized boolean remove(int streamId, FrameHandler frameHandler) {
     return true;
   }
 
-  public synchronized void terminate() {
-    this.terminating = true;
+  public void terminate() {
+    final boolean terminated;
+    synchronized (this) {
+      this.terminating = true;
+
+      if (activeStreams.size() == 0) {
+        terminated = true;
+        this.terminated = true;
+      } else {
+        terminated = false;
+      }
+    }
+
+    if (terminated) {
+      onAllStreamsTerminatedSink.tryEmitEmpty();
+    } else {
+      
+    }
+  }
+
+  public Mono<Void> onAllStreamsTerminated() {
+    return onAllStreamsTerminatedSink.asMono();
   }
 }