diff --git a/components/camel-http-common/src/main/java/org/apache/camel/http/common/CamelServlet.java b/components/camel-http-common/src/main/java/org/apache/camel/http/common/CamelServlet.java index db554a350ce99..2f7386a9817b5 100644 --- a/components/camel-http-common/src/main/java/org/apache/camel/http/common/CamelServlet.java +++ b/components/camel-http-common/src/main/java/org/apache/camel/http/common/CamelServlet.java @@ -59,6 +59,12 @@ public class CamelServlet extends HttpServlet implements HttpRegistryProvider { public static final String EXECUTOR_REF_PARAM = "executorRef"; public static final List METHODS = Arrays.asList("GET", "HEAD", "POST", "PUT", "DELETE", "TRACE", "OPTIONS", "CONNECT", "PATCH"); + /** + * Request attribute holding the {@link CompletionStage} of a request still being processed on another thread when + * {@link #doService(HttpServletRequest, HttpServletResponse)} returns. The {@link AsyncContext} must not be + * completed before it. + */ + protected static final String ASYNC_PROMISE_ATTRIBUTE_NAME = "CamelAsyncPromise"; private static final long serialVersionUID = -7061982839117697829L; @@ -141,14 +147,25 @@ private void doAsyncExecution( HttpServletRequest req, HttpServletResponse resp, HttpConsumer consumer, AsyncContext context) { try { final CompletionStage promise = doExecute(req, resp, consumer); - if (promise == null) { // early quit + completeOnCompletion(context, promise); + } catch (Exception e) { + try { + onError(resp, e); + } finally { context.complete(); - } else { - promise.whenComplete((r, e) -> context.complete()); } - } catch (Exception e) { - onError(resp, e); + } + } + + /** + * Completes the {@link AsyncContext} once the promise is done, or immediately if the request was handled + * synchronously (no promise). + */ + private static void completeOnCompletion(AsyncContext context, CompletionStage promise) { + if (promise == null) { context.complete(); + } else { + promise.whenComplete((r, e) -> context.complete()); } } @@ -213,11 +230,16 @@ protected void doServiceAsync(AsyncContext context) { final HttpServletResponse response = (HttpServletResponse) context.getResponse(); try { doService(request, response); + // doService is void (overridden by subclasses) so in-flight processing is handed over via the request + final CompletionStage promise = (CompletionStage) request.getAttribute(ASYNC_PROMISE_ATTRIBUTE_NAME); + completeOnCompletion(context, promise); } catch (Exception e) { //An error shouldn't occur as we should handle most of error in doService - onError(response, e); - } finally { - context.complete(); + try { + onError(response, e); + } finally { + context.complete(); + } } } @@ -232,7 +254,11 @@ protected void doService(HttpServletRequest request, HttpServletResponse respons log.trace("Service: {}", request); HttpConsumer consumer = doResolve(request, response); if (consumer != null) { - doExecute(request, response, consumer); + CompletionStage promise = doExecute(request, response, consumer); + if (promise != null) { + // still in-flight on another thread, which will write the response + request.setAttribute(ASYNC_PROMISE_ATTRIBUTE_NAME, promise); + } } } @@ -314,6 +340,8 @@ private CompletionStage doExecute(HttpServletRequest req, HttpServletResponse } } catch (Exception e) { exchange.setException(e); + // processAsync failed synchronously so write the response here + isAsync = false; } try { @@ -334,12 +362,12 @@ private CompletionStage tryAsyncProcess( .whenComplete((r, ex) -> { if (ex != null) { exchange.setException(ex); - } else { - try { - afterProcess(res, consumer, exchange, false); - } catch (Exception e) { - exchange.setException(e); - } + } + // always write the response (error or not) and finish the UoW + try { + afterProcess(res, consumer, exchange, false); + } catch (Exception e) { + exchange.setException(e); } }); return result; diff --git a/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/JettyAsyncDelayedRouteTest.java b/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/JettyAsyncDelayedRouteTest.java new file mode 100644 index 0000000000000..3d56c3c1a35a8 --- /dev/null +++ b/components/camel-jetty/src/test/java/org/apache/camel/component/jetty/JettyAsyncDelayedRouteTest.java @@ -0,0 +1,48 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.jetty; + +import org.apache.camel.builder.RouteBuilder; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * async=true must wait for a route step that resumes on another thread before completing the request. + */ +public class JettyAsyncDelayedRouteTest extends BaseJettyTest { + + @Test + public void testAsyncRouteCompletesBeforeResponse() { + String body = template.requestBody("http://localhost:{{port}}/racy", "hello", String.class); + assertEquals("delayed-response", body); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + public void configure() { + // async and continuation is not compatible! + from("jetty:http://localhost:{{port}}/racy?async=true&useContinuation=false") + // resumes on another thread + .delay(300).asyncDelayed().end() + .transform().constant("delayed-response"); + } + }; + } + +} diff --git a/components/camel-servlet/src/test/java/org/apache/camel/component/servlet/ServletAsyncErrorTest.java b/components/camel-servlet/src/test/java/org/apache/camel/component/servlet/ServletAsyncErrorTest.java new file mode 100644 index 0000000000000..d72d807d6f745 --- /dev/null +++ b/components/camel-servlet/src/test/java/org/apache/camel/component/servlet/ServletAsyncErrorTest.java @@ -0,0 +1,98 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.servlet; + +import java.util.concurrent.CompletableFuture; + +import io.undertow.servlet.Servlets; +import io.undertow.servlet.api.DeploymentInfo; +import org.apache.camel.AsyncCallback; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.support.AsyncProcessorSupport; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * async=true must still write an error response when the route fails after resuming on another thread, or when the + * processor's stage completes exceptionally. + */ +public class ServletAsyncErrorTest extends ServletCamelRouterTestSupport { + + @Test + public void testAsyncRouteThrows() throws Exception { + WebResponse response = query(new GetMethodWebRequest(contextUrl + "/services/async-error"), false); + + assertEquals(500, response.getResponseCode()); + } + + @Test + public void testAsyncStageCompletesExceptionally() throws Exception { + // a route processor never completes the stage exceptionally (the exception lands on the exchange), so + // plug a consumer with a custom AsyncProcessor directly + ServletEndpoint endpoint = context.getEndpoint("servlet:///async-failed", ServletEndpoint.class); + ServletConsumer consumer = (ServletConsumer) endpoint.createConsumer(new AsyncProcessorSupport() { + @Override + public boolean process(Exchange exchange, AsyncCallback callback) { + callback.done(true); + return true; + } + + @Override + public CompletableFuture processAsync(Exchange exchange) { + return CompletableFuture.failedFuture(new IllegalStateException("stage failed")); + } + }); + consumer.start(); + try { + WebResponse response = query(new GetMethodWebRequest(contextUrl + "/services/async-failed"), false); + + assertEquals(500, response.getResponseCode()); + } finally { + consumer.stop(); + } + } + + @Override + protected DeploymentInfo getDeploymentInfo() { + return Servlets.deployment() + .setClassLoader(getClass().getClassLoader()) + .setContextPath(CONTEXT) + .setDeploymentName(getClass().getName()) + .addServlet(Servlets.servlet("CamelServlet", CamelHttpTransportServlet.class) + .addInitParam("async", "true") + .setAsyncSupported(true) + .addMapping("/services/*")); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from("servlet:///async-error") + // resumes on another thread, then fails + .delay(100).asyncDelayed().end() + .process(e -> { + throw new IllegalStateException("boom"); + }); + } + }; + } + +} diff --git a/components/camel-servlet/src/test/java/org/apache/camel/component/servlet/ServletAsyncNoExecutorRefRaceTest.java b/components/camel-servlet/src/test/java/org/apache/camel/component/servlet/ServletAsyncNoExecutorRefRaceTest.java new file mode 100644 index 0000000000000..c3cbaa984a800 --- /dev/null +++ b/components/camel-servlet/src/test/java/org/apache/camel/component/servlet/ServletAsyncNoExecutorRefRaceTest.java @@ -0,0 +1,56 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.component.servlet; + +import io.undertow.servlet.Servlets; +import io.undertow.servlet.api.DeploymentInfo; +import org.junit.jupiter.api.Test; +import org.springframework.web.context.ContextLoaderListener; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +/** + * async=true without executorRef must wait for a route step that resumes on another thread before completing. + */ +public class ServletAsyncNoExecutorRefRaceTest extends ServletCamelRouterTestSupport { + + @Test + public void testAsyncRouteCompletesBeforeResponse() throws Exception { + WebRequest req = new GetMethodWebRequest(contextUrl + "/services/racy"); + WebResponse response = query(req, false); + + assertEquals(200, response.getResponseCode()); + assertEquals("delayed-response", response.getText()); + } + + @Override + protected DeploymentInfo getDeploymentInfo() { + return Servlets.deployment() + .setClassLoader(getClass().getClassLoader()) + .setContextPath(CONTEXT) + .setDeploymentName(getClass().getName()) + .addInitParameter("contextConfigLocation", + "classpath:org/apache/camel/component/servlet/example-camelContext-race.xml") + .addListener(Servlets.listener(ContextLoaderListener.class)) + .addServlet(Servlets.servlet("CamelServlet", CamelHttpTransportServlet.class) + .addInitParam("async", "true") + .setLoadOnStartup(1) + .setAsyncSupported(true) + .addMapping("/services/*")); + } + +} diff --git a/components/camel-servlet/src/test/resources/org/apache/camel/component/servlet/example-camelContext-race.xml b/components/camel-servlet/src/test/resources/org/apache/camel/component/servlet/example-camelContext-race.xml new file mode 100644 index 0000000000000..a9436e99ae1aa --- /dev/null +++ b/components/camel-servlet/src/test/resources/org/apache/camel/component/servlet/example-camelContext-race.xml @@ -0,0 +1,44 @@ + + + + + + + + + + + 300 + + + delayed-response + + + + +