diff --git a/changes.md b/changes.md index 30ee1c7b3f..9cc07086b0 100644 --- a/changes.md +++ b/changes.md @@ -70,6 +70,8 @@ Changes log 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. + - Active client requests can be aborted; HTTP stream failures surface as I/O errors and a + communication-error response callback. - 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/Request.java b/org.restlet/src/main/java/org/restlet/Request.java index e05da2e07e..53a7ef36b6 100644 --- a/org.restlet/src/main/java/org/restlet/Request.java +++ b/org.restlet/src/main/java/org/restlet/Request.java @@ -13,6 +13,7 @@ import java.util.Set; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArraySet; +import java.util.function.BooleanSupplier; import org.restlet.data.CacheDirective; import org.restlet.data.ChallengeResponse; import org.restlet.data.CharacterSet; @@ -81,6 +82,9 @@ public static Request getCurrent() { /** The cookies provided by the client. */ private volatile Series cookies; + /** Action used by an active client connector to abort this request. */ + private volatile BooleanSupplier abortAction; + /** The trailing headers received with the request entity. */ private volatile Series
trailers; @@ -192,6 +196,7 @@ public Request(Method method, String resourceUri, Representation entity) { */ public Request(Request request) { this(request.getMethod(), new Reference(request.getResourceRef()), request.getEntity()); + this.abortAction = null; challengeResponse = request.getChallengeResponse(); copyClientInfo(request); @@ -311,7 +316,18 @@ private void copyClientInfo(final Request request) { * @return True if the request was aborted. */ public boolean abort() { - return false; + BooleanSupplier action = abortAction; + return action != null && action.getAsBoolean(); + } + + /** + * 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. + * + * @param action The abort action, or null when no exchange is active. + */ + public void setAbortAction(BooleanSupplier action) { + this.abortAction = action; } /** @@ -485,7 +501,9 @@ public Method getMethod() { /** * Returns the callback invoked on response reception. If the value is not null, then the - * associated request will be executed asynchronously. + * associated request will be executed asynchronously. For a streaming response, the callback + * can be invoked again if the exchange fails after the initial response has been received; the + * response then has a connector-error status and reading its entity fails. * * @return The callback invoked on response reception. */ 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 c26afc6cca..9b8a509347 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 @@ -92,6 +92,8 @@ class NettyClientCall { private final Condition requestSentCondition = lock.newCondition(); + private final ReentrantLock responseCallbackLock = new ReentrantLock(); + /** The kind of channel carrying the exchange, set by the helper when it is acquired. */ private volatile StreamKind kind = StreamKind.HTTP1; @@ -104,6 +106,8 @@ class NettyClientCall { /** The response being updated by this exchange. */ private volatile org.restlet.Response response; + private volatile org.restlet.Request currentRequest; + /** The request trailers to send after the request entity. */ private volatile Series
requestTrailers = new Series<>(Header.class); @@ -130,6 +134,14 @@ class NettyClientCall { private final AtomicBoolean forceEntityRelease = new AtomicBoolean(); + private final AtomicBoolean cancelled = new AtomicBoolean(); + + private boolean responseCallbackDispatched; + + private boolean responseFailureCallbackDispatched; + + private Throwable responseFailure; + /** Indicates whether the request was fully sent. */ private boolean requestSent; @@ -197,6 +209,8 @@ boolean isConfidential() { */ void handle(org.restlet.Request request, org.restlet.Response response) { this.response = response; + this.currentRequest = request; + request.setAbortAction(this::cancel); this.requestTrailers = request.getTrailers(); HeaderUtils.addGeneralHeaders(request, requestHeaders); if (request.getEntity() != null) { @@ -214,18 +228,75 @@ void handle(org.restlet.Request request, org.restlet.Response response) { if (onSent != null) { onSent.handle(request, response); } - Uniform onResponse = request.getOnResponse(); - if (onResponse != null) { - try { - onResponse.handle(request, response); - } catch (RuntimeException e) { - helper.getLogger() - .log( - Level.WARNING, - "Unexpected error or exception inside the user call back", - e); - } + dispatchInitialResponseCallback(request, response); + } + } + + private void dispatchInitialResponseCallback( + org.restlet.Request request, org.restlet.Response response) { + Uniform callback = request.getOnResponse(); + if (callback == null) { + return; + } + responseCallbackLock.lock(); + try { + Throwable failureAfterHead = getResponseFailure(); + if (failureAfterHead != null) { + response.setStatus(Status.CONNECTOR_ERROR_COMMUNICATION, failureAfterHead); + responseFailureCallbackDispatched = true; + } + responseCallbackDispatched = true; + invokeResponseCallback(callback, request, response); + } finally { + responseCallbackLock.unlock(); + } + } + + private void dispatchResponseFailureCallback(Throwable cause) { + org.restlet.Request request = currentRequest; + org.restlet.Response currentResponse = response; + if (request == null || currentResponse == null) { + return; + } + currentResponse.setStatus(Status.CONNECTOR_ERROR_COMMUNICATION, cause); + Uniform callback = request.getOnResponse(); + if (callback == null) { + return; + } + if (responseCallbackLock.isHeldByCurrentThread()) { + Thread.startVirtualThread(() -> dispatchResponseFailureCallback(cause)); + return; + } + responseCallbackLock.lock(); + try { + if (responseCallbackDispatched && !responseFailureCallbackDispatched) { + responseFailureCallbackDispatched = true; + invokeResponseCallback(callback, request, currentResponse); } + } finally { + responseCallbackLock.unlock(); + } + } + + private void invokeResponseCallback( + Uniform callback, org.restlet.Request request, org.restlet.Response response) { + try { + callback.handle(request, response); + } catch (RuntimeException e) { + helper.getLogger() + .log( + Level.WARNING, + "Unexpected error or exception inside the user call back", + e); + } + } + + private Throwable getResponseFailure() { + lock.lock(); + try { + return responseFailure; + } finally { + lock.unlock(); } } @@ -246,78 +317,108 @@ private Status sendRequest(org.restlet.Request request) { boolean backgroundWriterStarted = false; try { this.channel = helper.acquire(this); - 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); + if (cancelled.get()) { + throw new IOException("Request cancelled by client"); } + backgroundWriterStarted = sendRequestBody(request, entity, connectorService); awaitHead(); result = new Status(statusCode, reasonPhrase); - if (backgroundWriterStarted) { - boolean followRedirect = - helper.isFollowRedirects() - && !request.isAsynchronous() - && result.isRedirection(); - releaseRequestEntity.complete(!followRedirect); - if (followRedirect) { - awaitRequestSent(); - } - } + finishRequestBody(request, result, backgroundWriterStarted); } catch (ConnectException e) { - completeEntityRelease(); - helper.logFailure("Unable to connect to the remote HTTP server", e); - abort(); - result = new Status(Status.CONNECTOR_ERROR_CONNECTION, e); + result = handleConnectFailure(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); + result = handleCommunicationFailure(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); + result = handleInternalFailure(e); } finally { 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); - } + releaseRequestEntityAfterSend(request, entity, connectorService); } } return result; } + private boolean sendRequestBody( + org.restlet.Request request, Representation entity, ConnectorService connectorService) + throws IOException { + if (kind != StreamKind.HTTP2 || entity == null || entity.getSize() == 0) { + writeRequest(entity); + return false; + } + + 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); + } + } + }); + return true; + } + + private void finishRequestBody( + org.restlet.Request request, Status result, boolean backgroundWriterStarted) + throws IOException { + if (!backgroundWriterStarted) { + return; + } + boolean followRedirect = + helper.isFollowRedirects() && !request.isAsynchronous() && result.isRedirection(); + releaseRequestEntity.complete(!followRedirect); + if (followRedirect) { + awaitRequestSent(); + } + } + + private Status handleConnectFailure(ConnectException 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) { + completeEntityRelease(); + helper.logFailure( + "An error occurred during the communication with the remote HTTP server", cause); + abort(); + return new Status(Status.CONNECTOR_ERROR_COMMUNICATION, cause); + } + + private Status handleInternalFailure(RuntimeException cause) { + completeEntityRelease(); + helper.logFailure("An unexpected error occurred while sending the request", cause); + abort(); + return new Status(Status.CONNECTOR_ERROR_INTERNAL, cause); + } + + private void releaseRequestEntityAfterSend( + org.restlet.Request request, Representation entity, ConnectorService connectorService) { + // 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); + } + } + private void completeEntityRelease() { forceEntityRelease.set(true); CompletableFuture releaseEntity = releaseRequestEntity; @@ -691,20 +792,26 @@ void onResponseComplete(io.netty.handler.codec.http.HttpHeaders trailers) { * @param cause The failure cause. */ void onFailure(Throwable cause) { + boolean notifyResponseFailure = false; + boolean responseHeadReceived; lock.lock(); try { if (!requestSent) { requestFailure = cause; requestSentCondition.signalAll(); } - if (responseComplete) { + if (responseComplete && requestSent) { return; } reusable = false; if (!head) { failure = cause; headReceived.signalAll(); + } else if (responseFailure == null) { + responseFailure = cause; + notifyResponseFailure = true; } + responseHeadReceived = head; } finally { lock.unlock(); } @@ -712,9 +819,28 @@ void onFailure(Throwable cause) { if (stream != null) { stream.fail(cause); } + if (responseHeadReceived && notifyResponseFailure) { + dispatchResponseFailureCallback(cause); + } abort(); } + private boolean cancel() { + lock.lock(); + try { + if (released + || (responseComplete && requestSent) + || !cancelled.compareAndSet(false, true)) { + return false; + } + } finally { + lock.unlock(); + } + completeEntityRelease(); + onFailure(new IOException("Request cancelled by client")); + return true; + } + // ------------------------------------------------------------------ connection release /** Called when the consumer closes the response body. */ @@ -746,6 +872,7 @@ private void maybeRelease() { } Channel ch = this.channel; if (ch != null) { + clearAbortAction(); helper.release(ch, reuse); } } @@ -762,9 +889,17 @@ private void abort() { } finally { lock.unlock(); } + clearAbortAction(); Channel ch = this.channel; if (ch != null) { helper.release(ch, false); } } + + private void clearAbortAction() { + org.restlet.Request request = currentRequest; + if (request != null) { + request.setAbortAction(null); + } + } } diff --git a/org.restlet/src/main/java/org/restlet/util/WrapperRequest.java b/org.restlet/src/main/java/org/restlet/util/WrapperRequest.java index fc342133f7..d5ad4f25f7 100644 --- a/org.restlet/src/main/java/org/restlet/util/WrapperRequest.java +++ b/org.restlet/src/main/java/org/restlet/util/WrapperRequest.java @@ -11,6 +11,7 @@ import java.util.List; import java.util.Set; import java.util.concurrent.ConcurrentMap; +import java.util.function.BooleanSupplier; import org.restlet.Request; import org.restlet.Response; import org.restlet.Uniform; @@ -51,6 +52,11 @@ public boolean abort() { return wrappedRequest.abort(); } + @Override + public void setAbortAction(BooleanSupplier action) { + wrappedRequest.setAbortAction(action); + } + @Override public void commit(Response response) { wrappedRequest.commit(response); diff --git a/org.restlet/src/test/java/org/restlet/engine/connector/Http2CancellationTestCase.java b/org.restlet/src/test/java/org/restlet/engine/connector/Http2CancellationTestCase.java new file mode 100644 index 0000000000..d9b75bfd2a --- /dev/null +++ b/org.restlet/src/test/java/org/restlet/engine/connector/Http2CancellationTestCase.java @@ -0,0 +1,136 @@ +/** + * 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.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.CountDownLatch; +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.Uniform; +import org.restlet.data.MediaType; +import org.restlet.data.Method; +import org.restlet.data.Protocol; +import org.restlet.data.Status; +import org.restlet.engine.Engine; +import org.restlet.representation.OutputRepresentation; + +class Http2CancellationTestCase { + + 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 abortFailsTheResponseStreamAndNotifiesTheResponseCallback() throws Exception { + CountDownLatch responseWriteStarted = new CountDownLatch(1); + CountDownLatch continueResponseWrite = new CountDownLatch(1); + CountDownLatch responseWriteFailed = new CountDownLatch(1); + + Restlet streamingResponse = + new Restlet() { + @Override + public void handle(Request request, Response response) { + response.setEntity( + new OutputRepresentation(MediaType.TEXT_PLAIN) { + @Override + public void write(OutputStream output) throws IOException { + try { + output.write("first".getBytes(StandardCharsets.UTF_8)); + output.flush(); + responseWriteStarted.countDown(); + if (!continueResponseWrite.await( + 10, TimeUnit.SECONDS)) { + throw new IOException( + "Timed out waiting to continue the response"); + } + output.write(new byte[2 * 1024 * 1024]); + output.flush(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IOException( + "Interrupted while writing the response", e); + } catch (IOException e) { + responseWriteFailed.countDown(); + throw e; + } + } + }); + } + }; + Context serverContext = new Context(); + serverContext.getParameters().add("http.transport.protocol", "HTTP2"); + server = new Server(serverContext, Protocol.HTTP, 0, streamingResponse); + server.start(); + + Context clientContext = new Context(); + clientContext.getParameters().add("httpClientTransportMode", "HTTP2"); + client = new Client(clientContext, Protocol.HTTP); + client.start(); + + List callbackStatuses = new ArrayList<>(); + Request request = + new Request(Method.GET, "http://localhost:" + server.getActualPort() + "/"); + Uniform onResponse = (req, res) -> callbackStatuses.add(res.getStatus()); + request.setOnResponse(onResponse); + + try { + Response response = client.handle(request); + assertTrue(responseWriteStarted.await(5, TimeUnit.SECONDS)); + assertEquals(List.of(Status.SUCCESS_OK), callbackStatuses); + + response.abort(); + + assertEquals(Status.CONNECTOR_ERROR_COMMUNICATION, response.getStatus()); + assertEquals(2, callbackStatuses.size()); + assertEquals(Status.CONNECTOR_ERROR_COMMUNICATION, callbackStatuses.get(1)); + InputStream responseBody = response.getEntity().getStream(); + assertThrows(IOException.class, responseBody::readAllBytes); + } finally { + continueResponseWrite.countDown(); + } + + assertTrue(responseWriteFailed.await(10, TimeUnit.SECONDS)); + } +}