Add R1 transport and lifecycle foundations

changes.md

@@ -77,7 +77,25 @@ Changes log
           explicitly. Without a valid configuration, `CorsFilter` adds no CORS headers. Issue #1611.
         - CORS: `CorsFilter` now applies configuration changes made after the first request (the
           internal `CorsResponseHelper` is rebuilt by the setters). Issue #1611.
+        - STDIO connector construction fails fast without an explicit or uniquely registered
+          supporting helper class. Applications with multiple STDIO protocol helpers must
+          select one explicitly.
     - Enhancements
+        - Added the supported `org.restlet.util.StreamTransport` seam for protocol-neutral,
+          bounded synchronous byte I/O, serialized writes, borrowed-stream cancellation and
+          owned subprocess lifecycle with separate bounded stderr capture. No protocol
+          framing, default STDIO helper or SDK dependency is added to core.
+        - STDIO connector construction now requires an explicit or uniquely registered
+          supporting helper class, rather than implicitly selecting the first protocol helper.
+        - Added `Request.getDisconnectSignal()` and connector notification of abnormal
+          termination independently of entity I/O. Abort requests made before connector
+          registration are remembered; wrappers delegate lifecycle and copies remain independent.
+        - HTTP clients publish response completion before exposing body EOF to consumers,
+          so an empty terminal response is not exposed while its exchange still appears active.
+        - HTTP/1.1 observes idle disconnects after request-body consumption while preserving
+          ordered pipelining with bounded decoded lookahead (1024 messages / 1 MiB accounted
+          content/header storage). Overflow closes the connection; protocol upgrades retain
+          their existing handoff. HTTP/2 and HTTP/3 keep stream-level disconnect notification.
         - HTTP/2 and HTTP/3 clients deliver response headers without waiting for the first
           response body byte. This allows idle duplex and SSE responses to expose their
           headers before the peer sends a message.

org.restlet/src/main/java/org/restlet/Request.java

@@ -10,9 +10,13 @@
 
 import java.util.List;
 import java.util.Map.Entry;
+import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.CopyOnWriteArraySet;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.BooleanSupplier;
 import org.restlet.data.CacheDirective;
 import org.restlet.data.ChallengeResponse;
