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,7 +85,11 @@ default double availability() {
   }
 
   @Override
-  default void dispose() {}
+  default void dispose() {
+    dispose(true);
+  }
+
+  default void dispose(boolean force) {}
 
   @Override
   default boolean isDisposed() {

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

@@ -24,6 +24,8 @@ class RequesterResponderSupport {
   @Nullable final StreamIdSupplier streamIdSupplier;
   final IntObjectMap<FrameHandler> activeStreams;
 
+  boolean terminating;
+
   public RequesterResponderSupport(
       int mtu,
       int maxFrameLength,
@@ -88,6 +90,7 @@ public int getNextStreamId() {
     final StreamIdSupplier streamIdSupplier = this.streamIdSupplier;
     if (streamIdSupplier != null) {
       synchronized (this) {
+        if
         return streamIdSupplier.nextStreamId(this.activeStreams);
       }
     } else {
@@ -119,6 +122,9 @@ public int addAndGetNextStreamId(FrameHandler frameHandler) {
   }
 
   public synchronized boolean add(int streamId, FrameHandler frameHandler) {
+    if (this.terminating) {
+      throw
+    }
     final IntObjectMap<FrameHandler> activeStreams = this.activeStreams;
     // copy of Map.putIfAbsent(key, value) without `streamId` boxing
     final FrameHandler previousHandler = activeStreams.get(streamId);
@@ -158,4 +164,8 @@ public synchronized boolean remove(int streamId, FrameHandler frameHandler) {
     activeStreams.remove(streamId);
     return true;
   }
+
+  public synchronized void terminate() {
+    this.terminating = true;
+  }
 }

rsocket-transport-local/src/test/java/io/rsocket/transport/local/LocalResumableTransportTest.java

@@ -16,12 +16,28 @@
 
 package io.rsocket.transport.local;
 
+import io.netty.buffer.ByteBufUtil;
+import io.netty.buffer.Unpooled;
+import io.rsocket.frame.FrameHeaderCodec;
+import io.rsocket.frame.FrameType;
 import io.rsocket.test.TransportTest;
 import java.time.Duration;
 import java.util.UUID;
 
 final class LocalResumableTransportTest implements TransportTest {
 
+  public static void main(String[] args) {
+
+    System.out.println(
+        FrameHeaderCodec.frameType(
+            Unpooled.copiedBuffer(
+                ByteBufUtil.decodeHexDump("000000003800000000000007ef4a"))));
+    System.out.println(
+        FrameHeaderCodec.frameType(
+            Unpooled.copiedBuffer(
+                ByteBufUtil.decodeHexDump("000000002c0000000004"))));
+  }
+
   private final TransportPair transportPair =
       new TransportPair<>(
           () -> "test-" + UUID.randomUUID(),