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

@@ -433,10 +433,8 @@ protected void hookOnError(Throwable throwable) {
 
           @Override
           protected void hookOnComplete() {
-            if (isEmpty) {
-              if (sendingSubscriptions.remove(streamId, this)) {
-                sendProcessor.onNext(PayloadFrameCodec.encodeComplete(allocator, streamId));
-              }
+            if (sendingSubscriptions.remove(streamId, this) && isEmpty) {
+              sendProcessor.onNext(PayloadFrameCodec.encodeComplete(allocator, streamId));
             }
           }
         };