fixes removal ordering

Signed-off-by: Oleh Dokuka <odokuka@vmware.com>
Signed-off-by: Oleh Dokuka <oleh.dokuka@icloud.com>
Signed-off-by: OlegDokuka <odokuka@vmware.com>

Author: Oleh Dokuka

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

@@ -238,11 +238,11 @@ void sendFirstPayload(Payload payload, long initialRequestN) {
         return;
       }
 
-      sm.remove(streamId, this);
-
       final ByteBuf cancelFrame = CancelFrameCodec.encode(allocator, streamId);
       connection.sendFrame(streamId, cancelFrame);
 
+      sm.remove(streamId, this);
+
       if (requestInterceptor != null) {
         requestInterceptor.onCancel(streamId, FrameType.REQUEST_STREAM);
       }
@@ -276,12 +276,13 @@ public final void cancel() {
 
     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));
 
+      this.requesterResponderSupport.remove(streamId, this);
+
       final RequestInterceptor requestInterceptor = this.requestInterceptor;
       if (requestInterceptor != null) {
         requestInterceptor.onCancel(streamId, FrameType.REQUEST_STREAM);
@@ -309,13 +310,14 @@ public final void handlePayload(Payload p) {
       }
 
       final int streamId = this.streamId;
-      this.requesterResponderSupport.remove(streamId, this);
 
       final IllegalStateException cause =
           Exceptions.failWithOverflow(
               "The number of messages received exceeds the number requested");
       this.connection.sendFrame(streamId, CancelFrameCodec.encode(this.allocator, streamId));
 
+      this.requesterResponderSupport.remove(streamId, this);
+
       final RequestInterceptor requestInterceptor = this.requestInterceptor;
       if (requestInterceptor != null) {
         requestInterceptor.onTerminate(streamId, FrameType.REQUEST_STREAM, cause);

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

@@ -144,6 +144,8 @@ public void onNext(Payload p) {
         final ByteBuf errorFrame = ErrorFrameCodec.encode(allocator, streamId, e);
         sender.sendFrame(streamId, errorFrame);
 
+        this.requesterResponderSupport.remove(streamId, this);
+
         final RequestInterceptor requestInterceptor = this.requestInterceptor;
         if (requestInterceptor != null) {
           requestInterceptor.onTerminate(streamId, FrameType.REQUEST_STREAM, e);
@@ -162,6 +164,8 @@ public void onNext(Payload p) {
               new CanceledException("Failed to validate payload. Cause" + e.getMessage()));
       sender.sendFrame(streamId, errorFrame);
 
+      this.requesterResponderSupport.remove(streamId, this);
+
       final RequestInterceptor requestInterceptor = this.requestInterceptor;
       if (requestInterceptor != null) {
         requestInterceptor.onTerminate(streamId, FrameType.REQUEST_STREAM, e);
@@ -176,6 +180,8 @@ public void onNext(Payload p) {
         return;
       }
 
+      this.requesterResponderSupport.remove(streamId, this);
+
       final RequestInterceptor requestInterceptor = this.requestInterceptor;
       if (requestInterceptor != null) {
         requestInterceptor.onTerminate(streamId, FrameType.REQUEST_STREAM, t);
@@ -195,8 +201,6 @@ boolean tryTerminateOnError() {
       return false;
     }
 
-    this.requesterResponderSupport.remove(this.streamId, this);
-
     currentSubscription.cancel();
 
     return true;
@@ -222,11 +226,12 @@ public void onError(Throwable t) {
     }
 
     final int streamId = this.streamId;
-    this.requesterResponderSupport.remove(streamId, this);
 
     final ByteBuf errorFrame = ErrorFrameCodec.encode(this.allocator, streamId, t);
     this.connection.sendFrame(streamId, errorFrame);
 
+    this.requesterResponderSupport.remove(streamId, this);
+
     final RequestInterceptor requestInterceptor = this.requestInterceptor;
     if (requestInterceptor != null) {
       requestInterceptor.onTerminate(streamId, FrameType.REQUEST_STREAM, t);
@@ -246,11 +251,12 @@ public void onComplete() {
     }
 
     final int streamId = this.streamId;
-    this.requesterResponderSupport.remove(streamId, this);
 
     final ByteBuf completeFrame = PayloadFrameCodec.encodeComplete(this.allocator, streamId);
     this.connection.sendFrame(streamId, completeFrame);
 
+    this.requesterResponderSupport.remove(streamId, this);
+
     final RequestInterceptor requestInterceptor = this.requestInterceptor;
     if (requestInterceptor != null) {
       requestInterceptor.onTerminate(streamId, FrameType.REQUEST_STREAM, null);
@@ -321,7 +327,6 @@ public void handleNext(ByteBuf followingFrame, boolean hasFollows, boolean isLas
       S.lazySet(this, Operators.cancelledSubscription());
 
       final int streamId = this.streamId;
-      this.requesterResponderSupport.remove(streamId, this);
 
       this.frames = null;
       frames.release();
@@ -334,6 +339,8 @@ public void handleNext(ByteBuf followingFrame, boolean hasFollows, boolean isLas
               new CanceledException("Failed to reassemble payload. Cause: " + e.getMessage()));
       this.connection.sendFrame(streamId, errorFrame);
 
+      this.requesterResponderSupport.remove(streamId, this);
+
       final RequestInterceptor requestInterceptor = this.requestInterceptor;
       if (requestInterceptor != null) {
         requestInterceptor.onTerminate(streamId, FrameType.REQUEST_STREAM, e);
@@ -354,7 +361,6 @@ public void handleNext(ByteBuf followingFrame, boolean hasFollows, boolean isLas
         this.done = true;
 
         final int streamId = this.streamId;
-        this.requesterResponderSupport.remove(streamId, this);
 
         ReferenceCountUtil.safeRelease(frames);
 
@@ -366,6 +372,8 @@ public void handleNext(ByteBuf followingFrame, boolean hasFollows, boolean isLas
                 new CanceledException("Failed to reassemble payload. Cause: " + t.getMessage()));
         this.connection.sendFrame(streamId, errorFrame);
 
+        this.requesterResponderSupport.remove(streamId, this);
+
         final RequestInterceptor requestInterceptor = this.requestInterceptor;
         if (requestInterceptor != null) {
           requestInterceptor.onTerminate(streamId, FrameType.REQUEST_STREAM, t);