diff --git a/changes.md b/changes.md index 753c374a9c..30ee1c7b3f 100644 --- a/changes.md +++ b/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. diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java index 2a9abc787c..c26afc6cca 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientCall.java +++ b/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. * *

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 @@ class NettyClientCall { 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 @@ class NettyClientCall { /** 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 releaseRequestEntity; + + private final AtomicBoolean forceEntityRelease = new AtomicBoolean(); + /** Indicates whether the request was fully sent. */ private boolean requestSent; @@ -229,41 +243,118 @@ private Status sendRequest(org.restlet.Request request) { 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 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 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 @@ private void writeRequest(Representation entity) throws IOException { lock.lock(); try { requestSent = true; + requestSentCondition.signalAll(); } finally { lock.unlock(); } @@ -601,6 +693,10 @@ void onResponseComplete(io.netty.handler.codec.http.HttpHeaders trailers) { void onFailure(Throwable cause) { lock.lock(); try { + if (!requestSent) { + requestFailure = cause; + requestSentCondition.signalAll(); + } if (responseComplete) { return; } @@ -629,7 +725,6 @@ private void onBodyClosed() { // The rest of the body would have to be drained. reusable = false; responseComplete = true; - requestSent = true; } } finally { lock.unlock(); diff --git a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHelper.java b/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHelper.java index 2f14deefca..862361fb40 100644 --- a/org.restlet/src/main/java/org/restlet/engine/netty/NettyClientHelper.java +++ b/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(); } } diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/Http2FullDuplexTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/Http2FullDuplexTestCase.java new file mode 100644 index 0000000000..8e053de9a0 --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/engine/connector/Http2FullDuplexTestCase.java @@ -0,0 +1,123 @@ +/** + * Copyright 2005-2026 Qlik + *

+ * 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 + *

+ * 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)); + } + } +}