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 52b62b9eaf7 CXF-9267: HttpClientHTTPConduit.HttpClientBodyPublisher
hides ConnectionTimeout Exception (#3660)
52b62b9eaf7 is described below
commit 52b62b9eaf79ec86b6a5d3f96632f23d3af66b5b
Author: Valentino Porta <[email protected]>
AuthorDate: Wed Nov 4 20:28:57 2026 +0100
CXF-9267: HttpClientHTTPConduit.HttpClientBodyPublisher hides
ConnectionTimeout Exception (#3660)
* Race condition CXF-9267 ...made pin available also if stream==null
* Race condition CXF-9267 ...add some tests.
testSubscribeAfterPublisherClose failed with the code before.
* Add comments
* Address issue of java.io.IOException: Pipe not connected in test where
http method was without body (GET/DELETE)
* Add one new http2 GET test with chunking enable to cover also this
scenario
---
.../cxf/transport/http/HttpClientHTTPConduit.java | 18 +-
.../http/HttpClientBodyPublisherTest.java | 319 +++++++++++++++++++++
.../apache/cxf/systest/http2_jetty/BookStore.java | 7 +
.../http2_jetty/JettyClientServerHttp2Test.java | 32 +++
4 files changed, 369 insertions(+), 7 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 1e217005bdb..8b068f77a20 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
@@ -653,12 +653,12 @@ public class HttpClientHTTPConduit extends
URLConnectionHTTPConduit {
/**
* The {@link BodyPublisher} that wraps around the output stream.
*/
- private static final class HttpClientBodyPublisher implements
CloseableBodyPublisher {
+ static final class HttpClientBodyPublisher implements
CloseableBodyPublisher {
private Supplier<InputStream> pin;
private HttpClientWrappedOutputStream stream;
private long contentLen;
- private HttpClientBodyPublisher(HttpClientWrappedOutputStream s,
Supplier<InputStream> pin) {
+ HttpClientBodyPublisher(HttpClientWrappedOutputStream s,
Supplier<InputStream> pin) {
this.stream = s;
this.pin = pin;
}
@@ -682,13 +682,17 @@ public class HttpClientHTTPConduit extends
URLConnectionHTTPConduit {
if (stream != null) {
contentLen = stream.contentLen;
}
- BodyPublishers.ofInputStream(pin).subscribe(subscriber);
stream = null;
- pin = null;
- return;
}
}
- BodyPublishers.noBody().subscribe(subscriber);
+ // See https://issues.apache.org/jira/browse/CXF-9267
+ if (pin != null) {
+ BodyPublishers.ofInputStream(pin).subscribe(subscriber);
+ pin = null;
+ } else {
+ // Avoiding java.io.IOException: Pipe closed if pin already
subscribed
+ BodyPublishers.noBody().subscribe(subscriber);
+ }
}
@Override
@@ -1018,7 +1022,7 @@ public class HttpClientHTTPConduit extends
URLConnectionHTTPConduit {
if (csPolicy.isAllowChunking() || contentLen >= 0) {
final PipedInputStream pin = new
PipedInputStream(csPolicy.getChunkLength() <= 0
? 4096 : csPolicy.getChunkLength());
- this.publisher = new HttpClientBodyPublisher(this, () -> pin);
+ this.publisher = new HttpClientBodyPublisher(this, contentLen
!= 0 ? () -> pin : null);
if (contentLen != 0) {
pout = new HttpClientPipedOutputStream(this, pin,
csPolicy, publisher);
}
diff --git
a/rt/transports/http/src/test/java/org/apache/cxf/transport/http/HttpClientBodyPublisherTest.java
b/rt/transports/http/src/test/java/org/apache/cxf/transport/http/HttpClientBodyPublisherTest.java
new file mode 100644
index 00000000000..6a993927eee
--- /dev/null
+++
b/rt/transports/http/src/test/java/org/apache/cxf/transport/http/HttpClientBodyPublisherTest.java
@@ -0,0 +1,319 @@
+/**
+ * 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.io.PipedInputStream;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Flow;
+import java.util.concurrent.TimeUnit;
+
+import
org.apache.cxf.transport.http.HttpClientHTTPConduit.HttpClientBodyPublisher;
+import
org.apache.cxf.transport.http.HttpClientHTTPConduit.HttpClientPipedOutputStream;
+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.assertTrue;
+import static org.mockito.Mockito.mock;
+
+/**
+ * See https://issues.apache.org/jira/browse/CXF-9267
+ */
+public class HttpClientBodyPublisherTest {
+
+ private static final int REQUEST_SIZE = 206;
+ private static final int CHUNK_THRESHOLD_SIZE = 4096;
+
+ private static final int AWAIT_TEST_TIMEOUT = 4;
+
+ /**
+ * The HTTP client subscribes and receives the complete request body.
+ */
+ @Test
+ public void testSubscribeBeforePublisherClose() throws Exception {
+ byte[] expected = body(REQUEST_SIZE);
+
+ TestPubPipeWrap pubStream = streams(expected.length);
+
+ pubStream.pipedOutputStream.write(expected);
+
+ /*
+ * We subscribe first
+ */
+ CountDownLatch countDownLatchBeforeSubscribe = new CountDownLatch(1);
+ CountDownLatch countDownLatchAfterSubscribe = new CountDownLatch(1);
+ // The HttpClient arrive before the close ... and subscribe...
+ // Using PipedInputStream as pin I use another thread to subscribe
+ CompletableFuture<byte[]> received = subscribeAsync(
+ pubStream.publisher,
+ countDownLatchBeforeSubscribe,
+ countDownLatchAfterSubscribe
+ );
+ // Wait to be subscribed
+ assertTrue(countDownLatchBeforeSubscribe.await(AWAIT_TEST_TIMEOUT,
TimeUnit.SECONDS));
+ assertTrue(countDownLatchAfterSubscribe.await(AWAIT_TEST_TIMEOUT,
TimeUnit.SECONDS));
+
+ /*
+ * ...and then close
+ */
+
+ // The caller close the stream
+ pubStream.wrappedOutputStream.close();
+ // Close also the publisher
+ pubStream.pipedOutputStream.close();
+
+ // Assert the body is the same
+ assertArrayEquals(expected, received.get(AWAIT_TEST_TIMEOUT,
TimeUnit.SECONDS));
+ // ... and the connection marked completed
+ assertTrue(pubStream.wrappedOutputStream.connectionComplete);
+ }
+
+ /**
+ * The HTTP client subscribes twice .
+ */
+ @Test
+ public void testSubscribeTwiceBeforePublisherClose() throws Exception {
+ byte[] expected = body(REQUEST_SIZE);
+
+ TestPubPipeWrap pubStream = streams(expected.length);
+
+ pubStream.pipedOutputStream.write(expected);
+
+ /*
+ * We subscribe first ... 1°
+ */
+ CountDownLatch countDownLatchBeforeSubscribe1 = new CountDownLatch(1);
+ CountDownLatch countDownLatchAfterSubscribe1 = new CountDownLatch(1);
+ // The HttpClient arrive before the close ... and subscribe...
+ // Using PipedInputStream as pin I use another thread to subscribe
+ CompletableFuture<byte[]> received = subscribeAsync(
+ pubStream.publisher,
+ countDownLatchBeforeSubscribe1,
+ countDownLatchAfterSubscribe1
+ );
+ // Wait to be subscribed 1°
+ assertTrue(countDownLatchBeforeSubscribe1.await(AWAIT_TEST_TIMEOUT,
TimeUnit.SECONDS));
+ assertTrue(countDownLatchAfterSubscribe1.await(AWAIT_TEST_TIMEOUT,
TimeUnit.SECONDS));
+
+ /*
+ * We subscribe first ... 2°
+ */
+ CountDownLatch countDownLatchBeforeSubscribe2 = new CountDownLatch(1);
+ CountDownLatch countDownLatchAfterSubscribe2 = new CountDownLatch(1);
+ // The HttpClient arrive before the close ... and subscribe...
+ // Using PipedInputStream as pin I use another thread to subscribe
+ CompletableFuture<byte[]> received2 = subscribeAsync(
+ pubStream.publisher,
+ countDownLatchBeforeSubscribe2,
+ countDownLatchAfterSubscribe2
+ );
+
+ //We assure also the 2° thread is in position in
publisher.subscribe(...)
+ assertTrue(countDownLatchBeforeSubscribe2.await(AWAIT_TEST_TIMEOUT,
TimeUnit.SECONDS));
+
+ /*
+ * ...and then close
+ */
+
+ // The caller close the stream
+ pubStream.wrappedOutputStream.close();
+
+ // Close also the publisher
+ pubStream.pipedOutputStream.close();
+
+ // Wait to be subscribed 2° ...the lock on the pin is Publisher is
release by now
+ assertTrue(countDownLatchAfterSubscribe2.await(AWAIT_TEST_TIMEOUT,
TimeUnit.SECONDS));
+
+ // Assert the body is the same
+ assertArrayEquals(expected, received.get(AWAIT_TEST_TIMEOUT,
TimeUnit.SECONDS));
+ assertArrayEquals(new byte[0], received2.get(AWAIT_TEST_TIMEOUT,
TimeUnit.SECONDS));
+ // ... and the connection marked completed
+ assertTrue(pubStream.wrappedOutputStream.connectionComplete);
+ }
+
+ /**
+ * The publisher is closed before the HTTP client subscribes.
+ * The complete request body must still be available.
+ */
+ @Test
+ public void testSubscribeAfterPublisherClose() throws Exception {
+ byte[] expected = body(REQUEST_SIZE);
+
+ TestPubPipeWrap pubStream = streams(expected.length);
+
+ pubStream.pipedOutputStream.write(expected);
+
+ /*
+ * We close before the publisher
+ */
+
+ // The caller close the stream
+ pubStream.wrappedOutputStream.close();
+ // Close also the publisher
+ pubStream.pipedOutputStream.close();
+
+
+ /*
+ * ...and then subscribe
+ */
+
+ CountDownLatch countDownLatchBeforeSubscribe = new CountDownLatch(1);
+ CountDownLatch countDownLatchAfterSubscribe = new CountDownLatch(1);
+ // The HttpClient arrive before the close ... and subscribe...
+ // Using PipedInputStream as pin I use another thread to subscribe
+ CompletableFuture<byte[]> received = subscribeAsync(
+ pubStream.publisher,
+ countDownLatchBeforeSubscribe,
+ countDownLatchAfterSubscribe
+ );
+ // Wait to be subscribed
+ assertTrue(countDownLatchBeforeSubscribe.await(AWAIT_TEST_TIMEOUT,
TimeUnit.SECONDS));
+ assertTrue(countDownLatchAfterSubscribe.await(AWAIT_TEST_TIMEOUT,
TimeUnit.SECONDS));
+
+ // Assert the body is the same
+ assertArrayEquals(expected, received.get(AWAIT_TEST_TIMEOUT,
TimeUnit.SECONDS));
+ // ... and the connection not marked completed (the stream is null in
the publisher)
+ assertFalse(pubStream.wrappedOutputStream.connectionComplete);
+ }
+
+ /**
+ * The content-length must remain available even after publisher.close()
+ */
+ @Test
+ public void testContentLengthAfterClose() throws Exception {
+ TestPubPipeWrap streams = streams(REQUEST_SIZE);
+
+ assertEquals(REQUEST_SIZE, streams.publisher.contentLength());
+
+ streams.publisher.close();
+
+ assertEquals(REQUEST_SIZE, streams.publisher.contentLength());
+ }
+
+ private static TestPubPipeWrap streams(int contentLen) throws Exception {
+ PipedInputStream pin = new PipedInputStream(CHUNK_THRESHOLD_SIZE);
+
+ HttpClientWrappedOutputStream wrappedOutputStream =
+ mock(HttpClientWrappedOutputStream.class);
+
+ wrappedOutputStream.future = new CompletableFuture<>();
+ wrappedOutputStream.contentLen = contentLen;
+
+ HTTPClientPolicy policy = new HTTPClientPolicy();
+ policy.setConnectionTimeout(180);
+
+ HttpClientBodyPublisher publisher =
+ new HttpClientBodyPublisher(wrappedOutputStream, () -> pin);
+
+ HttpClientPipedOutputStream pipedOutputStream =
+ new HttpClientPipedOutputStream(wrappedOutputStream, pin,
policy, publisher);
+
+ wrappedOutputStream.pout = pipedOutputStream;
+
+ return new TestPubPipeWrap(publisher, pipedOutputStream,
wrappedOutputStream);
+ }
+
+ private static CompletableFuture<byte[]> subscribeAsync(
+ HttpClientBodyPublisher publisher,
+ CountDownLatch countDownLatchBeforeSubscribe,
+ CountDownLatch countDownLatchAfterSubscribe
+ ) {
+
+ CompletableFuture<byte[]> received = new CompletableFuture<>();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+
+ executor.submit(() -> {
+ try {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+ Flow.Subscriber<ByteBuffer> subscriber = new
Flow.Subscriber<ByteBuffer>() {
+ @Override
+ public void onSubscribe(Flow.Subscription subscription) {
+ countDownLatchAfterSubscribe.countDown();
+ subscription.request(Long.MAX_VALUE);
+ }
+
+ @Override
+ public void onNext(ByteBuffer buffer) {
+ byte[] bytes = new byte[buffer.remaining()];
+ buffer.get(bytes);
+ baos.writeBytes(bytes);
+ }
+
+ @Override
+ public void onError(Throwable error) {
+ received.completeExceptionally(error);
+ executor.shutdown();
+ }
+
+ @Override
+ public void onComplete() {
+ received.complete(baos.toByteArray());
+ executor.shutdown();
+ }
+ };
+
+ countDownLatchBeforeSubscribe.countDown();
+ publisher.subscribe(subscriber);
+
+ } catch (Throwable e) {
+ received.completeExceptionally(e);
+ executor.shutdown();
+ }
+ });
+
+ return received;
+ }
+
+ 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;
+ }
+
+ private static class TestPubPipeWrap {
+ final HttpClientBodyPublisher publisher;
+ final HttpClientPipedOutputStream pipedOutputStream;
+ final HttpClientWrappedOutputStream wrappedOutputStream;
+
+ TestPubPipeWrap(
+ HttpClientBodyPublisher publisher,
+ HttpClientPipedOutputStream pipedOutputStream,
+ HttpClientWrappedOutputStream wrappedOutputStream
+ ) {
+ this.publisher = publisher;
+ this.pipedOutputStream = pipedOutputStream;
+ this.wrappedOutputStream = wrappedOutputStream;
+ }
+ }
+}
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 52fcd88ad02..9ce552cdb8a 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
@@ -50,6 +50,13 @@ public class BookStore {
return "CXF in Action".getBytes();
}
+ @GET
+ @Path("/pluto")
+ @Produces("text/plain")
+ public byte[] getPluto(@Context HttpServletRequest request) {
+ return (request.getProtocol() + " CXF in Action -
Pluto").getBytes(StandardCharsets.UTF_8);
+ }
+
@POST
@Path("/echo")
@Consumes("text/plain")
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 e041a50d273..13645927fbd 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
@@ -49,6 +49,38 @@ public class JettyClientServerHttp2Test extends
AbstractJettyClientServerHttp2Te
createStaticBus();
}
+ @Test(timeout = 60000)
+ public void testGetWithChunkingHttp2() 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/pluto")
+ .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(true);
+ 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.get()) {
+ assertEquals(200, resp.getStatus());
+ assertEquals("HTTP/2.0 CXF in Action - Pluto",
resp.readEntity(String.class));
+ }
+ }
/**
* The trust decision is made after the request is sent and before its
body is written. Without
