This is an automated email from the ASF dual-hosted git repository.
ffang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/cxf.git
The following commit(s) were added to refs/heads/main by this push:
new c4586170790 CXF-9135: Send the body of asynchronous non-chunked
requests through … (#3640)
c4586170790 is described below
commit c4586170790aaec6a83c6ca3d3c2a1a9eb419d8d
Author: Freeman(Yue) Fang <[email protected]>
AuthorDate: Fri Oct 30 09:13:35 2026 -0400
CXF-9135: Send the body of asynchronous non-chunked requests through …
(#3640)
* CXF-9135: Send the body of asynchronous non-chunked requests through the
HttpClient conduit
With chunking disallowed, HttpClientHTTPConduit buffers the request and
publishes it with a Content-Length. For an asynchronous call the caller
closes the output stream without waiting for the HttpClient, so the
publisher could be closed before the HttpClient asked for the content
length and subscribed, and the request went out chunked without a body.
The publisher now keeps the buffered body itself, so contentLength()
and subscribe() see the complete body however late they are called.
* address review feedback
* Do not hang the trust decision of a non-chunked HTTPS request
---
.../cxf/transport/http/HttpClientHTTPConduit.java | 86 ++++---
.../http/HttpClientSizedBodyPublisherTest.java | 275 +++++++++++++++++++++
.../http/HttpClientAsyncNoChunkingTest.java | 211 ++++++++++++++++
.../apache/cxf/systest/http2_jetty/BookStore.java | 11 +
.../http2_jetty/JettyClientServerHttp2Test.java | 53 ++++
.../systest/https/conduit/HTTPSConduitTest.java | 64 +++++
6 files changed, 663 insertions(+), 37 deletions(-)
diff --git
a/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HttpClientHTTPConduit.java
b/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HttpClientHTTPConduit.java
index a3960acd60f..f059a8128d6 100644
---
a/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HttpClientHTTPConduit.java
+++
b/rt/transports/http/src/main/java/org/apache/cxf/transport/http/HttpClientHTTPConduit.java
@@ -695,69 +695,70 @@ public class HttpClientHTTPConduit extends
URLConnectionHTTPConduit {
* so the content length becomes known (sized). It is used when the
chunked transfer is not allowed
* but the content length is not specified up-front.
*/
- private static final class HttpClientSizedBodyPublisher implements
CloseableBodyPublisher {
- private HTTPClientPolicy csPolicy;
- private Supplier<ByteArrayInputStream> pin;
+ static final class HttpClientSizedBodyPublisher implements
CloseableBodyPublisher {
+ private final HTTPClientPolicy csPolicy;
+ // Kept when the stream is closed: the caller of an asynchronous
request closes the stream
+ // without waiting, possibly before the HttpClient has read the
content length and the body
+ private final CloseableByteArrayOutputStream body;
private HttpClientWrappedOutputStream stream;
- private long contentLen;
- private HttpClientSizedBodyPublisher(HttpClientWrappedOutputStream s,
HTTPClientPolicy cs,
- Supplier<ByteArrayInputStream> pin) {
+ HttpClientSizedBodyPublisher(HttpClientWrappedOutputStream s,
HTTPClientPolicy cs,
+ CloseableByteArrayOutputStream body) {
this.stream = s;
this.csPolicy = cs;
- this.pin = pin;
+ this.body = body;
}
public synchronized void close() {
- if (stream != null) {
- contentLen = stream.contentLen;
- stream = null;
- }
+ stream = null;
}
@Override
public synchronized void subscribe(Subscriber<? super ByteBuffer>
subscriber) {
+ // The stream is only kept to signal that the connection is
complete, which
+ // getHttpsURLConnectionInfo() waits for. It is null when the
caller of an
+ // asynchronous request already closed it, so connectionComplete
is not set then
if (stream != null) {
stream.connectionComplete = true;
- if (stream.pout != null) {
- synchronized (stream.pout) {
- stream.pout.notifyAll();
- }
-
- BodyPublishers.ofInputStream(pin).subscribe(subscriber);
- stream = null;
- pin = null;
- return;
- }
}
- BodyPublishers.noBody().subscribe(subscriber);
+
BodyPublishers.ofInputStream(body::getInputStream).subscribe(subscriber);
}
@Override
public long contentLength() {
- if (stream != null && stream.pout != null) {
- final CloseableByteArrayOutputStream baos =
(CloseableByteArrayOutputStream) stream.pout;
-
- try {
- synchronized (baos) {
- if (!baos.closed) {
- baos.wait(csPolicy.getConnectionTimeout());
- }
+ // The HttpClient is connected by now, so a trust decision can go
ahead and write the body
+ synchronized (this) {
+ if (stream != null) {
+ stream.connectionComplete = true;
+ }
+ }
+ // The request is sent on the first write, so wait until the whole
body is written
+ long timeout = csPolicy.getConnectionTimeout();
+ long deadline = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(timeout);
+ synchronized (body) {
+ while (!body.closed) {
+ long remaining = TimeUnit.NANOSECONDS.toMillis(deadline -
System.nanoTime());
+ if (timeout > 0 && remaining <= 0) {
+ // Fail the request rather than send it with the
length of a partial body
+ throw new IllegalStateException("The request body was
not written within "
+ + timeout + " ms");
+ }
+ try {
+ body.wait(timeout > 0 ? remaining : 0);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("Interrupted while
waiting for the request body", e);
}
- contentLen = (int) baos.size();
- } catch (InterruptedException e) {
- //ignore
}
+ return body.size();
}
-
- return contentLen;
}
}
/**
* The {@link ByteArrayOutputStream} implementation that tracks the
closeability state.
*/
- private static final class CloseableByteArrayOutputStream extends
ByteArrayOutputStream {
+ static final class CloseableByteArrayOutputStream extends
ByteArrayOutputStream {
private boolean closed;
/**
@@ -1016,7 +1017,7 @@ public class HttpClientHTTPConduit extends
URLConnectionHTTPConduit {
// buffer the request body stream until it is fully flushed by
the client and only
// than send the request.
final CloseableByteArrayOutputStream baos = new
CloseableByteArrayOutputStream();
- this.publisher = new HttpClientSizedBodyPublisher(this,
csPolicy, baos::getInputStream);
+ this.publisher = new HttpClientSizedBodyPublisher(this,
csPolicy, baos);
pout = baos;
}
@@ -1276,6 +1277,17 @@ public class HttpClientHTTPConduit extends
URLConnectionHTTPConduit {
HttpClient cl = outMessage.get(HttpClient.class);
while (!connectionComplete ||
!cl.sslContext().getClientSessionContext().getIds().hasMoreElements()) {
+ if (future.isCompletedExceptionally()) {
+ // The request failed before the connection was complete,
so it never will be
+ try {
+ future.get();
+ } catch (InterruptedException | ExecutionException e) {
+ if (e.getCause() instanceof IOException) {
+ throw new Fault("Could not send Message.", LOG,
(IOException)e.getCause());
+ }
+ throw new IOException("Could not send Message.",
e.getCause());
+ }
+ }
Thread.yield();
}
byte[] key =
cl.sslContext().getClientSessionContext().getIds().nextElement();
diff --git
a/rt/transports/http/src/test/java/org/apache/cxf/transport/http/HttpClientSizedBodyPublisherTest.java
b/rt/transports/http/src/test/java/org/apache/cxf/transport/http/HttpClientSizedBodyPublisherTest.java
new file mode 100644
index 00000000000..ae0f7d5cfdd
--- /dev/null
+++
b/rt/transports/http/src/test/java/org/apache/cxf/transport/http/HttpClientSizedBodyPublisherTest.java
@@ -0,0 +1,275 @@
+/**
+ * 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.transport.http;
+
+import java.io.ByteArrayOutputStream;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Flow;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import
org.apache.cxf.transport.http.HttpClientHTTPConduit.CloseableByteArrayOutputStream;
+import
org.apache.cxf.transport.http.HttpClientHTTPConduit.HttpClientSizedBodyPublisher;
+import
org.apache.cxf.transport.http.HttpClientHTTPConduit.HttpClientWrappedOutputStream;
+import org.apache.cxf.transports.http.configuration.HTTPClientPolicy;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+
+/**
+ * CXF-9135: the caller of an asynchronous request closes the stream without
waiting for the HttpClient,
+ * which reads the content length and subscribes to the body on its own
thread. Each test drives one
+ * ordering of the two threads deterministically.
+ */
+public class HttpClientSizedBodyPublisherTest {
+ // The size of the request in the reporter's reproducer: "Too few bytes
returned by the publisher (0/206)"
+ private static final int REQUEST_SIZE = 206;
+
+ /**
+ * The ordering of the reporter's reproducer: the HttpClient asks for the
content length while the
+ * request is being written and subscribes only after the caller closed
the stream.
+ */
+ @Test
+ public void testContentLengthBeforeCloseAndSubscriptionAfterClose() throws
Exception {
+ byte[] expected = body(REQUEST_SIZE);
+ CloseableByteArrayOutputStream body = new
CloseableByteArrayOutputStream();
+ HttpClientWrappedOutputStream stream = stream(body);
+ HttpClientSizedBodyPublisher publisher = publisher(stream, body);
+
+ ExecutorService httpClient = Executors.newSingleThreadExecutor();
+ try {
+ CompletableFuture<Thread> httpClientThread = new
CompletableFuture<>();
+ Future<Long> contentLength = httpClient.submit(() -> {
+ httpClientThread.complete(Thread.currentThread());
+ return publisher.contentLength();
+ });
+ awaitWaiting(httpClientThread.get(5, TimeUnit.SECONDS));
+ assertFalse(contentLength.isDone());
+
+ body.write(expected);
+ closeLikeTheCaller(stream, body, publisher);
+
+ assertEquals(REQUEST_SIZE, contentLength.get(5,
TimeUnit.SECONDS).longValue());
+ assertArrayEquals(expected, readBody(publisher));
+ } finally {
+ httpClient.shutdownNow();
+ }
+ }
+
+ /** The caller closes the stream before the HttpClient reads the content
length and the body. */
+ @Test
+ public void testContentLengthAndSubscriptionAfterClose() throws Exception {
+ byte[] expected = body(REQUEST_SIZE);
+ CloseableByteArrayOutputStream body = new
CloseableByteArrayOutputStream();
+ HttpClientWrappedOutputStream stream = stream(body);
+ HttpClientSizedBodyPublisher publisher = publisher(stream, body);
+ body.write(expected);
+ closeLikeTheCaller(stream, body, publisher);
+
+ assertEquals(REQUEST_SIZE, publisher.contentLength());
+ assertArrayEquals(expected, readBody(publisher));
+ }
+
+ /** The HttpClient may subscribe again, for instance to retry the request.
*/
+ @Test
+ public void testRepeatedSubscriptionsAfterClose() throws Exception {
+ byte[] expected = body(128 * 1024);
+ CloseableByteArrayOutputStream body = new
CloseableByteArrayOutputStream();
+ HttpClientWrappedOutputStream stream = stream(body);
+ HttpClientSizedBodyPublisher publisher = publisher(stream, body);
+ body.write(expected);
+ closeLikeTheCaller(stream, body, publisher);
+
+ assertEquals(expected.length, publisher.contentLength());
+ assertArrayEquals(expected, readBody(publisher));
+ assertArrayEquals(expected, readBody(publisher));
+ }
+
+ @Test
+ public void testEmptyBodyAfterClose() throws Exception {
+ CloseableByteArrayOutputStream body = new
CloseableByteArrayOutputStream();
+ HttpClientWrappedOutputStream stream = stream(body);
+ HttpClientSizedBodyPublisher publisher = publisher(stream, body);
+ closeLikeTheCaller(stream, body, publisher);
+
+ assertEquals(0, publisher.contentLength());
+ assertArrayEquals(new byte[0], readBody(publisher));
+ }
+
+ /** Waking up the HttpClient thread without closing the stream must not
return a partial length. */
+ @Test
+ public void testContentLengthWaitsAfterWakeupWithoutClose() throws
Exception {
+ byte[] expected = body(REQUEST_SIZE);
+ CloseableByteArrayOutputStream body = new
CloseableByteArrayOutputStream();
+ HttpClientWrappedOutputStream stream = stream(body);
+ HttpClientSizedBodyPublisher publisher = publisher(stream, body);
+
+ ExecutorService httpClient = Executors.newSingleThreadExecutor();
+ try {
+ CompletableFuture<Thread> httpClientThread = new
CompletableFuture<>();
+ Future<Long> contentLength = httpClient.submit(() -> {
+ httpClientThread.complete(Thread.currentThread());
+ return publisher.contentLength();
+ });
+ awaitWaiting(httpClientThread.get(5, TimeUnit.SECONDS));
+ body.write(expected, 0, 10);
+ synchronized (body) {
+ body.notifyAll();
+ }
+ // let the HttpClient thread wake up, it has to wait again
+ Thread.sleep(100);
+ awaitWaiting(httpClientThread.get());
+ assertFalse(contentLength.isDone());
+
+ body.write(expected, 10, expected.length - 10);
+ closeLikeTheCaller(stream, body, publisher);
+
+ assertEquals(REQUEST_SIZE, contentLength.get(5,
TimeUnit.SECONDS).longValue());
+ assertArrayEquals(expected, readBody(publisher));
+ } finally {
+ httpClient.shutdownNow();
+ }
+ }
+
+ /** A body which is not written within the timeout fails the request
instead of sending a partial one. */
+ @Test
+ public void testContentLengthFailsWhenTheBodyIsNotWrittenInTime() throws
Exception {
+ CloseableByteArrayOutputStream body = new
CloseableByteArrayOutputStream();
+ HttpClientSizedBodyPublisher publisher = publisher(stream(body), body,
200);
+ body.write(body(REQUEST_SIZE), 0, 10);
+
+ long start = System.nanoTime();
+ IllegalStateException ex = assertThrows(IllegalStateException.class,
publisher::contentLength);
+ assertTrue(ex.getMessage(), ex.getMessage().contains("200 ms"));
+ assertTrue(System.nanoTime() - start >=
TimeUnit.MILLISECONDS.toNanos(190));
+ }
+
+ @Test
+ public void testContentLengthFailsWhenInterrupted() throws Exception {
+ CloseableByteArrayOutputStream body = new
CloseableByteArrayOutputStream();
+ HttpClientSizedBodyPublisher publisher = publisher(stream(body), body);
+
+ ExecutorService httpClient = Executors.newSingleThreadExecutor();
+ try {
+ CompletableFuture<Thread> httpClientThread = new
CompletableFuture<>();
+ Future<Boolean> interrupted = httpClient.submit(() -> {
+ httpClientThread.complete(Thread.currentThread());
+ try {
+ publisher.contentLength();
+ return false;
+ } catch (IllegalStateException e) {
+ return Thread.currentThread().isInterrupted();
+ }
+ });
+ Thread thread = httpClientThread.get(5, TimeUnit.SECONDS);
+ awaitWaiting(thread);
+ thread.interrupt();
+ assertTrue(interrupted.get(5, TimeUnit.SECONDS));
+ } finally {
+ httpClient.shutdownNow();
+ }
+ }
+
+ // The state of HttpClientWrappedOutputStream for a request of unknown
length without chunking
+ private static HttpClientWrappedOutputStream
stream(CloseableByteArrayOutputStream body) {
+ HttpClientWrappedOutputStream stream =
mock(HttpClientWrappedOutputStream.class);
+ stream.pout = body;
+ stream.contentLen = -1;
+ return stream;
+ }
+
+ private static HttpClientSizedBodyPublisher
publisher(HttpClientWrappedOutputStream stream,
+ CloseableByteArrayOutputStream body) {
+ return publisher(stream, body, 30000);
+ }
+
+ private static HttpClientSizedBodyPublisher
publisher(HttpClientWrappedOutputStream stream,
+ CloseableByteArrayOutputStream body, long connectionTimeout) {
+ HTTPClientPolicy policy = new HTTPClientPolicy();
+ policy.setConnectionTimeout(connectionTimeout);
+ return new HttpClientSizedBodyPublisher(stream, policy, body);
+ }
+
+ // As HttpClientWrappedOutputStream.close() does: close the buffer and
drop it, then close the publisher
+ private static void closeLikeTheCaller(HttpClientWrappedOutputStream
stream,
+ CloseableByteArrayOutputStream body, HttpClientSizedBodyPublisher
publisher) throws Exception {
+ body.close();
+ stream.pout = null;
+ publisher.close();
+ }
+
+ private static byte[] body(int size) {
+ byte[] request = "<soap:Envelope
xmlns:soap=\"http://schemas.xmlsoap.org/soap/envelope/\";>"
+ .getBytes(StandardCharsets.UTF_8);
+ byte[] body = Arrays.copyOf(request, size);
+ Arrays.fill(body, Math.min(request.length, size), size, (byte)' ');
+ return body;
+ }
+
+ // Wait until the HttpClient thread waits in contentLength() for the
request to be written
+ private static void awaitWaiting(Thread thread) throws Exception {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ while (thread.getState() != Thread.State.TIMED_WAITING
+ || Arrays.stream(thread.getStackTrace()).noneMatch(e ->
"contentLength".equals(e.getMethodName()))) {
+ assertTrue("contentLength() did not wait for the request",
System.nanoTime() < deadline);
+ Thread.sleep(10);
+ }
+ }
+
+ private static byte[] readBody(HttpClientSizedBodyPublisher publisher)
throws Exception {
+ CompletableFuture<byte[]> completed = new CompletableFuture<>();
+ ByteArrayOutputStream received = new ByteArrayOutputStream();
+ publisher.subscribe(new Flow.Subscriber<ByteBuffer>() {
+ @Override
+ public void onSubscribe(Flow.Subscription subscription) {
+ subscription.request(Long.MAX_VALUE);
+ }
+
+ @Override
+ public void onNext(ByteBuffer buffer) {
+ byte[] bytes = new byte[buffer.remaining()];
+ buffer.get(bytes);
+ received.writeBytes(bytes);
+ }
+
+ @Override
+ public void onError(Throwable error) {
+ completed.completeExceptionally(error);
+ }
+
+ @Override
+ public void onComplete() {
+ completed.complete(received.toByteArray());
+ }
+ });
+ return completed.get(5, TimeUnit.SECONDS);
+ }
+}
diff --git
a/systests/transports/src/test/java/org/apache/cxf/systest/http/HttpClientAsyncNoChunkingTest.java
b/systests/transports/src/test/java/org/apache/cxf/systest/http/HttpClientAsyncNoChunkingTest.java
new file mode 100644
index 00000000000..edfccb9f743
--- /dev/null
+++
b/systests/transports/src/test/java/org/apache/cxf/systest/http/HttpClientAsyncNoChunkingTest.java
@@ -0,0 +1,211 @@
+/**
+ * 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.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.PrintWriter;
+import java.io.Reader;
+import java.io.StringReader;
+import java.io.StringWriter;
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import javax.xml.namespace.QName;
+import javax.xml.transform.Source;
+import javax.xml.transform.stream.StreamSource;
+
+import jakarta.xml.ws.Dispatch;
+import jakarta.xml.ws.Service;
+import jakarta.xml.ws.WebServiceException;
+import jakarta.xml.ws.soap.SOAPBinding;
+import org.apache.cxf.jaxws.DispatchImpl;
+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.transports.http.configuration.HTTPClientPolicy;
+
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assume.assumeFalse;
+
+/**
+ * CXF-9135: an asynchronous call through the JDK HttpClient conduit with
chunking disallowed must send
+ * its whole body with a Content-Length, although the caller closes the
request before the HttpClient
+ * reads it.
+ */
+public class HttpClientAsyncNoChunkingTest {
+ public static final String PORT =
TestUtil.getPortNumber(HttpClientAsyncNoChunkingTest.class);
+ private static final QName SERVICE = new QName("urn:cxf9135",
"EchoService");
+ private static final QName PORT_NAME = new QName("urn:cxf9135",
"EchoPort");
+ private static final String REQUEST = "<echo
xmlns=\"urn:cxf9135\">hello</echo>";
+ private static final String RESPONSE =
+ "<soap:Envelope
xmlns:soap=\"http://schemas.xmlsoap.org/soap/envelope/\";><soap:Body>"
+ + "<echoResponse
xmlns=\"urn:cxf9135\">hello</echoResponse></soap:Body></soap:Envelope>";
+ private static final int CALLS = 10;
+
+ // the JDK server, not the HttpServer of this package
+ private static com.sun.net.httpserver.HttpServer server;
+ private static ExecutorService executor;
+ private static final List<String> RECEIVED = new CopyOnWriteArrayList<>();
+
+ @BeforeClass
+ public static void startServer() throws IOException {
+ executor = Executors.newCachedThreadPool();
+ server = com.sun.net.httpserver.HttpServer.create(
+ new InetSocketAddress("localhost", Integer.parseInt(PORT)), 0);
+ server.setExecutor(executor);
+ // records how each request was framed and how much body it carried
+ server.createContext("/echo", ex -> {
+ byte[] body;
+ try (InputStream is = ex.getRequestBody()) {
+ body = is.readAllBytes();
+ }
+ RECEIVED.add("Content-Length=" +
ex.getRequestHeaders().getFirst("Content-Length")
+ + ", Transfer-Encoding=" +
ex.getRequestHeaders().getFirst("Transfer-Encoding")
+ + ", body=" + new String(body,
StandardCharsets.UTF_8).contains(REQUEST));
+ byte[] response = RESPONSE.getBytes(StandardCharsets.UTF_8);
+ ex.getResponseHeaders().add("Content-Type", "text/xml;
charset=UTF-8");
+ ex.sendResponseHeaders(200, response.length);
+ try (OutputStream os = ex.getResponseBody()) {
+ os.write(response);
+ }
+ });
+ server.start();
+ }
+
+ @AfterClass
+ public static void stopServer() {
+ server.stop(0);
+ executor.shutdownNow();
+ }
+
+ @Before
+ public void reset() {
+ RECEIVED.clear();
+ }
+
+ @Test
+ public void testAsyncWithoutChunking() throws Exception {
+ Dispatch<Source> dispatch = createDispatch();
+ try {
+ for (int i = 0; i < CALLS; i++) {
+ Source response = dispatch.invokeAsync(new StreamSource(new
StringReader(REQUEST)))
+ .get(30, TimeUnit.SECONDS);
+ String result = StaxUtils.toString(response);
+ assertTrue(result, result.contains(">hello<"));
+ }
+ } finally {
+ ((DispatchImpl<?>)dispatch).getClient().destroy();
+ }
+ assertEquals(CALLS, RECEIVED.size());
+ for (String received : RECEIVED) {
+ assertFalse(received, received.startsWith("Content-Length=null"));
+ assertTrue(received, received.endsWith("Transfer-Encoding=null,
body=true"));
+ }
+ }
+
+ /**
+ * The request is sent on the first write, and the HttpClient waits up to
the connection timeout for
+ * the rest of the body to know its length. A body written more slowly
fails the call, rather than
+ * sending a Content-Length which does not match the body.
+ */
+ @Test
+ public void testWithoutChunkingFailsWhenTheBodyIsWrittenTooSlowly() throws
Exception {
+ // the URLConnection conduit buffers the whole body before it sends
the request
+
assumeFalse(Boolean.getBoolean("org.apache.cxf.transport.http.forceURLConnection"));
+ Dispatch<Source> dispatch = createDispatch();
+
((HTTPConduit)((DispatchImpl<?>)dispatch).getClient().getConduit()).getClient()
+ .setConnectionTimeout(500);
+ try {
+ WebServiceException ex = assertThrows(WebServiceException.class,
+ () -> dispatch.invoke(new StreamSource(new SlowReader())));
+ assertTrue(stackTraceOf(ex), stackTraceOf(ex).contains("The
request body was not written within 500 ms"));
+ } finally {
+ ((DispatchImpl<?>)dispatch).getClient().destroy();
+ }
+ assertTrue(RECEIVED.toString(), RECEIVED.isEmpty());
+ }
+
+ private static String stackTraceOf(Throwable t) {
+ StringWriter sw = new StringWriter();
+ t.printStackTrace(new PrintWriter(sw));
+ return sw.toString();
+ }
+
+ // A request larger than the writers' buffers, so the request is sent
before it is fully written,
+ // which pauses for longer than the connection timeout after it started
+ private static final class SlowReader extends Reader {
+ private final Reader reader = new StringReader("<echo
xmlns=\"urn:cxf9135\">"
+ + "x".repeat(256 * 1024) + "</echo>");
+ private int read;
+ private boolean paused;
+
+ @Override
+ public int read(char[] cbuf, int off, int len) throws IOException {
+ if (!paused && read >= 128 * 1024) {
+ paused = true;
+ try {
+ Thread.sleep(2000);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IOException(e);
+ }
+ }
+ int n = reader.read(cbuf, off, Math.min(len, 8192));
+ if (n > 0) {
+ read += n;
+ }
+ return n;
+ }
+
+ @Override
+ public void close() {
+ }
+ }
+
+ private static Dispatch<Source> createDispatch() {
+ Service service = Service.create(SERVICE);
+ service.addPort(PORT_NAME, SOAPBinding.SOAP11HTTP_BINDING,
+ "http://localhost:"; + PORT + "/echo");
+ Dispatch<Source> dispatch = service.createDispatch(PORT_NAME,
Source.class, Service.Mode.PAYLOAD);
+ // use the JDK HttpClient conduit, which is the default when no async
conduit is available
+ dispatch.getRequestContext().put(AsyncHTTPConduit.USE_ASYNC, false);
+ HTTPClientPolicy policy = new HTTPClientPolicy();
+ policy.setAllowChunking(false);
+ policy.setReceiveTimeout(30000);
+
((HTTPConduit)((DispatchImpl<?>)dispatch).getClient().getConduit()).setClient(policy);
+ return dispatch;
+ }
+}
diff --git
a/systests/transports/src/test/java/org/apache/cxf/systest/http2_jetty/BookStore.java
b/systests/transports/src/test/java/org/apache/cxf/systest/http2_jetty/BookStore.java
index 90a31eb64bb..52fcd88ad02 100644
---
a/systests/transports/src/test/java/org/apache/cxf/systest/http2_jetty/BookStore.java
+++
b/systests/transports/src/test/java/org/apache/cxf/systest/http2_jetty/BookStore.java
@@ -27,7 +27,10 @@ import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.ws.rs.Consumes;
import jakarta.ws.rs.GET;
+import jakarta.ws.rs.POST;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.Produces;
import jakarta.ws.rs.core.Context;
@@ -47,6 +50,14 @@ public class BookStore {
return "CXF in Action".getBytes();
}
+ @POST
+ @Path("/echo")
+ @Consumes("text/plain")
+ @Produces("text/plain")
+ public String echo(String body, @Context HttpServletRequest request) {
+ return request.getProtocol() + " " + body;
+ }
+
@GET
@Path("/redirect")
@Produces("application/xml")
diff --git
a/systests/transports/src/test/java/org/apache/cxf/systest/http2_jetty/JettyClientServerHttp2Test.java
b/systests/transports/src/test/java/org/apache/cxf/systest/http2_jetty/JettyClientServerHttp2Test.java
index 5790fa08699..e041a50d273 100644
---
a/systests/transports/src/test/java/org/apache/cxf/systest/http2_jetty/JettyClientServerHttp2Test.java
+++
b/systests/transports/src/test/java/org/apache/cxf/systest/http2_jetty/JettyClientServerHttp2Test.java
@@ -19,11 +19,25 @@
package org.apache.cxf.systest.http2_jetty;
+import jakarta.ws.rs.core.Response;
+import org.apache.cxf.configuration.jsse.TLSClientParameters;
+import org.apache.cxf.jaxrs.client.ClientConfiguration;
+import org.apache.cxf.jaxrs.client.WebClient;
import org.apache.cxf.jaxrs.model.AbstractResourceInfo;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.transport.http.HTTPConduit;
+import org.apache.cxf.transport.http.MessageTrustDecider;
+import org.apache.cxf.transport.http.URLConnectionInfo;
+import org.apache.cxf.transport.http.UntrustedURLConnectionIOException;
+import org.apache.cxf.transport.https.HttpsURLConnectionInfo;
+import org.apache.cxf.transport.https.InsecureTrustManager;
import org.junit.BeforeClass;
+import org.junit.Test;
+import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
+import static org.junit.Assume.assumeFalse;
public class JettyClientServerHttp2Test extends
AbstractJettyClientServerHttp2Test {
private static final String PORT = BookServerHttp2.PORT;
@@ -36,6 +50,45 @@ public class JettyClientServerHttp2Test extends
AbstractJettyClientServerHttp2Te
}
+ /**
+ * The trust decision is made after the request is sent and before its
body is written. Without
+ * chunking, the HttpClient waits for the whole body before it sends the
request, so it must let the
+ * trust decision go ahead, over HTTP/2 as well.
+ */
+ @Test(timeout = 60000)
+ public void testHttpsTrustWithoutChunkingHttp2() throws Exception {
+ // the URLConnection conduit does not support HTTP/2
+
assumeFalse(Boolean.getBoolean("org.apache.cxf.transport.http.forceURLConnection"));
+ final WebClient wc = WebClient
+ .create(getAddress() + getContext() + "/web/bookstore/echo")
+ .type("text/plain")
+ .accept("text/plain");
+
+ final ClientConfiguration config = WebClient.getConfig(wc);
+ config.getRequestContext().put(HTTPConduit.FORCE_HTTP_VERSION, "2");
+ final HTTPConduit conduit = config.getHttpConduit();
+ conduit.getClient().setAllowChunking(false);
+ TLSClientParameters params = new TLSClientParameters();
+
params.setTrustManagers(InsecureTrustManager.getNoOpX509TrustManagers());
+ params.setDisableCNCheck(true);
+ conduit.setTlsClientParameters(params);
+ conduit.setTrustDecider(new MessageTrustDecider() {
+ @Override
+ public void establishTrust(String conduitName, URLConnectionInfo
connectionInfo, Message message)
+ throws UntrustedURLConnectionIOException {
+ String peer =
((HttpsURLConnectionInfo)connectionInfo).getPeerPrincipal().getName();
+ if (!peer.contains("OU=Bethal")) {
+ throw new UntrustedURLConnectionIOException("Untrusted
peer " + peer);
+ }
+ }
+ });
+
+ try (Response resp = wc.post("CXF in Action")) {
+ assertEquals(200, resp.getStatus());
+ assertEquals("HTTP/2.0 CXF in Action",
resp.readEntity(String.class));
+ }
+ }
+
@Override
protected String getAddress() {
return "https://localhost:"; + PORT;
diff --git
a/systests/transports/src/test/java/org/apache/cxf/systest/https/conduit/HTTPSConduitTest.java
b/systests/transports/src/test/java/org/apache/cxf/systest/https/conduit/HTTPSConduitTest.java
index 52bd110d50f..efa05cf46ed 100644
---
a/systests/transports/src/test/java/org/apache/cxf/systest/https/conduit/HTTPSConduitTest.java
+++
b/systests/transports/src/test/java/org/apache/cxf/systest/https/conduit/HTTPSConduitTest.java
@@ -55,6 +55,7 @@ import javax.net.ssl.X509ExtendedTrustManager;
import javax.xml.namespace.QName;
import jakarta.xml.ws.BindingProvider;
+import jakarta.xml.ws.WebServiceException;
import org.apache.cxf.Bus;
import org.apache.cxf.BusFactory;
import org.apache.cxf.bus.spring.BusApplicationContext;
@@ -68,6 +69,7 @@ import org.apache.cxf.helpers.JavaUtils;
import org.apache.cxf.message.Message;
import org.apache.cxf.systest.https.HttpsBusServer;
import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
+import org.apache.cxf.testutil.common.TestUtil;
import org.apache.cxf.transport.http.HTTPConduit;
import org.apache.cxf.transport.http.MessageTrustDecider;
import org.apache.cxf.transport.http.URLConnectionInfo;
@@ -89,6 +91,7 @@ import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -118,6 +121,8 @@ import static org.junit.Assert.fail;
*/
public class HTTPSConduitTest extends AbstractBusClientServerTestBase {
private static final boolean IN_PROCESS = true;
+ // no server listens on this port
+ private static final String NO_SERVER_PORT =
TestUtil.getPortNumber(HTTPSConduitTest.class);
private static TLSClientParameters tlsClientParameters = new
TLSClientParameters();
private static List<String> servers = new ArrayList<>();
@@ -652,6 +657,65 @@ public class HTTPSConduitTest extends
AbstractBusClientServerTestBase {
assertProxyRequestCount(0);
}
+ /**
+ * The trust decision is made after the request is sent and before its
body is written. Without
+ * chunking, the HttpClient waits for the whole body before it sends the
request, so it must let the
+ * trust decision go ahead.
+ */
+ @Test(timeout = 60000)
+ public void testHttpsTrustWithoutChunking() throws Exception {
+ startServer("Bethal");
+
+ Greeter bethal = createBethal(getPort("PORT4"), false);
+ HTTPConduit http = (HTTPConduit)
ClientProxy.getClient(bethal).getConduit();
+
+ http.setTrustDecider(new MyHttpsTrustDecider("Bethal"));
+ assertEquals("Bonjour from Bethal", bethal.sayHi());
+
+ http.setTrustDecider(new MyHttpsTrustDecider("Nobody"));
+ assertThrows(WebServiceException.class, bethal::sayHi);
+ }
+
+ /**
+ * A request which fails before the connection is complete must fail the
call, rather than leave
+ * the trust decision waiting for the connection.
+ */
+ @Test(timeout = 60000)
+ public void testHttpsTrustConnectionRefused() throws Exception {
+ for (boolean allowChunking : new boolean[] {true, false}) {
+ Greeter bethal = createBethal(NO_SERVER_PORT, allowChunking);
+ HTTPConduit http = (HTTPConduit)
ClientProxy.getClient(bethal).getConduit();
+ http.setTrustDecider(new MyHttpsTrustDecider("Bethal"));
+
+ assertThrows(WebServiceException.class, bethal::sayHi);
+ }
+ }
+
+ private Greeter createBethal(String port, boolean allowChunking) throws
Exception {
+ URL wsdl = getClass().getResource("greeting.wsdl");
+ assertNotNull("WSDL is null", wsdl);
+
+ SOAPService service = new SOAPService(wsdl, serviceName);
+ Greeter bethal = service.getPort(bethalQ, Greeter.class);
+ updateAddressPort(bethal, port);
+
+ Client client = ClientProxy.getClient(bethal);
+ client.getRequestContext().put("share.httpclient.http.conduit", false);
+ HTTPConduit http = (HTTPConduit) client.getConduit();
+
+ HTTPClientPolicy httpClientPolicy = new HTTPClientPolicy();
+ httpClientPolicy.setAutoRedirect(false);
+ httpClientPolicy.setAllowChunking(allowChunking);
+ AuthorizationPolicy authPolicy = new AuthorizationPolicy();
+ authPolicy.setUserName("Betty");
+ authPolicy.setPassword("password");
+
+ http.setClient(httpClientPolicy);
+ http.setTlsClientParameters(tlsClientParameters);
+ http.setAuthorization(authPolicy);
+ return bethal;
+ }
+
@Test
public void testHttpsTrustRedirect() throws Exception {
startServer("Tarpin");
