Ensure Subscriber is removed from sendingSubscriptions

Closes gh-961

Signed-off-by: Rossen Stoyanchev <[email protected]>

Author: rstoyanchev

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

@@ -406,22 +406,20 @@ private void handleRequestResponse(int streamId, Mono<Payload> response) {
 
           @Override
           protected void hookOnNext(Payload payload) {
-            if (isEmpty) {
-              isEmpty = false;
-            }
+            isEmpty = false;
 
             if (!PayloadValidationUtils.isValid(mtu, payload, maxFrameLength)) {
               payload.release();
               cancel();
-              final IllegalArgumentException t =
-                  new IllegalArgumentException(INVALID_PAYLOAD_ERROR_MESSAGE);
-              handleError(streamId, t);
+              sendingSubscriptions.remove(streamId, this);
+              handleError(streamId, new IllegalArgumentException(INVALID_PAYLOAD_ERROR_MESSAGE));
               return;
             }
 
             ByteBuf byteBuf =
                 PayloadFrameCodec.encodeNextCompleteReleasingPayload(allocator, streamId, payload);
             sendProcessor.onNext(byteBuf);
+            sendingSubscriptions.remove(streamId, this);
           }
 
           @Override
@@ -433,10 +431,8 @@ protected void hookOnError(Throwable throwable) {
 
           @Override
           protected void hookOnComplete() {
-            if (isEmpty) {
-              if (sendingSubscriptions.remove(streamId, this)) {
-                sendProcessor.onNext(PayloadFrameCodec.encodeComplete(allocator, streamId));
-              }
+            if (isEmpty && sendingSubscriptions.remove(streamId, this)) {
+              sendProcessor.onNext(PayloadFrameCodec.encodeComplete(allocator, streamId));
             }
           }
         };