This is an automated email from the ASF dual-hosted git repository.
ffang pushed a commit to branch 4.1.x-fixes
in repository https://gitbox.apache.org/repos/asf/cxf.git
The following commit(s) were added to refs/heads/4.1.x-fixes by this push:
new 8e5adcd2f11 CXF-7232: Do not block the caller thread of an async call
that may be… (#3645)
8e5adcd2f11 is described below
commit 8e5adcd2f116302c85ee2664c150df54fbda56eb
Author: Freeman(Yue) Fang <[email protected]>
AuthorDate: Fri Oct 30 16:06:21 2026 -0400
CXF-7232: Do not block the caller thread of an async call that may be…
(#3645)
* CXF-7232: Do not block the caller thread of an async call that may be
retransmitted
When the request was cached to be retransmitted, because of AutoRedirect,
MaxRetransmits or the auth supplier, HTTPConduit processed the retransmits
on the caller thread before it handed the response over to the work queue.
The retransmits wait for the response code, so an asynchronous call blocked
its caller thread until the response was received, with all the conduits.
The retransmits of an asynchronous exchange are now processed on the thread
that handles the response, before the response itself, and the cached
request is kept until then.
The Apache HttpClient conduits set the exchange as asynchronous again on
each retransmission, so the response of a redirect was handled while the
redirect was followed, and the empty body of the redirect was parsed as the
response. The thread that retransmits now handles the response of the
retransmission.
* Process the retransmits of synchronous and asynchronous exchanges in one
place
(cherry picked from commit e20f486b9ed8dc0b9749d211d671f421d7e27486)
---
.../http/asyncclient/AsyncHTTPConduit.java | 5 +-
.../asyncclient/URLConnectionAsyncHTTPConduit.java | 5 +-
.../http/asyncclient/hc5/AsyncHTTPConduit.java | 11 +-
.../hc5/URLConnectionAsyncHTTPConduit.java | 11 +-
.../org/apache/cxf/transport/http/HTTPConduit.java | 50 ++-
.../apache/cxf/transport/http/MockHTTPConduit.java | 5 +
.../cxf/systest/hc5/http/AsyncRetransmitTest.java | 368 +++++++++++++++++++++
.../cxf/systest/http/AsyncRetransmitTest.java | 368 +++++++++++++++++++++
8 files changed, 795 insertions(+), 28 deletions(-)
diff --git
a/rt/transports/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduit.java
b/rt/transports/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduit.java
index f7e4867d7e4..2e941b7bbac 100755
---
a/rt/transports/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduit.java
+++
b/rt/transports/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/AsyncHTTPConduit.java
@@ -870,8 +870,9 @@ public class AsyncHTTPConduit extends HttpClientHTTPConduit
{
protected void setupNewConnection(String newURL) throws IOException {
httpResponse = null;
- isAsync = outMessage != null && outMessage.getExchange() != null
- && !outMessage.getExchange().isSynchronous();
+ // only used to retransmit: the thread that retransmits waits for
the response and handles it,
+ // for an asynchronous exchange too, so the response must not be
dispatched again when it is received
+ isAsync = false;
exception = null;
connectionFuture = null;
session = null;
diff --git
a/rt/transports/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/URLConnectionAsyncHTTPConduit.java
b/rt/transports/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/URLConnectionAsyncHTTPConduit.java
index 9ced688c5e6..b563833d650 100755
---
a/rt/transports/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/URLConnectionAsyncHTTPConduit.java
+++
b/rt/transports/http-hc/src/main/java/org/apache/cxf/transport/http/asyncclient/URLConnectionAsyncHTTPConduit.java
@@ -867,8 +867,9 @@ public class URLConnectionAsyncHTTPConduit extends
URLConnectionHTTPConduit {
protected void setupNewConnection(String newURL) throws IOException {
httpResponse = null;
- isAsync = outMessage != null && outMessage.getExchange() != null
- && !outMessage.getExchange().isSynchronous();
+ // only used to retransmit: the thread that retransmits waits for
the response and handles it,
+ // for an asynchronous exchange too, so the response must not be
dispatched again when it is received
+ isAsync = false;
exception = null;
connectionFuture = null;
session = null;
diff --git
a/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/AsyncHTTPConduit.java
b/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/AsyncHTTPConduit.java
index 2d280edf8ba..806e6c2fa79 100644
---
a/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/AsyncHTTPConduit.java
+++
b/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/AsyncHTTPConduit.java
@@ -706,10 +706,8 @@ public class AsyncHTTPConduit extends
HttpClientHTTPConduit {
}
protected void handleResponseAsync() throws IOException {
- // The response hasn't been handled yet, should be handled
asynchronously
- if (httpResponse == null) {
- isAsync = true;
- }
+ // The response is handled when it is received, as isAsync is set
for an asynchronous
+ // exchange. It is not set again, as a retransmission may already
be in progress.
}
protected void closeInputStream() throws IOException {
@@ -883,8 +881,9 @@ public class AsyncHTTPConduit extends HttpClientHTTPConduit
{
protected void setupNewConnection(String newURL) throws IOException {
httpResponse = null;
- isAsync = outMessage != null && outMessage.getExchange() != null
- && !outMessage.getExchange().isSynchronous();
+ // only used to retransmit: the thread that retransmits waits for
the response and handles it,
+ // for an asynchronous exchange too, so the response must not be
dispatched again when it is received
+ isAsync = false;
exception = null;
connectionFuture = null;
session = null;
diff --git
a/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/URLConnectionAsyncHTTPConduit.java
b/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/URLConnectionAsyncHTTPConduit.java
index 900c778adad..2af1ca21d27 100644
---
a/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/URLConnectionAsyncHTTPConduit.java
+++
b/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/URLConnectionAsyncHTTPConduit.java
@@ -699,10 +699,8 @@ public class URLConnectionAsyncHTTPConduit extends
URLConnectionHTTPConduit {
}
protected void handleResponseAsync() throws IOException {
- // The response hasn't been handled yet, should be handled
asynchronously
- if (httpResponse == null) {
- isAsync = true;
- }
+ // The response is handled when it is received, as isAsync is set
for an asynchronous
+ // exchange. It is not set again, as a retransmission may already
be in progress.
}
protected void closeInputStream() throws IOException {
@@ -876,8 +874,9 @@ public class URLConnectionAsyncHTTPConduit extends
URLConnectionHTTPConduit {
protected void setupNewConnection(String newURL) throws IOException {
httpResponse = null;
- isAsync = outMessage != null && outMessage.getExchange() != null
- && !outMessage.getExchange().isSynchronous();
+ // only used to retransmit: the thread that retransmits waits for
the response and handles it,
+ // for an asynchronous exchange too, so the response must not be
dispatched again when it is received
+ isAsync = false;
exception = null;
connectionFuture = null;
session = null;
diff --git
a/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HTTPConduit.java
b/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HTTPConduit.java
index 77721d73466..4bf882924fd 100644
---
a/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HTTPConduit.java
+++
b/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HTTPConduit.java
@@ -1235,7 +1235,7 @@ public abstract class HTTPConduit
Runnable runnable = new Runnable() {
public void run() {
try {
- handleResponseInternal();
+ handleRetransmitsAndResponse();
} catch (Throwable e) {
((PhaseInterceptorChain)outMessage.getInterceptorChain()).abort();
outMessage.setContent(Exception.class, e);
@@ -1300,7 +1300,7 @@ public abstract class HTTPConduit
hasLoggedAsyncWarning = true;
}
LOG.fine("EXECUTOR_FULL");
- handleResponseInternal();
+ handleRetransmitsAndResponse();
}
}
}
@@ -1415,12 +1415,14 @@ public abstract class HTTPConduit
cachedStream.closeFlowthroughStream();
}
+ boolean handled = false;
try {
+ // the cached request is released once the retransmits are
done
handleResponse();
+ handled = true;
} finally {
- if (cachingForRetransmission && cachedStream != null) {
- cachedStream.close();
- cachedStream = null;
+ if (!handled) {
+ releaseCachedStream();
}
}
} catch (HttpRetryException e) {
@@ -1614,18 +1616,42 @@ public abstract class HTTPConduit
* @throws IOException
*/
protected void handleResponse() throws IOException {
- // Process retransmits until we fall out.
- handleRetransmits();
-
- if (outMessage == null
- || outMessage.getExchange() == null
- || outMessage.getExchange().isSynchronous()) {
- handleResponseInternal();
+ if (isSynchronous()) {
+ handleRetransmitsAndResponse();
} else {
+ // the retransmits wait for the response, so they are
processed with it
+ // on the thread that handles the response, not to block the
caller thread
handleResponseAsync();
}
}
+ private boolean isSynchronous() {
+ return outMessage == null
+ || outMessage.getExchange() == null
+ || outMessage.getExchange().isSynchronous();
+ }
+
+ /**
+ * Processes the retransmits until we fall out, and then the response.
It runs on the
+ * caller thread for a synchronous exchange, and on the thread that
handles the response
+ * for an asynchronous one. The cached request is released once the
retransmits are done.
+ */
+ private void handleRetransmitsAndResponse() throws IOException {
+ try {
+ handleRetransmits();
+ } finally {
+ releaseCachedStream();
+ }
+ handleResponseInternal();
+ }
+
+ private void releaseCachedStream() throws IOException {
+ if (cachingForRetransmission && cachedStream != null) {
+ cachedStream.close();
+ cachedStream = null;
+ }
+ }
+
/**
* This predicate returns true if the exchange indicates
* a oneway MEP.
diff --git
a/rt/transports/http/src/test/java/org/apache/cxf/transport/http/MockHTTPConduit.java
b/rt/transports/http/src/test/java/org/apache/cxf/transport/http/MockHTTPConduit.java
index 94b5d120f83..9eb4df2ccf6 100644
---
a/rt/transports/http/src/test/java/org/apache/cxf/transport/http/MockHTTPConduit.java
+++
b/rt/transports/http/src/test/java/org/apache/cxf/transport/http/MockHTTPConduit.java
@@ -106,6 +106,11 @@ public class MockHTTPConduit extends HTTPConduit {
}
+ @Override
+ protected void handleRetransmits() throws IOException {
+ // there is no response that could require a retransmission
+ }
+
@Override
protected void handleResponseInternal() throws IOException {
outMessage.put(Thread.class, Thread.currentThread());
diff --git
a/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/http/AsyncRetransmitTest.java
b/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/http/AsyncRetransmitTest.java
new file mode 100644
index 00000000000..19a279ec5a5
--- /dev/null
+++
b/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/http/AsyncRetransmitTest.java
@@ -0,0 +1,368 @@
+/**
+ * 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.cxf.systest.hc5.http;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.StringReader;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.xml.namespace.QName;
+import javax.xml.transform.Source;
+import javax.xml.transform.stream.StreamSource;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+import jakarta.xml.ws.Dispatch;
+import jakarta.xml.ws.Response;
+import jakarta.xml.ws.Service;
+import jakarta.xml.ws.soap.SOAPBinding;
+import org.apache.cxf.Bus;
+import org.apache.cxf.BusFactory;
+import org.apache.cxf.configuration.security.AuthorizationPolicy;
+import org.apache.cxf.io.CachedConstants;
+import org.apache.cxf.jaxws.DispatchImpl;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.staxutils.StaxUtils;
+import org.apache.cxf.testutil.common.TestUtil;
+import org.apache.cxf.transport.http.HTTPConduit;
+import org.apache.cxf.transport.http.asyncclient.hc5.AsyncHTTPConduit;
+import org.apache.cxf.transport.http.auth.HttpAuthSupplier;
+import org.apache.cxf.transports.http.configuration.HTTPClientPolicy;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * CXF-7232: an asynchronous call must not block the caller thread until the
response is received,
+ * when the request is cached to be retransmitted on a redirect or with
MaxRetransmits.
+ */
+@RunWith(Parameterized.class)
+public class AsyncRetransmitTest {
+ public static final String PORT =
TestUtil.getPortNumber(AsyncRetransmitTest.class);
+ private static final QName SERVICE = new QName("urn:cxf7232",
"EchoService");
+ private static final QName PORT_NAME = new QName("urn:cxf7232",
"EchoPort");
+ private static final String REQUEST = "<echo
xmlns=\"urn:cxf7232\">hello</echo>";
+ private static final String RESPONSE =
+ "<soap:Envelope
xmlns:soap=\"http://schemas.xmlsoap.org/soap/envelope/\";><soap:Body>"
+ + "<echoResponse
xmlns=\"urn:cxf7232\">hello</echoResponse></soap:Body></soap:Envelope>";
+
+ private static HttpServer server;
+ private static ExecutorService executor;
+ private static final AtomicInteger REDIRECTS = new AtomicInteger();
+ private static final AtomicBoolean RESPONDED = new AtomicBoolean();
+ private static volatile CountDownLatch release;
+ private static final BlockingQueue<byte[]> REQUESTS = new
LinkedBlockingQueue<>();
+ private static final AtomicInteger CHALLENGES = new AtomicInteger();
+
+ @Rule
+ public TemporaryFolder cacheFolder = new TemporaryFolder();
+
+ private final String conduit;
+
+ public AsyncRetransmitTest(String conduit) {
+ this.conduit = conduit;
+ }
+
+ @Parameters(name = "{0}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {{"HttpClient"},
{"URLConnection"}, {"hc5"}});
+ }
+
+ @BeforeClass
+ public static void startServer() throws IOException {
+ executor = Executors.newCachedThreadPool();
+ server = HttpServer.create(new InetSocketAddress("localhost",
Integer.parseInt(PORT)), 0);
+ server.setExecutor(executor);
+ // redirects at once
+ server.createContext("/redirect", ex -> {
+ captureRequest(ex);
+ REDIRECTS.incrementAndGet();
+ ex.getResponseHeaders().add("Location", address("/echo"));
+ ex.sendResponseHeaders(307, -1);
+ ex.close();
+ });
+ // answers when the test releases it
+ server.createContext("/echo", ex -> {
+ captureRequest(ex);
+ awaitRelease();
+ respond(ex);
+ });
+ server.createContext("/auth", ex -> {
+ captureRequest(ex);
+ awaitRelease();
+ if ("Basic
dXNlcjpwYXNz".equals(ex.getRequestHeaders().getFirst("Authorization"))) {
+ respond(ex);
+ } else {
+ CHALLENGES.incrementAndGet();
+ RESPONDED.set(true);
+ ex.getResponseHeaders().add("WWW-Authenticate", "Basic
realm=\"cxf7232\"");
+ ex.sendResponseHeaders(401, -1);
+ ex.close();
+ }
+ });
+ server.createContext("/loop", ex -> {
+ captureRequest(ex);
+ awaitRelease();
+ RESPONDED.set(true);
+ ex.getResponseHeaders().add("Location", address("/loop"));
+ ex.sendResponseHeaders(307, -1);
+ ex.close();
+ });
+ server.start();
+ }
+
+ @AfterClass
+ public static void stopServer() {
+ server.stop(0);
+ executor.shutdownNow();
+ }
+
+ @Before
+ public void reset() {
+ REDIRECTS.set(0);
+ CHALLENGES.set(0);
+ REQUESTS.clear();
+ RESPONDED.set(false);
+ release = new CountDownLatch(1);
+ }
+
+ @After
+ public void releaseServer() {
+ release.countDown();
+ }
+
+ @Test
+ public void testAutoRedirect() throws Exception {
+ HTTPClientPolicy policy = new HTTPClientPolicy();
+ policy.setAutoRedirect(true);
+ invokeAsync("/redirect", policy);
+ assertEquals(1, REDIRECTS.get());
+ }
+
+ @Test
+ public void testAutoRedirectWithoutRedirect() throws Exception {
+ HTTPClientPolicy policy = new HTTPClientPolicy();
+ policy.setAutoRedirect(true);
+ invokeAsync("/echo", policy);
+ assertEquals(0, REDIRECTS.get());
+ }
+
+ @Test
+ public void testMaxRetransmits() throws Exception {
+ HTTPClientPolicy policy = new HTTPClientPolicy();
+ policy.setMaxRetransmits(1);
+ invokeAsync("/echo", policy);
+ assertEquals(0, REDIRECTS.get());
+ }
+
+ @Test
+ public void testRedirectWithDiskCachedRequest() throws Exception {
+ String payload = "<echo xmlns=\"urn:cxf7232\">" +
"0123456789abcdef".repeat(16384) + "</echo>";
+ File directory = cacheFolder.newFolder("request-cache");
+ Bus bus = BusFactory.getThreadDefaultBus();
+ Object oldDirectory =
bus.getProperty(CachedConstants.OUTPUT_DIRECTORY_BUS_PROP);
+ Object oldThreshold =
bus.getProperty(CachedConstants.THRESHOLD_BUS_PROP);
+ bus.setProperty(CachedConstants.OUTPUT_DIRECTORY_BUS_PROP,
directory.getAbsolutePath());
+ bus.setProperty(CachedConstants.THRESHOLD_BUS_PROP, "1024");
+ HTTPClientPolicy policy = new HTTPClientPolicy();
+ policy.setAutoRedirect(true);
+ Dispatch<Source> dispatch = createDispatch("/redirect", policy);
+ try {
+ Response<Source> response = dispatch.invokeAsync(new
StreamSource(new StringReader(payload)));
+ assertFalse("The caller thread waited for the response",
RESPONDED.get());
+ File[] files = directory.listFiles();
+ assertNotNull(files);
+ assertTrue("The request was not cached on disk", files.length > 0);
+ release.countDown();
+ assertResponse(response);
+ assertRequests(payload, 2);
+ assertEquals(1, REDIRECTS.get());
+ } finally {
+ release.countDown();
+ destroy(dispatch);
+ bus.setProperty(CachedConstants.OUTPUT_DIRECTORY_BUS_PROP,
oldDirectory);
+ bus.setProperty(CachedConstants.THRESHOLD_BUS_PROP, oldThreshold);
+ }
+ }
+
+ @Test
+ public void testAuthenticationChallenge() throws Exception {
+ Dispatch<Source> dispatch = createDispatch("/auth", new
HTTPClientPolicy());
+ AtomicInteger authorizations = new AtomicInteger();
+
((HTTPConduit)((DispatchImpl<?>)dispatch).getClient().getConduit()).setAuthSupplier(new
HttpAuthSupplier() {
+ @Override
+ public boolean requiresRequestCaching() {
+ return true;
+ }
+
+ @Override
+ public String getAuthorization(AuthorizationPolicy policy, URI
uri, Message message, String header) {
+ if (header == null) {
+ return null;
+ }
+ authorizations.incrementAndGet();
+ return "Basic dXNlcjpwYXNz";
+ }
+ });
+ try {
+ assertResponse(invokeAsync(dispatch, REQUEST));
+ assertRequests(REQUEST, 2);
+ assertEquals(1, CHALLENGES.get());
+ assertEquals(1, authorizations.get());
+ } finally {
+ destroy(dispatch);
+ }
+ }
+
+ @Test
+ public void testRedirectLoopCompletesExceptionally() throws Exception {
+ HTTPClientPolicy policy = new HTTPClientPolicy();
+ policy.setAutoRedirect(true);
+ Dispatch<Source> dispatch = createDispatch("/loop", policy);
+ try {
+ Response<Source> response = invokeAsync(dispatch, REQUEST);
+ try {
+ response.get(30, TimeUnit.SECONDS);
+ fail("A redirect loop must fail the asynchronous response");
+ } catch (ExecutionException ex) {
+ Throwable cause = ex.getCause();
+ while (cause != null && !(cause instanceof IOException)) {
+ cause = cause.getCause();
+ }
+ assertNotNull("Expected an IOException for the redirect loop",
cause);
+ assertTrue(cause.getMessage(),
cause.getMessage().contains("Redirect loop"));
+ }
+ } finally {
+ destroy(dispatch);
+ }
+ }
+
+ private void invokeAsync(String path, HTTPClientPolicy policy) throws
Exception {
+ Dispatch<Source> dispatch = createDispatch(path, policy);
+ try {
+ assertResponse(invokeAsync(dispatch, REQUEST));
+ assertRequests(REQUEST, "/redirect".equals(path) ? 2 : 1);
+ } finally {
+ destroy(dispatch);
+ }
+ }
+
+ private Dispatch<Source> createDispatch(String path, HTTPClientPolicy
policy) {
+ Service service = Service.create(SERVICE);
+ service.addPort(PORT_NAME, SOAPBinding.SOAP11HTTP_BINDING,
address(path));
+ Dispatch<Source> dispatch = service.createDispatch(PORT_NAME,
Source.class, Service.Mode.PAYLOAD);
+ dispatch.getRequestContext().put(AsyncHTTPConduit.USE_ASYNC,
"hc5".equals(conduit));
+ dispatch.getRequestContext().put("force.urlconnection.http.conduit",
"URLConnection".equals(conduit));
+ policy.setReceiveTimeout(30000);
+
((HTTPConduit)((DispatchImpl<?>)dispatch).getClient().getConduit()).setClient(policy);
+ return dispatch;
+ }
+
+ private static Response<Source> invokeAsync(Dispatch<Source> dispatch,
String payload) {
+ Response<Source> response = dispatch.invokeAsync(new StreamSource(new
StringReader(payload)));
+ assertFalse("The caller thread waited for the response",
RESPONDED.get());
+ release.countDown();
+ return response;
+ }
+
+ private static void assertResponse(Response<Source> response) throws
Exception {
+ String result = StaxUtils.toString(response.get(30, TimeUnit.SECONDS));
+ assertTrue(result, result.contains(">hello<"));
+ }
+
+ private static void assertRequests(String payload, int count) throws
Exception {
+ byte[] original = REQUESTS.poll(5, TimeUnit.SECONDS);
+ assertNotNull("No request received", original);
+ String expected = StaxUtils.read(new
StringReader(payload)).getDocumentElement().getTextContent();
+ assertEquals(expected, StaxUtils.read(new
ByteArrayInputStream(original))
+ .getElementsByTagNameNS("urn:cxf7232",
"echo").item(0).getTextContent());
+ for (int i = 1; i < count; i++) {
+ byte[] replay = REQUESTS.poll(5, TimeUnit.SECONDS);
+ assertNotNull("No retransmitted request received", replay);
+ assertArrayEquals("Retransmission changed the request body",
original, replay);
+ }
+ assertTrue("Unexpected extra request", REQUESTS.isEmpty());
+ }
+
+ private static void destroy(Dispatch<Source> dispatch) {
+ ((DispatchImpl<?>)dispatch).getClient().destroy();
+ }
+
+ private static String address(String path) {
+ return "http://localhost:"; + PORT + path;
+ }
+
+ private static void awaitRelease() {
+ try {
+ release.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private static void respond(HttpExchange ex) throws IOException {
+ RESPONDED.set(true);
+ byte[] body = RESPONSE.getBytes(StandardCharsets.UTF_8);
+ ex.getResponseHeaders().add("Content-Type", "text/xml; charset=UTF-8");
+ ex.sendResponseHeaders(200, body.length);
+ try (OutputStream os = ex.getResponseBody()) {
+ os.write(body);
+ }
+ }
+
+ private static void captureRequest(HttpExchange ex) throws IOException {
+ try (InputStream is = ex.getRequestBody()) {
+ REQUESTS.add(is.readAllBytes());
+ }
+ }
+}
diff --git
a/systests/transports/src/test/java/org/apache/cxf/systest/http/AsyncRetransmitTest.java
b/systests/transports/src/test/java/org/apache/cxf/systest/http/AsyncRetransmitTest.java
new file mode 100644
index 00000000000..8a2d00bce60
--- /dev/null
+++
b/systests/transports/src/test/java/org/apache/cxf/systest/http/AsyncRetransmitTest.java
@@ -0,0 +1,368 @@
+/**
+ * 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.cxf.systest.http;
+
+import java.io.ByteArrayInputStream;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.StringReader;
+import java.net.InetSocketAddress;
+import java.net.URI;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import javax.xml.namespace.QName;
+import javax.xml.transform.Source;
+import javax.xml.transform.stream.StreamSource;
+
+import com.sun.net.httpserver.HttpExchange;
+import com.sun.net.httpserver.HttpServer;
+
+import jakarta.xml.ws.Dispatch;
+import jakarta.xml.ws.Response;
+import jakarta.xml.ws.Service;
+import jakarta.xml.ws.soap.SOAPBinding;
+import org.apache.cxf.Bus;
+import org.apache.cxf.BusFactory;
+import org.apache.cxf.configuration.security.AuthorizationPolicy;
+import org.apache.cxf.io.CachedConstants;
+import org.apache.cxf.jaxws.DispatchImpl;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.staxutils.StaxUtils;
+import org.apache.cxf.testutil.common.TestUtil;
+import org.apache.cxf.transport.http.HTTPConduit;
+import org.apache.cxf.transport.http.asyncclient.AsyncHTTPConduit;
+import org.apache.cxf.transport.http.auth.HttpAuthSupplier;
+import org.apache.cxf.transports.http.configuration.HTTPClientPolicy;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.junit.runners.Parameterized.Parameters;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+/**
+ * CXF-7232: an asynchronous call must not block the caller thread until the
response is received,
+ * when the request is cached to be retransmitted on a redirect or with
MaxRetransmits.
+ */
+@RunWith(Parameterized.class)
+public class AsyncRetransmitTest {
+ public static final String PORT =
TestUtil.getPortNumber(AsyncRetransmitTest.class);
+ private static final QName SERVICE = new QName("urn:cxf7232",
"EchoService");
+ private static final QName PORT_NAME = new QName("urn:cxf7232",
"EchoPort");
+ private static final String REQUEST = "<echo
xmlns=\"urn:cxf7232\">hello</echo>";
+ private static final String RESPONSE =
+ "<soap:Envelope
xmlns:soap=\"http://schemas.xmlsoap.org/soap/envelope/\";><soap:Body>"
+ + "<echoResponse
xmlns=\"urn:cxf7232\">hello</echoResponse></soap:Body></soap:Envelope>";
+
+ private static HttpServer server;
+ private static ExecutorService executor;
+ private static final AtomicInteger REDIRECTS = new AtomicInteger();
+ private static final AtomicBoolean RESPONDED = new AtomicBoolean();
+ private static volatile CountDownLatch release;
+ private static final BlockingQueue<byte[]> REQUESTS = new
LinkedBlockingQueue<>();
+ private static final AtomicInteger CHALLENGES = new AtomicInteger();
+
+ @Rule
+ public TemporaryFolder cacheFolder = new TemporaryFolder();
+
+ private final String conduit;
+
+ public AsyncRetransmitTest(String conduit) {
+ this.conduit = conduit;
+ }
+
+ @Parameters(name = "{0}")
+ public static Collection<Object[]> data() {
+ return Arrays.asList(new Object[][] {{"HttpClient"},
{"URLConnection"}, {"hc"}});
+ }
+
+ @BeforeClass
+ public static void startServer() throws IOException {
+ executor = Executors.newCachedThreadPool();
+ server = HttpServer.create(new InetSocketAddress("localhost",
Integer.parseInt(PORT)), 0);
+ server.setExecutor(executor);
+ // redirects at once
+ server.createContext("/redirect", ex -> {
+ captureRequest(ex);
+ REDIRECTS.incrementAndGet();
+ ex.getResponseHeaders().add("Location", address("/echo"));
+ ex.sendResponseHeaders(307, -1);
+ ex.close();
+ });
+ // answers when the test releases it
+ server.createContext("/echo", ex -> {
+ captureRequest(ex);
+ awaitRelease();
+ respond(ex);
+ });
+ server.createContext("/auth", ex -> {
+ captureRequest(ex);
+ awaitRelease();
+ if ("Basic
dXNlcjpwYXNz".equals(ex.getRequestHeaders().getFirst("Authorization"))) {
+ respond(ex);
+ } else {
+ CHALLENGES.incrementAndGet();
+ RESPONDED.set(true);
+ ex.getResponseHeaders().add("WWW-Authenticate", "Basic
realm=\"cxf7232\"");
+ ex.sendResponseHeaders(401, -1);
+ ex.close();
+ }
+ });
+ server.createContext("/loop", ex -> {
+ captureRequest(ex);
+ awaitRelease();
+ RESPONDED.set(true);
+ ex.getResponseHeaders().add("Location", address("/loop"));
+ ex.sendResponseHeaders(307, -1);
+ ex.close();
+ });
+ server.start();
+ }
+
+ @AfterClass
+ public static void stopServer() {
+ server.stop(0);
+ executor.shutdownNow();
+ }
+
+ @Before
+ public void reset() {
+ REDIRECTS.set(0);
+ CHALLENGES.set(0);
+ REQUESTS.clear();
+ RESPONDED.set(false);
+ release = new CountDownLatch(1);
+ }
+
+ @After
+ public void releaseServer() {
+ release.countDown();
+ }
+
+ @Test
+ public void testAutoRedirect() throws Exception {
+ HTTPClientPolicy policy = new HTTPClientPolicy();
+ policy.setAutoRedirect(true);
+ invokeAsync("/redirect", policy);
+ assertEquals(1, REDIRECTS.get());
+ }
+
+ @Test
+ public void testAutoRedirectWithoutRedirect() throws Exception {
+ HTTPClientPolicy policy = new HTTPClientPolicy();
+ policy.setAutoRedirect(true);
+ invokeAsync("/echo", policy);
+ assertEquals(0, REDIRECTS.get());
+ }
+
+ @Test
+ public void testMaxRetransmits() throws Exception {
+ HTTPClientPolicy policy = new HTTPClientPolicy();
+ policy.setMaxRetransmits(1);
+ invokeAsync("/echo", policy);
+ assertEquals(0, REDIRECTS.get());
+ }
+
+ @Test
+ public void testRedirectWithDiskCachedRequest() throws Exception {
+ String payload = "<echo xmlns=\"urn:cxf7232\">" +
"0123456789abcdef".repeat(16384) + "</echo>";
+ File directory = cacheFolder.newFolder("request-cache");
+ Bus bus = BusFactory.getThreadDefaultBus();
+ Object oldDirectory =
bus.getProperty(CachedConstants.OUTPUT_DIRECTORY_BUS_PROP);
+ Object oldThreshold =
bus.getProperty(CachedConstants.THRESHOLD_BUS_PROP);
+ bus.setProperty(CachedConstants.OUTPUT_DIRECTORY_BUS_PROP,
directory.getAbsolutePath());
+ bus.setProperty(CachedConstants.THRESHOLD_BUS_PROP, "1024");
+ HTTPClientPolicy policy = new HTTPClientPolicy();
+ policy.setAutoRedirect(true);
+ Dispatch<Source> dispatch = createDispatch("/redirect", policy);
+ try {
+ Response<Source> response = dispatch.invokeAsync(new
StreamSource(new StringReader(payload)));
+ assertFalse("The caller thread waited for the response",
RESPONDED.get());
+ File[] files = directory.listFiles();
+ assertNotNull(files);
+ assertTrue("The request was not cached on disk", files.length > 0);
+ release.countDown();
+ assertResponse(response);
+ assertRequests(payload, 2);
+ assertEquals(1, REDIRECTS.get());
+ } finally {
+ release.countDown();
+ destroy(dispatch);
+ bus.setProperty(CachedConstants.OUTPUT_DIRECTORY_BUS_PROP,
oldDirectory);
+ bus.setProperty(CachedConstants.THRESHOLD_BUS_PROP, oldThreshold);
+ }
+ }
+
+ @Test
+ public void testAuthenticationChallenge() throws Exception {
+ Dispatch<Source> dispatch = createDispatch("/auth", new
HTTPClientPolicy());
+ AtomicInteger authorizations = new AtomicInteger();
+
((HTTPConduit)((DispatchImpl<?>)dispatch).getClient().getConduit()).setAuthSupplier(new
HttpAuthSupplier() {
+ @Override
+ public boolean requiresRequestCaching() {
+ return true;
+ }
+
+ @Override
+ public String getAuthorization(AuthorizationPolicy policy, URI
uri, Message message, String header) {
+ if (header == null) {
+ return null;
+ }
+ authorizations.incrementAndGet();
+ return "Basic dXNlcjpwYXNz";
+ }
+ });
+ try {
+ assertResponse(invokeAsync(dispatch, REQUEST));
+ assertRequests(REQUEST, 2);
+ assertEquals(1, CHALLENGES.get());
+ assertEquals(1, authorizations.get());
+ } finally {
+ destroy(dispatch);
+ }
+ }
+
+ @Test
+ public void testRedirectLoopCompletesExceptionally() throws Exception {
+ HTTPClientPolicy policy = new HTTPClientPolicy();
+ policy.setAutoRedirect(true);
+ Dispatch<Source> dispatch = createDispatch("/loop", policy);
+ try {
+ Response<Source> response = invokeAsync(dispatch, REQUEST);
+ try {
+ response.get(30, TimeUnit.SECONDS);
+ fail("A redirect loop must fail the asynchronous response");
+ } catch (ExecutionException ex) {
+ Throwable cause = ex.getCause();
+ while (cause != null && !(cause instanceof IOException)) {
+ cause = cause.getCause();
+ }
+ assertNotNull("Expected an IOException for the redirect loop",
cause);
+ assertTrue(cause.getMessage(),
cause.getMessage().contains("Redirect loop"));
+ }
+ } finally {
+ destroy(dispatch);
+ }
+ }
+
+ private void invokeAsync(String path, HTTPClientPolicy policy) throws
Exception {
+ Dispatch<Source> dispatch = createDispatch(path, policy);
+ try {
+ assertResponse(invokeAsync(dispatch, REQUEST));
+ assertRequests(REQUEST, "/redirect".equals(path) ? 2 : 1);
+ } finally {
+ destroy(dispatch);
+ }
+ }
+
+ private Dispatch<Source> createDispatch(String path, HTTPClientPolicy
policy) {
+ Service service = Service.create(SERVICE);
+ service.addPort(PORT_NAME, SOAPBinding.SOAP11HTTP_BINDING,
address(path));
+ Dispatch<Source> dispatch = service.createDispatch(PORT_NAME,
Source.class, Service.Mode.PAYLOAD);
+ dispatch.getRequestContext().put(AsyncHTTPConduit.USE_ASYNC,
"hc".equals(conduit));
+ dispatch.getRequestContext().put("force.urlconnection.http.conduit",
"URLConnection".equals(conduit));
+ policy.setReceiveTimeout(30000);
+
((HTTPConduit)((DispatchImpl<?>)dispatch).getClient().getConduit()).setClient(policy);
+ return dispatch;
+ }
+
+ private static Response<Source> invokeAsync(Dispatch<Source> dispatch,
String payload) {
+ Response<Source> response = dispatch.invokeAsync(new StreamSource(new
StringReader(payload)));
+ assertFalse("The caller thread waited for the response",
RESPONDED.get());
+ release.countDown();
+ return response;
+ }
+
+ private static void assertResponse(Response<Source> response) throws
Exception {
+ String result = StaxUtils.toString(response.get(30, TimeUnit.SECONDS));
+ assertTrue(result, result.contains(">hello<"));
+ }
+
+ private static void assertRequests(String payload, int count) throws
Exception {
+ byte[] original = REQUESTS.poll(5, TimeUnit.SECONDS);
+ assertNotNull("No request received", original);
+ String expected = StaxUtils.read(new
StringReader(payload)).getDocumentElement().getTextContent();
+ assertEquals(expected, StaxUtils.read(new
ByteArrayInputStream(original))
+ .getElementsByTagNameNS("urn:cxf7232",
"echo").item(0).getTextContent());
+ for (int i = 1; i < count; i++) {
+ byte[] replay = REQUESTS.poll(5, TimeUnit.SECONDS);
+ assertNotNull("No retransmitted request received", replay);
+ assertArrayEquals("Retransmission changed the request body",
original, replay);
+ }
+ assertTrue("Unexpected extra request", REQUESTS.isEmpty());
+ }
+
+ private static void destroy(Dispatch<Source> dispatch) {
+ ((DispatchImpl<?>)dispatch).getClient().destroy();
+ }
+
+ private static String address(String path) {
+ return "http://localhost:"; + PORT + path;
+ }
+
+ private static void awaitRelease() {
+ try {
+ release.await(10, TimeUnit.SECONDS);
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ }
+ }
+
+ private static void respond(HttpExchange ex) throws IOException {
+ RESPONDED.set(true);
+ byte[] body = RESPONSE.getBytes(StandardCharsets.UTF_8);
+ ex.getResponseHeaders().add("Content-Type", "text/xml; charset=UTF-8");
+ ex.sendResponseHeaders(200, body.length);
+ try (OutputStream os = ex.getResponseBody()) {
+ os.write(body);
+ }
+ }
+
+ private static void captureRequest(HttpExchange ex) throws IOException {
+ try (InputStream is = ex.getRequestBody()) {
+ REQUESTS.add(is.readAllBytes());
+ }
+ }
+}
