feat: support full-duplex HTTP/2 requests

changes.md

@@ -68,6 +68,8 @@ Changes log
           `Response.toString()` now shows it.
         - Added request and response trailers to the HTTP connectors, supporting HTTP/1.1 chunked
           transfers and HTTP/2 trailing headers.
+        - HTTP/2 clients receive a response as soon as its headers arrive, while the request entity
+          continues to be written on a virtual thread.
     - Bugs fixed
         - Slf4j bridge reports wrong line number and caller class in log4j. Issue #233.
         - When GET is an allowed method of a server resource, HEAD is now also listed (e.g. in the Allow header of a 405 response). Issue #266.

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

@@ -31,7 +31,9 @@
 import java.io.PushbackInputStream;
 import java.net.ConnectException;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.logging.Level;
@@ -51,13 +53,15 @@
 
 /**
  * HTTP client exchange based on Netty, mapping a Restlet request and response to Netty messages.
- * The calling thread blocks while the request is written and until the response head is received;
- * the response body is then streamed from the connection.
+ * The calling thread waits until the response head is received; on HTTP/2, the request body is
+ * written concurrently. The response body is then streamed from the connection.
  *
  * <p>This class is internal to the Netty connectors.
  */
 class NettyClientCall {
 
+    static final String ASYNC_REQUEST_BODY_ATTRIBUTE = "org.restlet.engine.netty.asyncRequestBody";
+
     /** The parent helper. */
     private final NettyClientHelper helper;
 
@@ -86,6 +90,8 @@
 
     private final Condition headReceived = lock.newCondition();
 
+    private final Condition requestSentCondition = lock.newCondition();
+
     /** The kind of channel carrying the exchange, set by the helper when it is acquired. */
     private volatile StreamKind kind = StreamKind.HTTP1;
 
@@ -116,6 +122,14 @@
     /** The failure before the response head, if any. */
     private Throwable failure;
 
+    /** The failure while writing the request body, if any. */
+    private Throwable requestFailure;
+
+    /** Signals whether the background writer or the redirect handler releases the entity. */
+    private volatile CompletableFuture<Boolean> releaseRequestEntity;
+
+    private final AtomicBoolean forceEntityRelease = new AtomicBoolean();
+
     /** Indicates whether the request was fully sent. */
     private boolean requestSent;
 
@@ -221,7 +235,7 @@
      * @param request The Restlet request.
      * @return The response status, or a connector error status.
      */
     private Status sendRequest(org.restlet.Request request) {
         Status result;
         Representation entity = request.isEntityAvailable() ? request.getEntity() : null;
         ConnectorService connectorService = ConnectorHelper.getConnectorService();
@@ -229,41 +243,118 @@
             connectorService.beforeSend(entity);
         }
 
+        boolean backgroundWriterStarted = false;
         try {
             this.channel = helper.acquire(this);
-            writeRequest(entity);
+            boolean fullDuplex =
+                    kind == StreamKind.HTTP2 && entity != null && entity.getSize() != 0;
+            if (fullDuplex) {
+                CompletableFuture<Boolean> releaseEntity = new CompletableFuture<>();
+                releaseRequestEntity = releaseEntity;
+                request.getAttributes().put(ASYNC_REQUEST_BODY_ATTRIBUTE, Boolean.TRUE);
+                Thread.startVirtualThread(
+                        () -> {
+                            try {
+                                writeRequest(entity);
+                            } catch (IOException | RuntimeException e) {
+                                onFailure(e);
+                            } finally {
+                                if (releaseEntity.join() || forceEntityRelease.get()) {
+                                    entity.release();
+                                }
+                                if (connectorService != null) {
+                                    connectorService.afterSend(entity);
+                                }
+                            }
+                        });
+                backgroundWriterStarted = true;
+            } else {
+                writeRequest(entity);
+            }
             awaitHead();
             result = new Status(statusCode, reasonPhrase);
+            if (backgroundWriterStarted) {
+                boolean followRedirect =
+                        helper.isFollowRedirects()
+                                && !request.isAsynchronous()
+                                && result.isRedirection();
+                releaseRequestEntity.complete(!followRedirect);
+                if (followRedirect) {
+                    awaitRequestSent();
+                }
+            }
         } catch (ConnectException e) {
+            completeEntityRelease();
             helper.logFailure("Unable to connect to the remote HTTP server", e);
             abort();
             result = new Status(Status.CONNECTOR_ERROR_CONNECTION, e);
         } catch (IOException e) {
+            completeEntityRelease();
             helper.logFailure(
                     "An error occurred during the communication with the remote HTTP server", e);
             abort();
             result = new Status(Status.CONNECTOR_ERROR_COMMUNICATION, e);
         } catch (RuntimeException e) {
+            completeEntityRelease();
             helper.logFailure("An unexpected error occurred while sending the request", e);
             abort();
             result = new Status(Status.CONNECTOR_ERROR_INTERNAL, e);
         } finally {
-            // A repeatable entity is kept for a possible redirect (307/308) and released by the
-            // helper once the redirects are followed.
-            if (entity != null
-                    && (entity.isTransient()
-                            || !helper.isFollowRedirects()
-                            || request.isAsynchronous())) {
-                entity.release();
-            }
-            if (connectorService != null) {
-                connectorService.afterSend(entity);
+            if (!backgroundWriterStarted) {
+                // A repeatable entity is kept for a possible redirect (307/308) and released by
+                // the helper once the redirects are followed.
+                if (entity != null
+                        && (entity.isTransient()
+                                || !helper.isFollowRedirects()
+                                || request.isAsynchronous())) {
+                    entity.release();
+                }
+                if (connectorService != null) {
+                    connectorService.afterSend(entity);
+                }
             }
         }
 
         return result;
     }
 
+    private void completeEntityRelease() {
+        forceEntityRelease.set(true);
+        CompletableFuture<Boolean> releaseEntity = releaseRequestEntity;
+        if (releaseEntity != null) {
+            releaseEntity.complete(true);
+        }
+    }
+
+    private void awaitRequestSent() throws IOException {
+        long timeout = helper.getIdleTimeout();
+        lock.lock();
+        try {
+            long remaining =
+                    (timeout > 0) ? TimeUnit.MILLISECONDS.toNanos(timeout) : Long.MAX_VALUE;
+            while (!requestSent && requestFailure == null) {
+                if (timeout > 0) {
+                    if (remaining <= 0) {
+                        throw new InterruptedIOException("Timed out waiting for the request body");
+                    }
+                    remaining = requestSentCondition.awaitNanos(remaining);
+                } else {
+                    requestSentCondition.await();
+                }
+            }
+            if (requestFailure != null) {
+                throw (requestFailure instanceof IOException ioe)
+                        ? ioe
+                        : new IOException("Unable to write the request body", requestFailure);
+            }
+        } catch (InterruptedException _) {
+            Thread.currentThread().interrupt();
+            throw new InterruptedIOException("Interrupted while waiting for the request body");
+        } finally {
+            lock.unlock();
+        }
+    }
+
     /**
      * Updates the Restlet response with the status, headers and entity received.
      *
@@ -407,6 +498,7 @@
         lock.lock();
         try {
             requestSent = true;
+            requestSentCondition.signalAll();
         } finally {
             lock.unlock();
         }
@@ -601,6 +693,10 @@
     void onFailure(Throwable cause) {
         lock.lock();
         try {
+            if (!requestSent) {
+                requestFailure = cause;
+                requestSentCondition.signalAll();
+            }
             if (responseComplete) {
                 return;
             }
@@ -629,7 +725,6 @@
                 // The rest of the body would have to be drained.
                 reusable = false;
                 responseComplete = true;
-                requestSent = true;
             }
         } finally {
             lock.unlock();

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

@@ -293,7 +293,11 @@ NettyClientCall create(Request request) {
     @Override
     public void handle(Request request, Response response) {
         if (!isFollowRedirects() || request.isAsynchronous()) {
-            handleOnce(request, response);
+            try {
+                handleOnce(request, response);
+            } finally {
+                request.getAttributes().remove(NettyClientCall.ASYNC_REQUEST_BODY_ATTRIBUTE);
+            }
             return;
         }
 
@@ -315,8 +319,13 @@ public void handle(Request request, Response response) {
                 }
             }
         } finally {
-            // The request entity was kept available for redirects, see NettyClientCall.
-            if (request.getEntity() != null) {
+            boolean entityReleasedByWriter =
+                    Boolean.TRUE.equals(
+                                    request.getAttributes()
+                                            .remove(NettyClientCall.ASYNC_REQUEST_BODY_ATTRIBUTE))
+                            && !response.getStatus().isRedirection();
+            // A full-duplex HTTP/2 body writer releases the entity once it has finished writing.
+            if (request.getEntity() != null && !entityReleasedByWriter) {
                 request.getEntity().release();
             }
         }

org.restlet/src/test/java/org/restlet/engine/connector/Http2FullDuplexTestCase.java

@@ -0,0 +1,123 @@
+/**
+ * Copyright 2005-2026 Qlik
+ *<p>
+ * The content of this file is subject to the terms of the Apache 2.0 open
+ * source license available at https://www.opensource.org/licenses/apache-2.0
+ *<p>
+ * Restlet is a registered trademark of QlikTech International AB.
+ */
+package org.restlet.engine.connector;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.restlet.Client;
+import org.restlet.Context;
+import org.restlet.Request;
+import org.restlet.Response;
+import org.restlet.Restlet;
+import org.restlet.Server;
+import org.restlet.data.MediaType;
+import org.restlet.data.Method;
+import org.restlet.data.Protocol;
+import org.restlet.engine.Engine;
+import org.restlet.representation.OutputRepresentation;
+
+class Http2FullDuplexTestCase {
+
+    private Server server;
+
+    private Client client;
+
+    @BeforeEach
+    void setUp() {
+        Engine.clearThreadLocalVariables();
+        Engine.register(false);
+        Engine.getInstance().getRegisteredServers().add(new HttpServerHelper(null));
+        Engine.getInstance().getRegisteredClients().add(new HttpClientHelper(null));
+    }
+
+    @AfterEach
+    void tearDown() throws Exception {
+        if (client != null) {
+            client.stop();
+        }
+        if (server != null) {
+            server.stop();
+        }
+        Engine.clearThreadLocalVariables();
+    }
+
+    @Test
+    void receivesResponseWhileRequestEntityIsStillBeingWritten() throws Exception {
+        CountDownLatch requestWriteStarted = new CountDownLatch(1);
+        CountDownLatch continueRequestWrite = new CountDownLatch(1);
+        CountDownLatch requestWriteFinished = new CountDownLatch(1);
+
+        Restlet earlyResponse =
+                new Restlet() {
+                    @Override
+                    public void handle(Request request, Response response) {
+                        response.setEntity("early response", MediaType.TEXT_PLAIN);
+                    }
+                };
+        Context serverContext = new Context();
+        serverContext.getParameters().add("http.transport.protocol", "HTTP2");
+        server = new Server(serverContext, Protocol.HTTP, 0, earlyResponse);
+        server.start();
+
+        Context clientContext = new Context();
+        clientContext.getParameters().add("httpClientTransportMode", "HTTP2");
+        client = new Client(clientContext, Protocol.HTTP);
+        client.start();
+
+        OutputRepresentation entity =
+                new OutputRepresentation(MediaType.TEXT_PLAIN) {
+                    @Override
+                    public void write(OutputStream output) throws IOException {
+                        try {
+                            output.write("first".getBytes(StandardCharsets.UTF_8));
+                            output.flush();
+                            requestWriteStarted.countDown();
+                            if (!continueRequestWrite.await(10, TimeUnit.SECONDS)) {
+                                throw new IOException("Timed out waiting to continue the request");
+                            }
+                            output.write("second".getBytes(StandardCharsets.UTF_8));
+                        } catch (InterruptedException e) {
+                            Thread.currentThread().interrupt();
+                            throw new IOException("Interrupted while writing the request", e);
+                        } finally {
+                            requestWriteFinished.countDown();
+                        }
+                    }
+                };
+        Request request =
+                new Request(
+                        Method.POST, "http://localhost:" + server.getActualPort() + "/", entity);
+
+        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
+            var responseFuture = executor.submit(() -> client.handle(request));
+            try {
+                assertTrue(requestWriteStarted.await(5, TimeUnit.SECONDS));
+                Response response = responseFuture.get(5, TimeUnit.SECONDS);
+                assertEquals("early response", response.getEntityAsText());
+                assertFalse(
+                        requestWriteFinished.await(100, TimeUnit.MILLISECONDS),
+                        "The response must arrive before the request entity is finished");
+            } finally {
+                continueRequestWrite.countDown();
+            }
+            assertTrue(requestWriteFinished.await(5, TimeUnit.SECONDS));
+        }
+    }
+}