enhancement/quic-http3-transports

README.md

@@ -31,9 +31,16 @@ repositories {
 dependencies {
     implementation 'io.rsocket:rsocket-core:1.2.0-SNAPSHOT'
     implementation 'io.rsocket:rsocket-transport-netty:1.2.0-SNAPSHOT'
+    implementation 'io.rsocket:rsocket-transport-quic:1.2.0-SNAPSHOT' // optional QUIC transport
+    implementation 'io.rsocket:rsocket-transport-h3:1.2.0-SNAPSHOT' // optional HTTP/3 transport
 }
 ```
 
+For a minimal HTTP/3 `"PING" -> "PONG"` example, see
+[Http3PingPongIntegrationTest.java](rsocket-transport-h3/src/test/java/io/rsocket/transport/netty/Http3PingPongIntegrationTest.java)
+and the transport notes in
+[QUIC_HTTP3_TRANSPORTS.md](rsocket-transport-h3/QUIC_HTTP3_TRANSPORTS.md).
+
 Snapshots are available via [oss.jfrog.org](oss.jfrog.org) (OJO).
 
 Example:
@@ -46,6 +53,8 @@ repositories {
 dependencies {
     implementation 'io.rsocket:rsocket-core:1.2.0-SNAPSHOT'
     implementation 'io.rsocket:rsocket-transport-netty:1.2.0-SNAPSHOT'
+    implementation 'io.rsocket:rsocket-transport-quic:1.2.0-SNAPSHOT' // optional QUIC transport
+    implementation 'io.rsocket:rsocket-transport-h3:1.2.0-SNAPSHOT' // optional HTTP/3 transport
 }
 ```
 

build.gradle

@@ -17,12 +17,14 @@
 plugins {
     id 'com.github.sherter.google-java-format' version '0.9' apply false
     id 'me.champeau.jmh' version '0.7.1' apply false
-    id 'io.spring.dependency-management' version '1.1.0' apply false
+    id 'io.spring.dependency-management' version '1.1.7' apply false
     id 'io.morethan.jmhreport' version '0.9.0' apply false
     id 'io.github.reyerizo.gradle.jcstress' version '0.8.15' apply false
     id 'com.github.vlsi.gradle-extensions'  version '1.89' apply false
 }
 
+def compositeConsumer = System.getProperty("rsocket.java.composite.consumer")
+
 boolean isCiServer = ["CI", "CONTINUOUS_INTEGRATION", "TRAVIS", "CIRCLECI", "bamboo_planKey", "GITHUB_ACTION"].with {
     retainAll(System.getenv().keySet())
     return !isEmpty()
@@ -226,12 +228,12 @@ subprojects {
 
     plugins.withType(JavaLibraryPlugin) {
         task sourcesJar(type: Jar) {
-            classifier 'sources'
+            archiveClassifier.set('sources')
             from sourceSets.main.allJava
         }
 
         task javadocJar(type: Jar, dependsOn: javadoc) {
-            classifier 'javadoc'
+            archiveClassifier.set('javadoc')
             from javadoc.destinationDir
         }
 
@@ -249,7 +251,9 @@ subprojects {
     }
 }
 
-apply from: "${rootDir}/gradle/publications.gradle"
+if (compositeConsumer == null) {
+    apply from: "${rootDir}/gradle/publications.gradle"
+}
 
 buildScan {
     termsOfServiceUrl = 'https://gradle.com/terms-of-service'

rsocket-core/build.gradle

@@ -18,9 +18,15 @@ plugins {
     id 'java-library'
     id 'maven-publish'
     id 'signing'
-    id 'io.morethan.jmhreport'
-    id 'me.champeau.jmh'
-    id 'io.github.reyerizo.gradle.jcstress'
+}
+
+def compositeConsumer = System.getProperty("rsocket.java.composite.consumer")
+def includePerformancePlugins = compositeConsumer == null
+
+if (includePerformancePlugins) {
+    apply plugin: 'io.morethan.jmhreport'
+    apply plugin: 'me.champeau.jmh'
+    apply plugin: 'io.github.reyerizo.gradle.jcstress'
 }
 
 dependencies {
@@ -40,15 +46,19 @@ dependencies {
     testRuntimeOnly 'ch.qos.logback:logback-classic'
     testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine'
 
-    jcstressImplementation(project(":rsocket-test"))
-    jcstressImplementation 'org.slf4j:slf4j-api'
-    jcstressImplementation "ch.qos.logback:logback-classic"
-    jcstressImplementation 'io.projectreactor:reactor-test'
+    if (includePerformancePlugins) {
+        jcstressImplementation(project(":rsocket-test"))
+        jcstressImplementation 'org.slf4j:slf4j-api'
+        jcstressImplementation "ch.qos.logback:logback-classic"
+        jcstressImplementation 'io.projectreactor:reactor-test'
+    }
 }
 
-jcstress {
-    mode = 'sanity' //sanity, quick, default, tough
-    jcstressDependency = "org.openjdk.jcstress:jcstress-core:0.16"
+if (includePerformancePlugins) {
+    jcstress {
+        mode = 'sanity' //sanity, quick, default, tough
+        jcstressDependency = "org.openjdk.jcstress:jcstress-core:0.16"
+    }
 }
 
 jar {

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

@@ -6,6 +6,8 @@
 import io.rsocket.RSocketErrorException;
 import java.net.SocketAddress;
 import java.nio.channels.ClosedChannelException;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
 import reactor.core.publisher.Flux;
@@ -24,8 +26,10 @@ class SetupHandlingDuplexConnection extends Flux<ByteBuf>
 
   Subscription s;
   boolean firstFrameReceived = false;
+  boolean subscriberReady = false;
 
-  CoreSubscriber<? super ByteBuf> actual;
+  volatile CoreSubscriber<? super ByteBuf> actual;
+  final Queue<ByteBuf> pendingFrames = new ConcurrentLinkedQueue<>();
 
   boolean done;
   Throwable t;
@@ -82,6 +86,7 @@ public void subscribe(CoreSubscriber<? super ByteBuf> actual) {
 
     this.actual = actual;
     actual.onSubscribe(this);
+    drainPendingFrames(actual);
   }
 
   @Override
@@ -91,7 +96,8 @@ public void request(long n) {
       return;
     }
 
-    s.request(Long.MAX_VALUE);
+    subscriberReady = true;
+    drainPendingFrames(actual);
   }
 
   @Override
@@ -104,7 +110,7 @@ public void cancel() {
   public void onSubscribe(Subscription s) {
     if (Operators.validate(this.s, s)) {
       this.s = s;
-      s.request(1);
+      s.request(Long.MAX_VALUE);
     }
   }
 
@@ -116,7 +122,12 @@ public void onNext(ByteBuf frame) {
       return;
     }
 
-    actual.onNext(frame);
+    final CoreSubscriber<? super ByteBuf> actual = this.actual;
+    if (actual != null && subscriberReady) {
+      actual.onNext(frame);
+    } else {
+      pendingFrames.offer(frame);
+    }
   }
 
   @Override
@@ -173,4 +184,15 @@ public ByteBufAllocator alloc() {
   public String toString() {
     return "SetupHandlingDuplexConnection{" + "source=" + source + ", done=" + done + '}';
   }
+
+  private void drainPendingFrames(CoreSubscriber<? super ByteBuf> actual) {
+    if (actual == null) {
+      return;
+    }
+
+    ByteBuf frame;
+    while ((frame = pendingFrames.poll()) != null) {
+      actual.onNext(frame);
+    }
+  }
 }