Split QUIC and HTTP/3 transports into dedicated artifacts

Author: veltmanj

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
 }
 ```
 

rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java

@@ -18,6 +18,7 @@
 
 import io.netty.buffer.ByteBuf;
 import io.rsocket.internal.jctools.queues.MpscUnboundedArrayQueue;
+import java.lang.reflect.Method;
 import java.util.Objects;
 import java.util.Queue;
 import java.util.concurrent.CancellationException;
@@ -97,6 +98,8 @@ public final class UnboundedProcessor extends Flux<ByteBuf>
 
   boolean outputFused;
 
+  @Nullable private static final Method THREAD_ID_METHOD = resolveThreadIdMethod();
+
   public UnboundedProcessor() {
     this(() -> {});
   }
@@ -141,7 +144,7 @@ public boolean tryEmitPrioritized(ByteBuf t) {
     }
 
     if (!this.priorityQueue.offer(t)) {
-      onError(Operators.onOperatorError(null, Exceptions.failWithOverflow(), t, currentContext()));
+      tryEmitError(Operators.onOperatorError(null, Exceptions.failWithOverflow(), t, currentContext()));
       release(t);
       return false;
     }
@@ -177,7 +180,7 @@ public boolean tryEmitNormal(ByteBuf t) {
     }
 
     if (!this.queue.offer(t)) {
-      onError(Operators.onOperatorError(null, Exceptions.failWithOverflow(), t, currentContext()));
+      tryEmitError(Operators.onOperatorError(null, Exceptions.failWithOverflow(), t, currentContext()));
       release(t);
       return false;
     }
@@ -207,126 +210,140 @@ public boolean tryEmitNormal(ByteBuf t) {
     return true;
   }
 
-  public boolean tryEmitFinal(ByteBuf t) {
+  public boolean tryEmitError(Throwable t) {
+    Objects.requireNonNull(t, "throwable must not be null");
+
     if (this.done || this.cancelled) {
-      release(t);
+      Operators.onErrorDropped(t, currentContext());
       return false;
     }
 
-    this.last = t;
+    this.error = t;
     this.done = true;
 
-    final long previousState = markValueAddedAndTerminated(this);
-    if (isFinalized(previousState)) {
-      this.clearSafely();
+    final long previousState = markTerminatedOrFinalized(this);
+    if (isFinalized(previousState)
+        || isDisposed(previousState)
+        || isCancelled(previousState)
+        || isTerminated(previousState)) {
+      Operators.onErrorDropped(t, currentContext());
       return false;
     }
 
     if (isSubscriberReady(previousState)) {
       if (this.outputFused) {
-        // fast path for fusion
-        this.actual.onNext(null);
-        this.actual.onComplete();
+        // fast path for fusion scenario
+        this.actual.onError(t);
         return true;
       }
 
       if (isWorkInProgress(previousState)) {
         return true;
       }
 
-      drainRegular((previousState | FLAG_TERMINATED | FLAG_HAS_VALUE) + 1);
+      if (!hasValue(previousState)) {
+        // fast path no-values scenario
+        this.actual.onError(t);
+        return true;
+      }
+
+      drainRegular((previousState | FLAG_TERMINATED) + 1);
     }
 
     return true;
   }
 
-  @Deprecated
-  public void onNextPrioritized(ByteBuf t) {
-    tryEmitPrioritized(t);
-  }
-
-  @Override
-  @Deprecated
-  public void onNext(ByteBuf t) {
-    tryEmitNormal(t);
-  }
-
-  @Override
-  @Deprecated
-  public void onError(Throwable t) {
+  public boolean tryEmitComplete() {
     if (this.done || this.cancelled) {
-      Operators.onErrorDropped(t, currentContext());
-      return;
+      return false;
     }
 
-    this.error = t;
     this.done = true;
 
     final long previousState = markTerminatedOrFinalized(this);
     if (isFinalized(previousState)
         || isDisposed(previousState)
         || isCancelled(previousState)
         || isTerminated(previousState)) {
-      Operators.onErrorDropped(t, currentContext());
-      return;
+      return false;
     }
 
     if (isSubscriberReady(previousState)) {
       if (this.outputFused) {
         // fast path for fusion scenario
-        this.actual.onError(t);
-        return;
+        this.actual.onComplete();
+        return true;
       }
 
       if (isWorkInProgress(previousState)) {
-        return;
+        return true;
       }
 
       if (!hasValue(previousState)) {
-        // fast path no-values scenario
-        this.actual.onError(t);
-        return;
+        this.actual.onComplete();
+        return true;
       }
 
       drainRegular((previousState | FLAG_TERMINATED) + 1);
     }
+
+    return true;
   }
 
-  @Override
-  @Deprecated
-  public void onComplete() {
+  public boolean tryEmitFinal(ByteBuf t) {
     if (this.done || this.cancelled) {
-      return;
+      release(t);
+      return false;
     }
 
+    this.last = t;
     this.done = true;
 
-    final long previousState = markTerminatedOrFinalized(this);
-    if (isFinalized(previousState)
-        || isDisposed(previousState)
-        || isCancelled(previousState)
-        || isTerminated(previousState)) {
-      return;
+    final long previousState = markValueAddedAndTerminated(this);
+    if (isFinalized(previousState)) {
+      this.clearSafely();
+      return false;
     }
 
     if (isSubscriberReady(previousState)) {
       if (this.outputFused) {
-        // fast path for fusion scenario
+        // fast path for fusion
+        this.actual.onNext(null);
         this.actual.onComplete();
-        return;
+        return true;
       }
 
       if (isWorkInProgress(previousState)) {
-        return;
-      }
-
-      if (!hasValue(previousState)) {
-        this.actual.onComplete();
-        return;
+        return true;
       }
 
-      drainRegular((previousState | FLAG_TERMINATED) + 1);
+      drainRegular((previousState | FLAG_TERMINATED | FLAG_HAS_VALUE) + 1);
     }
+
+    return true;
+  }
+
+  @Deprecated
+  public void onNextPrioritized(ByteBuf t) {
+    tryEmitPrioritized(t);
+  }
+
+  @Override
+  @Deprecated
+  public void onNext(ByteBuf t) {
+    tryEmitNormal(t);
+  }
+
+  @Override
+  @Deprecated
+  public void onError(Throwable t) {
+    tryEmitError(t);
+  }
+
+  @Override
+  @Deprecated
+  public void onComplete() {
+    tryEmitComplete();
   }
 
   void drainRegular(long expectedState) {
@@ -1091,7 +1108,7 @@ static void log(
               instance,
               action,
               action,
-              Thread.currentThread().getId(),
+              currentThreadId(),
               formatState(initialState, 64),
               formatState(committedState, 64)),
           new RuntimeException());
@@ -1101,7 +1118,7 @@ static void log(
               "[%s][%s][%s][%s-%s]",
               instance,
               action,
-              Thread.currentThread().getId(),
+              currentThreadId(),
               formatState(initialState, 64),
               formatState(committedState, 64)));
     }
@@ -1130,7 +1147,7 @@ static void log(
               instance,
               action,
               action,
-              Thread.currentThread().getId(),
+              currentThreadId(),
               formatState(initialState, 32),
               formatState(committedState, 32)),
           new RuntimeException());
@@ -1140,12 +1157,39 @@ static void log(
               "[%s][%s][%s][%s-%s]",
               instance,
               action,
-              Thread.currentThread().getId(),
+              currentThreadId(),
               formatState(initialState, 32),
               formatState(committedState, 32)));
     }
   }
 
+  private static long currentThreadId() {
+    Thread currentThread = Thread.currentThread();
+    if (THREAD_ID_METHOD != null) {
+      try {
+        return (long) THREAD_ID_METHOD.invoke(currentThread);
+      } catch (Throwable ignore) {
+        // Fallback to the legacy method below for older runtimes or reflective failures.
+      }
+    }
+
+    return legacyThreadId(currentThread);
+  }
+
+  @Nullable
+  private static Method resolveThreadIdMethod() {
+    try {
+      return Thread.class.getMethod("threadId");
+    } catch (NoSuchMethodException e) {
+      return null;
+    }
+  }
+
+  @SuppressWarnings("deprecation")
+  private static long legacyThreadId(Thread thread) {
+    return thread.getId();
+  }
+
   static String formatState(long state, int size) {
     final String defaultFormat = Long.toBinaryString(state);
     final StringBuilder formatted = new StringBuilder();

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

@@ -241,7 +241,7 @@ public void dispose() {
     if (activeConnection == DisposedConnection.INSTANCE) {
       return;
     }
-    savableFramesSender.onComplete();
+    savableFramesSender.tryEmitComplete();
     activeConnection
         .onClose()
         .subscribe(
@@ -262,7 +262,7 @@ void dispose(DuplexConnection nextConnection, @Nullable Throwable e) {
     if (activeConnection == DisposedConnection.INSTANCE) {
       return;
     }
-    savableFramesSender.onComplete();
+    savableFramesSender.tryEmitComplete();
     nextConnection
         .onClose()
         .subscribe(

…transport-netty/QUIC_HTTP3_TRANSPORTS.md …et-transport-h3/QUIC_HTTP3_TRANSPORTS.mdrsocket-transport-netty/QUIC_HTTP3_TRANSPORTS.md renamed to rsocket-transport-h3/QUIC_HTTP3_TRANSPORTS.md rsocket-transport-netty/QUIC_HTTP3_TRANSPORTS.md renamed to rsocket-transport-h3/QUIC_HTTP3_TRANSPORTS.md

@@ -1,10 +1,11 @@
 # QUIC and HTTP/3 transport notes
 
-This module now contains prototype Netty transports for:
+This module contains the prototype Netty HTTP/3 transport for:
 
-- `quic://` via `QuicClientTransport` and `QuicServerTransport`
 - `h3://` via `Http3ClientTransport` and `Http3ServerTransport`
 
