fixes concurrent queue consumption

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

Author: OlegDokuka

rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java

@@ -352,6 +352,10 @@ public void cancel() {
     }
     cancelled = true;
 
+    if (outputFused) {
+      return;
+    }
+
     if (WIP.getAndIncrement(this) == 0) {
       this.clear();
       hasDownstream = false;
@@ -418,6 +422,7 @@ public int requestFusion(int requestedMode) {
 
   @Override
   public void dispose() {
+    super.dispose();
     cancel();
   }