fixes

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

Author: OlegDokuka

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

@@ -374,7 +374,7 @@ private void tryShutdown() {
   }
 
   private void terminate(Throwable e) {
-    LOGGER.info("closing requester");
+    LOGGER.info("closing requester " + getDuplexConnection());
     if (keepAliveFramesAcceptor != null) {
       keepAliveFramesAcceptor.dispose();
     }

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

@@ -186,7 +186,7 @@ public Mono<Void> onClose() {
   }
 
   final void doOnDispose() {
-    LOGGER.info("closing responder");
+    LOGGER.info("closing responder " + getDuplexConnection());
     cleanUpSendingSubscriptions();
 
     getDuplexConnection().dispose();

rsocket-test/src/main/java/io/rsocket/test/TransportTest.java

@@ -21,11 +21,7 @@
 import io.netty.util.ReferenceCountUtil;
 import io.netty.util.ReferenceCounted;
 import io.netty.util.ResourceLeakDetector;
-import io.rsocket.Closeable;
-import io.rsocket.DuplexConnection;
-import io.rsocket.Payload;
-import io.rsocket.RSocket;
-import io.rsocket.RSocketErrorException;
+import io.rsocket.*;
 import io.rsocket.core.RSocketConnector;
 import io.rsocket.core.RSocketServer;
 import io.rsocket.core.Resume;
@@ -55,15 +51,11 @@
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 import org.reactivestreams.Subscription;
-import reactor.core.CoreSubscriber;
-import reactor.core.Disposable;
-import reactor.core.Disposables;
-import reactor.core.Exceptions;
-import reactor.core.Fuseable;
+import reactor.core.*;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
-import reactor.core.publisher.MonoProcessor;
 import reactor.core.publisher.Operators;
+import reactor.core.publisher.Sinks;
 import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 import reactor.test.StepVerifier;
@@ -625,8 +617,8 @@ public TransportPair(
 
     @Override
     public void dispose() {
-      server.dispose();
       client.dispose();
+      server.dispose();
     }
 
     RSocket getClient() {
@@ -794,12 +786,12 @@ private static class ByteBufReleaserOperator
         implements CoreSubscriber<ByteBuf>, Subscription, Fuseable.QueueSubscription<ByteBuf> {
 
       CoreSubscriber<? super ByteBuf> actual;
-      final MonoProcessor<Void> closeableMono;
+      final Sinks.Empty<Void> closeableMonoSink;
 
       Subscription s;
 
       public ByteBufReleaserOperator() {
-        this.closeableMono = MonoProcessor.create();
+        this.closeableMonoSink = Sinks.unsafe().empty();
       }
 
       @Override
@@ -820,19 +812,19 @@ public void onNext(ByteBuf buf) {
       }
 
       Mono<Void> onClose() {
-        return closeableMono;
+        return closeableMonoSink.asMono();
       }
 
       @Override
       public void onError(Throwable t) {
         actual.onError(t);
-        closeableMono.onError(t);
+        closeableMonoSink.tryEmitError(t);
       }
 
       @Override
       public void onComplete() {
         actual.onComplete();
-        closeableMono.onComplete();
+        closeableMonoSink.tryEmitEmpty();
       }
 
       @Override
@@ -843,7 +835,7 @@ public void request(long n) {
       @Override
       public void cancel() {
         s.cancel();
-        closeableMono.onComplete();
+        closeableMonoSink.tryEmitEmpty();
       }
 
       @Override