+The shared QUIC transport now lives in the sibling `rsocket-transport-quic` artifact.
+
 ## Current HTTP/3 transport contract
 
 The current `h3://` transport uses a long-lived HTTP/3 `POST` request stream:
@@ -57,6 +58,7 @@ Examples:
 ```bash
 ./scripts/http3-pingpong.sh integration
 ./scripts/http3-pingpong.sh pong-server
+./scripts/http3-pingpong.sh ping-pong
 ./scripts/http3-pingpong.sh request-response 1000
 ./scripts/http3-pingpong.sh request-stream 100
 ./scripts/http3-pingpong.sh all
@@ -65,6 +67,7 @@ Examples:
 The `all` command starts the HTTP/3 pong server in the background, waits for the UDP port to bind,
 then runs:
 
+- `Http3PingPongIntegrationTest`
 - `Http3TransportIntegrationTest`
 - `Http3Ping.requestResponseTest`
 - `Http3Ping.requestStreamTest`
@@ -77,3 +80,25 @@ Useful environment overrides:
 - `RSOCKET_HTTP3_PATH`
 - `RSOCKET_HTTP3_REQUEST_RESPONSE_INTERACTIONS`
 - `RSOCKET_HTTP3_REQUEST_STREAM_INTERACTIONS`
+
+## Minimal ping-pong example
+
+For the clearest end-to-end HTTP/3 example in the repository, see:
+
+- [Http3PingPongIntegrationTest.java](src/test/java/io/rsocket/transport/netty/Http3PingPongIntegrationTest.java)
+
+That test keeps the scenario intentionally small:
+
+- starts an `Http3ServerTransport`
+- sends `"PING"` from the client with `requestResponse`
+- asserts the server receives `"PING"`
+- returns `"PONG"`
+- asserts the client receives `"PONG"`
+
+You can run just that test with:
+
+```bash
+JAVA_HOME=$(/usr/libexec/java_home -v 13) ./gradlew --no-daemon --console=plain \
+  :rsocket-transport-h3:test \
+  --tests 'io.rsocket.transport.netty.Http3PingPongIntegrationTest'
+```