wip

Signed-off-by: Oleh Dokuka <oleh.dokuka@icloud.com>
Signed-off-by: OlegDokuka <odokuka@vmware.com>

build.gradle

@@ -33,9 +33,9 @@ subprojects {
     apply plugin: 'com.github.sherter.google-java-format'
     apply plugin: 'com.github.vlsi.gradle-extensions'
 
-    ext['reactor-bom.version'] = '2020.0.32'
+    ext['reactor-bom.version'] = '2020.0.39'
     ext['logback.version'] = '1.2.10'
-    ext['netty-bom.version'] = '4.1.93.Final'
+    ext['netty-bom.version'] = '4.1.106.Final'
     ext['netty-boringssl.version'] = '2.0.61.Final'
     ext['hdrhistogram.version'] = '2.1.12'
     ext['mockito.version'] = '4.11.0'

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

@@ -463,8 +463,14 @@ private Mono<Void> acceptSetup(
           return interceptors
               .initSocketAcceptor(acceptor)
               .accept(setupPayload, wrappedRSocketRequester)
-              .doOnError(
-                  err -> serverSetup.sendError(wrappedDuplexConnection, rejectedSetupError(err)))
+              .onErrorResume(
+                  err ->
+                      Mono.fromRunnable(
+                              () ->
+                                  serverSetup.sendError(
+                                      wrappedDuplexConnection, rejectedSetupError(err)))
+                          .then(wrappedDuplexConnection.onClose())
+                          .then(Mono.error(err)))
               .doOnNext(
                   rSocketHandler -> {
                     RSocket wrappedRSocketHandler = interceptors.initResponder(rSocketHandler);

rsocket-core/src/main/java/io/rsocket/frame/FrameUtil.java

@@ -33,11 +33,31 @@ public static String toString(ByteBuf frame) {
     if (FrameHeaderCodec.hasMetadata(frame)) {
       payload.append("\nMetadata:\n");
 
-      ByteBufUtil.appendPrettyHexDump(payload, getMetadata(frame, frameType));
+      ByteBuf metadata = getMetadata(frame, frameType);
+      if (metadata.readableBytes() < 100) {
+        ByteBufUtil.appendPrettyHexDump(payload, metadata);
+      } else {
+        payload.append(
+            "  +-------------------------------------------------+\n"
+                + "         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |\n"
+                + "+--------+-------------------------------------------------+----------------+\n"
+                + "|00000000| too large payload                               |                |\n"
+                + "+--------+-------------------------------------------------+----------------+\n");
+      }
     }
 
     payload.append("\nData:\n");
-    ByteBufUtil.appendPrettyHexDump(payload, getData(frame, frameType));
+    ByteBuf data = getData(frame, frameType);
+    if (data.readableBytes() < 100) {
+      ByteBufUtil.appendPrettyHexDump(payload, data);
+    } else {
+      payload.append(
+          "  +-------------------------------------------------+\n"
+              + "         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |\n"
+              + "+--------+-------------------------------------------------+----------------+\n"
+              + "|00000000| too large payload                               |                |\n"
+              + "+--------+-------------------------------------------------+----------------+\n");
+    }
 
     return payload.toString();
   }

rsocket-core/src/main/java/io/rsocket/resume/ServerRSocketSession.java

@@ -287,6 +287,9 @@ public void setKeepAliveSupport(KeepAliveSupport keepAliveSupport) {
 
   @Override
   public void dispose() {
+    if (logger.isDebugEnabled()) {
+      logger.debug("Side[server]|Session[{}]. Disposing session", session);
+    }
     Operators.terminate(S, this);
     resumableConnection.dispose();
   }

rsocket-transport-netty/src/test/java/io/rsocket/integration/AuthenticationTest.java

@@ -0,0 +1,78 @@
+package io.rsocket.integration;
+
+import io.rsocket.Payload;
+import io.rsocket.RSocket;
+import io.rsocket.core.RSocketConnector;
+import io.rsocket.core.RSocketServer;
+import io.rsocket.exceptions.RejectedSetupException;
+import io.rsocket.transport.netty.client.TcpClientTransport;
+import io.rsocket.transport.netty.server.CloseableChannel;
+import io.rsocket.transport.netty.server.TcpServerTransport;
+import io.rsocket.util.DefaultPayload;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Hooks;
+import reactor.core.publisher.Mono;
+import reactor.netty.tcp.TcpClient;
+import reactor.netty.tcp.TcpServer;
+import reactor.test.StepVerifier;
+
+public class AuthenticationTest {
+
+  private static final Logger LOG = LoggerFactory.getLogger(AuthenticationTest.class);
+  private static final int PORT = 23200;
+
+  @Test
+  void authTest() {
+    Hooks.onOperatorDebug();
+    createServer().block();
+    RSocket rsocketClient = createClient().block();
+
+    StepVerifier.create(rsocketClient.requestResponse(DefaultPayload.create("Client: Hello")))
+        .expectError(RejectedSetupException.class)
+        .verify();
+  }
+
+  private static Mono<CloseableChannel> createServer() {
+    LOG.info("Starting server at port {}", PORT);
+    RSocketServer rSocketServer =
+        RSocketServer.create((connectionSetupPayload, rSocket) -> Mono.just(new MyServerRsocket()));
+
+    TcpServer tcpServer = TcpServer.create().host("localhost").port(PORT);
+
+    return rSocketServer
+        .interceptors(
+            interceptorRegistry ->
+                interceptorRegistry.forSocketAcceptor(
+                    socketAcceptor ->
+                        (setup, sendingSocket) -> {
+                          if (true) { // TODO here would be an authentication check based on the
+                                      // setup payload
+                            return Mono.error(new RejectedSetupException("ACCESS_DENIED"));
+                          } else {
+                            return socketAcceptor.accept(setup, sendingSocket);
+                          }
+                        }))
+        .bind(TcpServerTransport.create(tcpServer))
+        .doOnNext(closeableChannel -> LOG.info("RSocket server started."));
+  }
+
+  private static Mono<RSocket> createClient() {
+    LOG.info("Connecting....");
+    return RSocketConnector.create()
+        .connect(TcpClientTransport.create(TcpClient.create().host("localhost").port(PORT)))
+        .doOnNext(rSocket -> LOG.info("Successfully connected to server"))
+        .doOnError(throwable -> LOG.error("Failed to connect to server"));
+  }
+
+  public static class MyServerRsocket implements RSocket {
+    private static final Logger LOG = LoggerFactory.getLogger(MyServerRsocket.class);
+
+    @Override
+    public Mono<Payload> requestResponse(Payload payload) {
+      LOG.info("Got a request with payload: {}", payload.getDataUtf8());
+      return Mono.just("Response data blah blah blah").map(DefaultPayload::create);
+    }
+  }
+}

rsocket-transport-netty/src/test/resources/logback-test.xml

@@ -23,11 +23,26 @@
         </encoder>
     </appender>
 
+    <timestamp key="bySecond" datePattern="yyyyMMdd'T'HHmmss"/>
+
+    <appender name="FILE" class="ch.qos.logback.core.FileAppender">
+        <!-- use the previously created timestamp to create a uniquely
+             named log file -->
+        <file>log-${bySecond}.txt</file>
+        <encoder>
+            <pattern>%date{HH:mm:ss.SSS} %-10thread %-42logger %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <appender name="ASYNC" class="ch.qos.logback.classic.AsyncAppender">
+        <appender-ref ref="FILE" />
+    </appender>
+
     <logger name="io.rsocket.transport.netty" level="INFO"/>
-    <logger name="io.rsocket.FrameLogger" level="INFO"/>
+    <logger name="io.rsocket.FrameLogger" level="DEBUG" additivity="false">
+        <appender-ref ref="ASYNC" />
+    </logger>
     <logger name="io.rsocket.fragmentation.FragmentationDuplexConnection" level="INFO"/>
-    <logger name="io.rsocket.transport.netty" level="INFO"/>
-    <logger name="io.rsocket.FrameLogger" level="INFO"/>
     <logger name="io.rsocket.core.RSocketRequester" level="DEBUG"/>
     <logger name="io.rsocket.core.RSocketResponder" level="DEBUG"/>
     <logger name="io.rsocket.test.TransportTest" level="DEBUG"/>
@@ -36,8 +51,9 @@
     <logger name="io.rsocket.resume.ResumableDuplexConnection" level="DEBUG"/>
     <logger name="io.rsocket.resume.InMemoryResumableFramesStore" level="DEBUG"/>
 
-    <root level="INFO">
+    <root level="DEBUG">
         <appender-ref ref="STDOUT"/>
+        <appender-ref ref="ASYNC"/>
     </root>
 
 </configuration>