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)'