polishes code (#967)

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

Author: OlegDokuka

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

@@ -385,13 +385,14 @@ private Mono<Void> acceptSetup(
           clientServerConnection,
           new InvalidSetupException(
               "Unsupported version: " + SetupFrameCodec.humanReadableVersion(setupFrame)));
+      return clientServerConnection.onClose();
     }
 
     boolean leaseEnabled = leasesSupplier != null;
     if (SetupFrameCodec.honorLease(setupFrame) && !leaseEnabled) {
       serverSetup.sendError(
           clientServerConnection, new InvalidSetupException("lease is not supported"));
-      return Mono.empty();
+      return clientServerConnection.onClose();
     }
 
     return serverSetup.acceptRSocketSetup(

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

@@ -65,7 +65,7 @@ public Mono<Void> acceptRSocketSetup(
 
       if (SetupFrameCodec.resumeEnabled(frame)) {
         sendError(duplexConnection, new UnsupportedSetupException("resume not supported"));
-        return Mono.empty();
+        return duplexConnection.onClose();
       } else {
         return then.apply(new DefaultKeepAliveHandler(duplexConnection), duplexConnection);
       }

rsocket-examples/src/main/java/io/rsocket/examples/transport/tcp/stream/ClientStreamingToServer.java

@@ -16,13 +16,12 @@
 
 package io.rsocket.examples.transport.tcp.stream;
 
-import io.rsocket.Payload;
 import io.rsocket.RSocket;
 import io.rsocket.SocketAcceptor;
 import io.rsocket.core.RSocketConnector;
 import io.rsocket.core.RSocketServer;
 import io.rsocket.transport.netty.client.TcpClientTransport;
-import io.rsocket.transport.netty.server.TcpServerTransport;
+import io.rsocket.transport.netty.server.WebsocketServerTransport;
 import io.rsocket.util.DefaultPayload;
 import java.time.Duration;
 import org.slf4j.Logger;
@@ -33,26 +32,28 @@ public final class ClientStreamingToServer {
 
   private static final Logger logger = LoggerFactory.getLogger(ClientStreamingToServer.class);
 
-  public static void main(String[] args) {
+  public static void main(String[] args) throws InterruptedException {
     RSocketServer.create(
             SocketAcceptor.forRequestStream(
                 payload ->
                     Flux.interval(Duration.ofMillis(100))
                         .map(aLong -> DefaultPayload.create("Interval: " + aLong))))
-        .bind(TcpServerTransport.create("localhost", 7000))
+        .bind(WebsocketServerTransport.create("localhost", 7000))
         .subscribe();
 
     RSocket socket =
         RSocketConnector.connectWith(TcpClientTransport.create("localhost", 7000)).block();
 
-    socket
-        .requestStream(DefaultPayload.create("Hello"))
-        .map(Payload::getDataUtf8)
-        .doOnNext(logger::debug)
-        .take(10)
-        .then()
-        .doFinally(signalType -> socket.dispose())
-        .then()
-        .block();
+    //    socket
+    //        .requestStream(DefaultPayload.create("Hello"))
+    //        .map(Payload::getDataUtf8)
+    //        .doOnNext(logger::debug)
+    //        .take(10)
+    //        .then()
+    //        .doFinally(signalType -> socket.dispose())
+    //        .then()
+    //        .block();
+
+    Thread.sleep(1000000);
   }
 }

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

@@ -80,7 +80,6 @@ public LeaksTrackingByteBufAllocator assertHasNoLeaks() {
           }
 
           if (!hasUnreleased) {
-            System.out.println(tag + " all the buffers are released...");
             return this;
           }
 
@@ -109,7 +108,6 @@ public LeaksTrackingByteBufAllocator assertHasNoLeaks() {
                 return checkResult;
               },
               tag);
-      System.out.println(tag + " all the buffers are released...");
     } finally {
       tracker.clear();
     }