@@ -85,6 +89,12 @@ public static Request getCurrent() {
     /** Action used by an active client connector to abort this request. */
     private volatile BooleanSupplier abortAction;
 
+    private final AtomicBoolean abortRequested = new AtomicBoolean();
+
+    private final CompletableFuture<Throwable> disconnectSignal = new CompletableFuture<>();
+
+    private final AtomicBoolean disconnectReported = new AtomicBoolean();
+
     /** The trailing headers received with the request entity. */
     private volatile Series<Header> trailers;
 
@@ -313,9 +323,14 @@ private void copyClientInfo(final Request request) {
      * Ask the connector to attempt to abort the related network connection, for example immediately
      * closing the socket.
      *
+     * <p>An attempt made before connector registration is remembered and applied when an active
+     * action is installed. It still returns false until an active action accepts it. Use a new
+     * Request for a new exchange after cancellation.
+     *
      * @return True if the request was aborted.
      */
     public boolean abort() {
+        abortRequested.set(true);
         BooleanSupplier action = abortAction;
         return action != null && action.getAsBoolean();
     }
@@ -324,10 +339,41 @@ public boolean abort() {
      * Sets the action used to abort an active client request. Connector implementations configure
      * this while a request is in flight and clear it when the exchange ends.
      *
+     * <p>The action must be idempotent: concurrent registration and abort attempts can invoke it
+     * more than once.
+     *
      * @param action The abort action, or null when no exchange is active.
      */
     public void setAbortAction(BooleanSupplier action) {
         this.abortAction = action;
+        if (action != null && abortRequested.get()) {
+            action.getAsBoolean();
+        }
+    }
+
+    /**
+     * Returns a read-only completion stage for abnormal transport disconnection of this exchange.
+     * It completes once with the cause, even if subscribed after the disconnect, and does not
+     * complete on normal response completion. Connectors report failure independently of entity
+     * reads/writes, so an idle application can observe cancellation. Completion runs off the
+     * connector event loop; consumers should still use their own executor for expensive work. Not
+     * all connectors support this signal. Copies have independent lifecycle state.
+     */
+    public CompletionStage<Throwable> getDisconnectSignal() {
+        return disconnectSignal.minimalCompletionStage();
+    }
+
+    /**
+     * Reports abnormal disconnection from a connector. The first cause wins; normal completion must
+     * not call this method.
+     *
+     * @param cause The non-null transport failure.
+     */
+    public void notifyDisconnect(Throwable cause) {
+        Objects.requireNonNull(cause, "cause");
+        if (disconnectReported.compareAndSet(false, true)) {
+            Thread.startVirtualThread(() -> disconnectSignal.complete(cause));
+        }
     }
 
     /**

org.restlet/src/main/java/org/restlet/engine/Engine.java

@@ -476,6 +476,7 @@ public org.restlet.engine.connector.ConnectorHelper<Client> createHelper(
         if (client.getProtocols().isEmpty()) {
             return result;
         }
+        requireUniqueStdioHelper(client.getProtocols(), helperClass, getRegisteredClients());
 
         org.restlet.engine.connector.ConnectorHelper<Client> connector = null;
         for (final Iterator<org.restlet.engine.connector.ConnectorHelper<Client>> iter =
@@ -533,6 +534,7 @@ public ConnectorHelper<Server> createHelper(Server server, String helperClass) {
         if (server.getProtocols().isEmpty()) {
             return result;
         }
+        requireUniqueStdioHelper(server.getProtocols(), helperClass, getRegisteredServers());
 
         ConnectorHelper<Server> connector = null;
         for (final Iterator<ConnectorHelper<Server>> iter = getRegisteredServers().iterator();
@@ -575,6 +577,34 @@ public ConnectorHelper<Server> createHelper(Server server, String helperClass) {
         return result;
     }
 
+    private static void requireUniqueStdioHelper(
+            List<Protocol> protocols,
+            String helperClass,
+            List<? extends ConnectorHelper<?>> helpers) {
+        if (!protocols.contains(Protocol.STDIO)) {
+            return;
+        }
+        long matches =
+                helpers.stream()
+                        .filter(
+                                helper ->
+                                        new HashSet<>(helper.getProtocols()).containsAll(protocols))
+                        .filter(
+                                helper ->
+                                        helperClass == null
+                                                || helper.getClass()
+                                                        .getCanonicalName()
+                                                        .equals(helperClass))
+                        .map(Object::getClass)
+                        .distinct()
+                        .count();
+        if (matches != 1) {
+            throw new IllegalStateException(
+                    "STDIO requires one explicitly selected or uniquely registered connector helper; found "
+                            + matches);
+        }
+    }
+
     /**
      * Discovers the authenticator helpers and register the default helpers.
      *

org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java

@@ -320,10 +320,10 @@ private Status sendRequest(org.restlet.Request request) {
 
         boolean backgroundWriterStarted = false;
         try {
-            this.channel = helper.acquire(this);
             if (cancelled.get()) {
                 throw new IOException("Request cancelled by client");
             }
+            attachChannel(helper.acquire(this));
             backgroundWriterStarted = sendRequestBody(request, entity, connectorService);
             awaitHead();
             result = new Status(statusCode, reasonPhrase);
@@ -343,6 +343,28 @@ private Status sendRequest(org.restlet.Request request) {
         return result;
     }
 
+    /**
+     * Attaches the acquired channel, unless the exchange was cancelled during the acquisition, in
+     * which case the channel is released at once.
+     *
+     * @param ch The acquired channel.
+     * @throws IOException If the exchange was cancelled.
+     */
+    private void attachChannel(Channel ch) throws IOException {
+        boolean alreadyReleased;
+        lock.lock();
+        try {
+            this.channel = ch;
+            alreadyReleased = released;
+        } finally {
+            lock.unlock();
+        }
+        if (alreadyReleased) {
+            helper.release(ch, false);
+            throw new IOException("Request cancelled by client");
+        }
+    }
+
     private boolean sendRequestBody(
             org.restlet.Request request, Representation entity, ConnectorService connectorService)
             throws IOException {
@@ -387,13 +409,15 @@ private void finishRequestBody(
     }
 
     private Status handleConnectFailure(ConnectException cause) {
+        currentRequest.notifyDisconnect(cause);
         completeEntityRelease();
         helper.logFailure("Unable to connect to the remote HTTP server", cause);
         abort();
         return new Status(Status.CONNECTOR_ERROR_CONNECTION, cause);
     }
 
     private Status handleCommunicationFailure(IOException cause) {
+        currentRequest.notifyDisconnect(cause);
         completeEntityRelease();
         helper.logFailure(
                 "An error occurred during the communication with the remote HTTP server", cause);
@@ -402,6 +426,7 @@ private Status handleCommunicationFailure(IOException cause) {
     }
 
     private Status handleInternalFailure(RuntimeException cause) {
+        currentRequest.notifyDisconnect(cause);
         completeEntityRelease();
         helper.logFailure("An unexpected error occurred while sending the request", cause);
         abort();
@@ -819,10 +844,6 @@ void onResponseComplete(io.netty.handler.codec.http.HttpHeaders trailers) {
                 currentResponse.getTrailers().add(name, trailer.getValue());
             }
         }
-        ChannelInputStream stream = this.body;
-        if (stream != null) {
-            stream.end();
-        }
         Representation emptyEntity;
         lock.lock();
         try {
@@ -832,6 +853,10 @@ void onResponseComplete(io.netty.handler.codec.http.HttpHeaders trailers) {
         } finally {
             lock.unlock();
         }
+        ChannelInputStream stream = this.body;
+        if (stream != null) {
+            stream.end();
+        }
         if (emptyEntity != null) {
             emptyEntity.release();
         }
@@ -868,6 +893,10 @@ void onFailure(Throwable cause) {
             lock.unlock();
         }
         ChannelInputStream stream = this.body;
+        org.restlet.Request request = currentRequest;
+        if (request != null) {
+            request.notifyDisconnect(cause);
+        }
         if (stream != null) {
             stream.fail(cause);
         }
@@ -931,18 +960,20 @@ private void maybeRelease() {
 
     /** Releases the channel without reusing it. */
     private void abort() {
+        Channel ch;
         lock.lock();
         try {
             if (released) {
                 return;
             }
             released = true;
             reusable = false;
+            // Read under the lock: a channel attached later is released by attachChannel().
+            ch = this.channel;
         } finally {
             lock.unlock();
         }
         clearAbortAction();
-        Channel ch = this.channel;
         if (ch != null) {
             helper.release(ch, false);
         }

org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHelper.java

@@ -311,6 +311,8 @@ public void handle(Request request, Response response) {
                 handleOnce(current, response);
                 next = (redirects < getMaxRedirects()) ? nextRedirect(current, response) : null;
                 if (next != null) {
+                    // The redirected exchange still belongs to the caller's logical request.
+                    next.getDisconnectSignal().thenAccept(request::notifyDisconnect);
                     if (response.getEntity() != null) {
                         response.getEntity().release();
                     }