Netty Transport Improvements (#507)

This change makes general improvements to the Netty Transport.  It updates the
code to make it better documented and tested.  In addition it introduces the
new TransportTest, a JUnit 5 way of testing uniformly across all of the
transport implementations.

Author: nebhale

build.gradle

@@ -17,7 +17,6 @@
 plugins {
     id 'com.gradle.build-scan' version '1.12.1'
 
-    id 'com.github.johnrengelman.shadow' version '2.0.2' apply false
     id 'com.github.sherter.google-java-format' version '0.6'
     id 'com.jfrog.artifactory' version '4.7.0'
     id 'com.jfrog.bintray' version '1.8.0'

rsocket-core/build.gradle

@@ -17,7 +17,6 @@
 plugins {
     id 'java-library'
     id 'maven-publish'
-    id 'com.github.johnrengelman.shadow'
     id 'com.jfrog.artifactory'
     id 'com.jfrog.bintray'
     id 'io.morethan.jmhreport'
@@ -29,10 +28,10 @@ dependencies {
     api 'io.projectreactor:reactor-core'
 
     implementation 'io.projectreactor.addons:reactor-extra'
+    implementation 'org.jctools:jctools-core'
     implementation 'org.slf4j:slf4j-api'
 
     compileOnly 'com.google.code.findbugs:jsr305'
-    compileOnly 'org.jctools:jctools-core'
 
     testImplementation 'io.projectreactor:reactor-test'
     testImplementation 'org.assertj:assertj-core'
@@ -49,28 +48,6 @@ dependencies {
     testRuntimeOnly 'org.junit.vintage:junit-vintage-engine'
 }
 
-jar {
-    enabled = false
-    dependsOn(shadowJar { classifier = null })
-}
-
-shadowJar {
-    configurations = [project.configurations.compileOnly]
-
-    dependencies {
-        include(dependency('org.jctools:jctools-core'))
-    }
-
-    include '*.jar'
-    include 'io/rsocket/**'
-    include 'org/jctools/maps/ConcurrentAutoTable*.class'
-    include 'org/jctools/maps/NonBlockingHashMapLong*.class'
-    include 'org/jctools/util/RangeUtil*.class'
-    include 'org/jctools/util/UnsafeAccess*.class'
-
-    relocate 'org.jctools', 'io.rsocket.shadowed.org.jctools'
-}
-
 description = "Core functionality for the RSocket library"
 
 apply from: 'jmh.gradle'

rsocket-core/src/jmh/java/io/rsocket/fragmentation/FragmentationPerformanceTest.java

@@ -57,7 +57,7 @@ public void largeFragmentation(Input input) {
 
   @Benchmark
   public void largeReassembly(Input input) {
-    input.largeFrames.forEach(frame -> input.reassembler.reassemble(frame, input.sink));
+    input.largeFrames.forEach(frame -> input.reassembler.reassemble(frame));
 
     input.bh.consume(input.sink.next);
   }
@@ -72,7 +72,7 @@ public void smallFragmentation(Input input) {
 
   @Benchmark
   public void smallReassembly(Input input) {
-    input.smallFrames.forEach(frame -> input.reassembler.reassemble(frame, input.sink));
+    input.smallFrames.forEach(frame -> input.reassembler.reassemble(frame));
 
     input.bh.consume(input.sink.next);
   }

rsocket-core/src/main/java/io/rsocket/transport/ServerTransport.java

@@ -30,6 +30,7 @@ public interface ServerTransport<T extends Closeable> extends Transport {
    *
    * @param acceptor An acceptor to process a newly accepted {@code DuplexConnection}
    * @return A handle to retrieve information about a started server.
+   * @throws NullPointerException if {@code acceptor} is {@code null}
    */
   Mono<T> start(ConnectionAcceptor acceptor);
 

rsocket-core/src/main/java/io/rsocket/transport/TransportHeaderAware.java

@@ -24,5 +24,12 @@
  * Http2.
  */
 public interface TransportHeaderAware {
+
+  /**
+   * Sets the transport headers
+   *
+   * @param transportHeaders the transport headers
+   * @throws NullPointerException if {@code transportHeaders} is {@code null}
+   */
   void setTransportHeaders(Supplier<Map<String, String>> transportHeaders);
 }

rsocket-test/build.gradle

@@ -16,16 +16,20 @@
 
 plugins {
     id 'java-library'
+    id 'maven-publish'
+    id 'com.jfrog.artifactory'
+    id 'com.jfrog.bintray'
 }
 
 dependencies {
+    api project(':rsocket-core')
     api 'org.hdrhistogram:HdrHistogram'
+    api 'org.junit.jupiter:junit-jupiter-api'
+
+    compileOnly  'com.google.code.findbugs:jsr305'
 
-    implementation project(':rsocket-core')
-    implementation 'com.google.code.findbugs:jsr305'
     implementation 'io.projectreactor:reactor-test'
     implementation 'org.assertj:assertj-core'
-    implementation 'org.junit.jupiter:junit-jupiter-api'
     implementation 'org.mockito:mockito-core'
 
     // TODO: Remove after JUnit5 migration

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

@@ -24,6 +24,9 @@
 import io.rsocket.transport.ServerTransport;
 import io.rsocket.util.DefaultPayload;
 import java.time.Duration;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import java.util.function.Supplier;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
@@ -258,21 +261,32 @@ default void requestStreamDelayedRequestN() {
         .verify(getTimeout());
   }
 
-  final class TransportPair implements Disposable {
+  final class TransportPair<T, S extends Closeable> implements Disposable {
 
     private final RSocket client;
 
-    private final Closeable server;
+    private final S server;
 
-    public TransportPair(ServerTransport<?> serverTransport, ClientTransport clientTransport) {
-      this.server =
+    public TransportPair(
+        Supplier<T> addressSupplier,
+        BiFunction<T, S, ClientTransport> clientTransportSupplier,
+        Function<T, ServerTransport<S>> serverTransportSupplier) {
+
+      T address = addressSupplier.get();
+
+      server =
           RSocketFactory.receive()
               .acceptor((setup, sendingSocket) -> Mono.just(new TestRSocket()))
-              .transport(serverTransport)
+              .transport(serverTransportSupplier.apply(address))
               .start()
               .block();
 
-      this.client = RSocketFactory.connect().transport(clientTransport).start().block();
+      client =
+          RSocketFactory.connect()
+              .transport(clientTransportSupplier.apply(address, server))
+              .start()
+              .doOnError(Throwable::printStackTrace)
+              .block();
     }
 
     @Override

rsocket-transport-local/src/test/java/io/rsocket/transport/local/LocalClientTransportTest.java

@@ -33,12 +33,7 @@ void connect() {
 
     serverTransport
         .start(duplexConnection -> Mono.empty())
-        .as(StepVerifier::create)
-        .expectNextCount(1)
-        .verifyComplete();
-
-    LocalClientTransport.create(serverTransport.getName())
-        .connect()
+        .flatMap(closeable -> LocalClientTransport.create(serverTransport.getName()).connect())
         .as(StepVerifier::create)
         .expectNextCount(1)
         .verifyComplete();

rsocket-transport-local/src/test/java/io/rsocket/transport/local/LocalTransportTest.java

@@ -18,14 +18,17 @@
 
 import io.rsocket.test.TransportTest;
 import java.time.Duration;
+import java.util.concurrent.atomic.AtomicInteger;
 
 final class LocalTransportTest implements TransportTest {
 
-  private final LocalServerTransport serverTransport = LocalServerTransport.createEphemeral();
+  private static final AtomicInteger UNIQUE_NAME_GENERATOR = new AtomicInteger();
 
-  private final LocalClientTransport clientTransport = serverTransport.clientTransport();
-
-  private final TransportPair transportPair = new TransportPair(serverTransport, clientTransport);
+  private final TransportPair transportPair =
+      new TransportPair<>(
+          () -> "test" + UNIQUE_NAME_GENERATOR.incrementAndGet(),
+          (address, server) -> LocalClientTransport.create(address),
+          LocalServerTransport::create);
 
   @Override
   public Duration getTimeout() {

rsocket-transport-netty/build.gradle

@@ -25,12 +25,15 @@ dependencies {
     api project(':rsocket-core')
     api 'io.projectreactor.ipc:reactor-netty'
 
+    compileOnly 'com.google.code.findbugs:jsr305'
+
     testImplementation project(':rsocket-test')
+    testImplementation 'io.projectreactor:reactor-test'
+    testImplementation 'org.assertj:assertj-core'
     testImplementation 'org.junit.jupiter:junit-jupiter-api'
 
-    // TODO: Remove after JUnit5 migration
-    testCompileOnly 'junit:junit'
-    testRuntimeOnly 'org.junit.vintage:junit-vintage-engine'
+    testRuntimeOnly 'ch.qos.logback:logback-classic'
+    testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine'
 }
 
 description = 'Reactor Netty RSocket transport implementations (TCP, Websocket)'

rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/RSocketLengthCodec.java

@@ -23,7 +23,13 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
 
-public class RSocketLengthCodec extends LengthFieldBasedFrameDecoder {
+/**
+ * An extension to the Netty {@link LengthFieldBasedFrameDecoder} that encapsulates the
+ * RSocket-specific frame length header details.
+ */
+public final class RSocketLengthCodec extends LengthFieldBasedFrameDecoder {
+
+  /** Creates a new instance of the decoder, specifying the RSocket frame length header size. */
   public RSocketLengthCodec() {
     super(FRAME_LENGTH_MASK, 0, FRAME_LENGTH_SIZE, 0, 0);
   }

rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/NettyDuplexConnection.java renamed to rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/TcpDuplexConnection.java

@@ -25,44 +25,55 @@
 import reactor.ipc.netty.NettyInbound;
 import reactor.ipc.netty.NettyOutbound;
 
-public class NettyDuplexConnection implements DuplexConnection {
+/** An implementation of {@link DuplexConnection} that connects via TCP. */
+public final class TcpDuplexConnection implements DuplexConnection {
+
+  private final NettyContext context;
+
   private final NettyInbound in;
+
   private final NettyOutbound out;
-  private final NettyContext context;
 
-  public NettyDuplexConnection(NettyInbound in, NettyOutbound out, NettyContext context) {
+  /**
+   * Creates a new instance
+   *
+   * @param in the {@link NettyInbound} to listen on
+   * @param out the {@link NettyOutbound} to send with
+   * @param context the {@link NettyContext} to for managing the server
+   */
+  public TcpDuplexConnection(NettyInbound in, NettyOutbound out, NettyContext context) {
     this.in = in;
     this.out = out;
     this.context = context;
   }
 
   @Override
-  public Mono<Void> send(Publisher<Frame> frames) {
-    return Flux.from(frames).concatMap(this::sendOne).then();
+  public void dispose() {
+    context.dispose();
   }
 
   @Override
-  public Mono<Void> sendOne(Frame frame) {
-    return out.sendObject(frame.content()).then();
+  public boolean isDisposed() {
+    return context.isDisposed();
   }
 
   @Override
-  public Flux<Frame> receive() {
-    return in.receive().map(buf -> Frame.from(buf.retain()));
+  public Mono<Void> onClose() {
+    return context.onClose();
   }
 
   @Override
-  public void dispose() {
-    context.dispose();
+  public Flux<Frame> receive() {
+    return in.receive().map(buf -> Frame.from(buf.retain()));
   }
 
   @Override
-  public boolean isDisposed() {
-    return context.isDisposed();
+  public Mono<Void> send(Publisher<Frame> frames) {
+    return Flux.from(frames).concatMap(this::sendOne).then();
   }
 
   @Override
-  public Mono<Void> onClose() {
-    return context.onClose();
+  public Mono<Void> sendOne(Frame frame) {
+    return out.sendObject(frame.content()).then();
   }
 }

rsocket-transport-netty/src/main/java/io/rsocket/transport/netty/UriUtils.java

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2015-2018 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.rsocket.transport.netty;
+
+import java.net.URI;
+import java.util.Objects;
+
+/** Utilities for dealing with with {@link URI}s */
+public final class UriUtils {
+
+  private UriUtils() {}
+
+  /**
+   * Returns the port of a URI. If the port is unset (i.e. {@code -1}) then returns the {@code
+   * defaultPort}.
+   *
+   * @param uri the URI to extract the port from
+   * @param defaultPort the default to use if the port is unset
+   * @return the port of a URI or {@code defaultPort} if unset
+   * @throws NullPointerException if {@code uri} is {@code null}
+   */
+  public static int getPort(URI uri, int defaultPort) {
+    Objects.requireNonNull(uri, "uri must not be null");
+    return uri.getPort() == -1 ? defaultPort : uri.getPort();
+  }
+
+  /**
+   * Returns whether the URI has a secure schema. Secure is defined as being either {@code wss} or
+   * {@code https}.
+   *
+   * @param uri the URI to examine
+   * @return whether the URI has a secure schema
+   * @throws NullPointerException if {@code uri} is {@code null}
+   */
+  public static boolean isSecure(URI uri) {
+    Objects.requireNonNull(uri, "uri must not be null");
+    return uri.getScheme().equals("wss") || uri.getScheme().equals("https");
+  }
+}