provides tests and some minor fixes to UnicastMonoProcessor

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

Author: OlegDokuka

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

@@ -45,8 +45,6 @@ public static <T> UnicastMonoProcessor<T> create() {
   static final AtomicIntegerFieldUpdater<UnicastMonoProcessor> ONCE =
       AtomicIntegerFieldUpdater.newUpdater(UnicastMonoProcessor.class, "once");
 
-  Publisher<? extends O> source;
-
   Throwable error;
   volatile boolean terminated;
   O value;
@@ -67,7 +65,6 @@ public final void cancel() {
       return;
     }
 
-    source = null;
     if (s != null) {
       s.cancel();
     }
@@ -84,7 +81,6 @@ public void dispose() {
     final CancellationException e = new CancellationException("Disposed");
     error = e;
     value = null;
-    source = null;
     terminated = true;
     if (s != null) {
       s.cancel();
@@ -159,7 +155,6 @@ public final void onError(Throwable cause) {
 
     error = cause;
     value = null;
-    source = null;
     terminated = true;
 
     final CoreSubscriber<? super O> a = actual;
@@ -182,8 +177,6 @@ public final void onNext(@Nullable O value) {
     }
 
     this.value = value;
-    final Publisher<? extends O> parent = source;
-    source = null;
     terminated = true;
 
     final CoreSubscriber<? super O> a = actual;
@@ -193,7 +186,7 @@ public final void onNext(@Nullable O value) {
         a.onComplete();
       }
     } else {
-      if (s != null && !(parent instanceof Mono)) {
+      if (s != null) {
         s.cancel();
       }