This is an automated email from the ASF dual-hosted git repository.
reta 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 c4e9c3c7cc CXF-8629: AsyncHTTPConduit (hc5) should support chunked
request / response (#2021)
c4e9c3c7cc is described below
commit c4e9c3c7cc58d0e8a5e401491f632e67e2cf1af7
Author: Andriy Redko <[email protected]>
AuthorDate: Sat Oct 5 11:10:45 2024 -0400
CXF-8629: AsyncHTTPConduit (hc5) should support chunked request / response
(#2021)
---
.../http/auth/DigestAuthSupplierSpringTest.java | 2 +
.../apache/cxf/systest/hc5/jaxrs/FileStore.java | 139 ++++++++++++++++++++
.../cxf/systest/hc5/jaxrs/FileStoreServer.java | 47 +++++++
.../hc5/jaxrs/JAXRSAsyncClientChunkingTest.java | 141 +++++++++++++++++++++
.../systest/hc5/jaxws/JAXWSAsyncClientTest.java | 38 ++++++
.../src/test/resources/logging.properties | 7 +
6 files changed, 374 insertions(+)
diff --git
a/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/http/auth/DigestAuthSupplierSpringTest.java
b/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/http/auth/DigestAuthSupplierSpringTest.java
index 3b00608544..124fb11f5a 100644
---
a/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/http/auth/DigestAuthSupplierSpringTest.java
+++
b/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/http/auth/DigestAuthSupplierSpringTest.java
@@ -22,6 +22,7 @@ import jakarta.ws.rs.NotAuthorizedException;
import jakarta.ws.rs.core.MediaType;
import org.apache.cxf.jaxrs.client.WebClient;
import org.apache.cxf.transport.http.HTTPConduit;
+import org.apache.cxf.transport.http.asyncclient.hc5.AsyncHTTPConduit;
import org.apache.cxf.transport.http.auth.DigestAuthSupplier;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -67,6 +68,7 @@ public class DigestAuthSupplierSpringTest {
@Test
public void test() {
WebClient client = WebClient.create("http://localhost:"; + port,
(String) null);
+
WebClient.getConfig(client).getBus().setProperty(AsyncHTTPConduit.USE_ASYNC,
true);
assertThrows(NotAuthorizedException.class, () ->
client.get(String.class));
diff --git
a/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/jaxrs/FileStore.java
b/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/jaxrs/FileStore.java
new file mode 100644
index 0000000000..11e36a3813
--- /dev/null
+++
b/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/jaxrs/FileStore.java
@@ -0,0 +1,139 @@
+/**
+ * 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.jaxrs;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import jakarta.activation.DataHandler;
+import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.POST;
+import jakarta.ws.rs.Path;
+import jakarta.ws.rs.QueryParam;
+import jakarta.ws.rs.WebApplicationException;
+import jakarta.ws.rs.container.AsyncResponse;
+import jakarta.ws.rs.container.Suspended;
+import jakarta.ws.rs.core.Context;
+import jakarta.ws.rs.core.HttpHeaders;
+import jakarta.ws.rs.core.Response;
+import jakarta.ws.rs.core.Response.Status;
+import jakarta.ws.rs.core.StreamingOutput;
+import jakarta.ws.rs.core.UriInfo;
+import org.apache.cxf.common.util.StringUtils;
+import org.apache.cxf.helpers.IOUtils;
+import org.apache.cxf.jaxrs.ext.multipart.Attachment;
+import org.apache.cxf.jaxrs.ext.multipart.MultipartBody;
+
+@Path("/file-store")
+public class FileStore {
+ private final ConcurrentMap<String, byte[]> store = new
ConcurrentHashMap<>();
+ @Context private HttpHeaders headers;
+
+ @POST
+ @Path("/stream")
+ @Consumes("*/*")
+ public Response addBook(@QueryParam("chunked") boolean chunked,
InputStream in) throws IOException {
+ String transferEncoding = headers.getHeaderString("Transfer-Encoding");
+
+ if (chunked != Objects.equals("chunked", transferEncoding)) {
+ throw new WebApplicationException(Status.EXPECTATION_FAILED);
+ }
+
+ try (in) {
+ if (chunked) {
+ return Response.ok(new StreamingOutput() {
+ @Override
+ public void write(OutputStream out) throws IOException,
WebApplicationException {
+ in.transferTo(out);
+ }
+ }).build();
+ } else {
+ // Make sure we have small amount of data for chunking to not
kick in
+ final byte[] content = in.readAllBytes();
+ return Response.ok(Arrays.copyOf(content, content.length /
10)).build();
+ }
+ }
+ }
+
+ @POST
+ @Consumes("multipart/form-data")
+ public void addBook(@QueryParam("chunked") boolean chunked,
+ @Suspended final AsyncResponse response, @Context final UriInfo
uri, final MultipartBody body) {
+
+ String transferEncoding = headers.getHeaderString("Transfer-Encoding");
+ if (chunked != Objects.equals("chunked", transferEncoding)) {
+
response.resume(Response.status(Status.EXPECTATION_FAILED).build());
+ return;
+ }
+
+ for (final Attachment attachment: body.getAllAttachments()) {
+ final DataHandler handler = attachment.getDataHandler();
+
+ if (handler != null) {
+ final String source = handler.getName();
+ if (StringUtils.isEmpty(source)) {
+
response.resume(Response.status(Status.BAD_REQUEST).build());
+ return;
+ }
+
+ try {
+ if (store.containsKey(source)) {
+
response.resume(Response.status(Status.CONFLICT).build());
+ return;
+ }
+
+ final byte[] content =
IOUtils.readBytesFromStream(handler.getInputStream());
+ if (store.putIfAbsent(source, content) != null) {
+
response.resume(Response.status(Status.CONFLICT).build());
+ return;
+ }
+
+ if (response.isSuspended()) {
+ final StreamingOutput stream = new StreamingOutput() {
+ @Override
+ public void write(OutputStream os) throws
IOException, WebApplicationException {
+ if (chunked) {
+ // Make sure we have enough data for
chunking to kick in
+ for (int i = 0; i < 10; ++i) {
+ os.write(content);
+ }
+ } else {
+ os.write(content);
+ }
+ }
+ };
+
response.resume(Response.created(uri.getRequestUriBuilder()
+ .path(source).build()).entity(stream)
+ .build());
+ }
+
+ } catch (final Exception ex) {
+ response.resume(Response.serverError().build());
+ }
+
+ }
+ }
+ }
+}
diff --git
a/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/jaxrs/FileStoreServer.java
b/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/jaxrs/FileStoreServer.java
new file mode 100644
index 0000000000..44fec9dfa4
--- /dev/null
+++
b/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/jaxrs/FileStoreServer.java
@@ -0,0 +1,47 @@
+/**
+ * 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.jaxrs;
+
+import org.apache.cxf.jaxrs.JAXRSServerFactoryBean;
+import org.apache.cxf.jaxrs.lifecycle.SingletonResourceProvider;
+import org.apache.cxf.testutil.common.AbstractBusTestServerBase;
+
+class FileStoreServer extends AbstractBusTestServerBase {
+ private org.apache.cxf.endpoint.Server server;
+ private final String port;
+
+ FileStoreServer(String port) {
+ this.port = port;
+ }
+
+ protected void run() {
+ JAXRSServerFactoryBean sf = new JAXRSServerFactoryBean();
+ sf.setResourceClasses(FileStore.class);
+ sf.setResourceProvider(FileStore.class, new
SingletonResourceProvider(new FileStore(), true));
+ sf.setAddress("http://localhost:"; + port);
+ server = sf.create();
+ }
+
+ public void tearDown() throws Exception {
+ server.stop();
+ server.destroy();
+ server = null;
+ }
+}
diff --git
a/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/jaxrs/JAXRSAsyncClientChunkingTest.java
b/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/jaxrs/JAXRSAsyncClientChunkingTest.java
new file mode 100644
index 0000000000..0f68738534
--- /dev/null
+++
b/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/jaxrs/JAXRSAsyncClientChunkingTest.java
@@ -0,0 +1,141 @@
+/**
+ * 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.jaxrs;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Random;
+
+import jakarta.ws.rs.client.Entity;
+import jakarta.ws.rs.core.MediaType;
+import jakarta.ws.rs.core.MultivaluedMap;
+import jakarta.ws.rs.core.Response;
+import org.apache.cxf.interceptor.LoggingInInterceptor;
+import org.apache.cxf.interceptor.LoggingOutInterceptor;
+import org.apache.cxf.jaxrs.client.ClientConfiguration;
+import org.apache.cxf.jaxrs.client.WebClient;
+import org.apache.cxf.jaxrs.ext.multipart.Attachment;
+import org.apache.cxf.jaxrs.ext.multipart.MultipartBody;
+import org.apache.cxf.jaxrs.impl.MetadataMap;
+import org.apache.cxf.jaxrs.model.AbstractResourceInfo;
+import org.apache.cxf.jaxrs.provider.MultipartProvider;
+import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
+import org.apache.cxf.transport.http.asyncclient.hc5.AsyncHTTPConduit;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized.Parameters;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.not;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+
+@RunWith(value = org.junit.runners.Parameterized.class)
+public class JAXRSAsyncClientChunkingTest extends
AbstractBusClientServerTestBase {
+ private static final String PORT = allocatePort(FileStoreServer.class);
+ private final Boolean chunked;
+
+ public JAXRSAsyncClientChunkingTest(Boolean chunked) {
+ this.chunked = chunked;
+ }
+
+ @BeforeClass
+ public static void startServers() throws Exception {
+ AbstractResourceInfo.clearAllMaps();
+ assertTrue("server did not launch correctly", launchServer(new
FileStoreServer(PORT)));
+ createStaticBus();
+ }
+
+ @Parameters(name = "{0}")
+ public static Collection<Boolean> data() {
+ return Arrays.asList(new Boolean[] {Boolean.FALSE, Boolean.TRUE});
+ }
+
+ @Test
+ public void testMultipartChunking() {
+ final String url = "http://localhost:"; + PORT + "/file-store";
+ final WebClient webClient = WebClient.create(url, List.of(new
MultipartProvider())).query("chunked", chunked);
+
+ final ClientConfiguration config = WebClient.getConfig(webClient);
+ config.getBus().setProperty(AsyncHTTPConduit.USE_ASYNC, true);
+ config.getHttpConduit().getClient().setAllowChunking(chunked);
+ configureLogging(config);
+
+ try {
+ final String filename = "keymanagers.jks";
+ final MultivaluedMap<String, String> headers = new MetadataMap<>();
+ headers.add("Content-ID", filename);
+ headers.add("Content-Type", "application/binary");
+ headers.add("Content-Disposition", "attachment; filename=" +
chunked + "_" + filename);
+ final Attachment att = new
Attachment(getClass().getResourceAsStream("/" + filename), headers);
+ final MultipartBody entity = new MultipartBody(att);
+ try (Response response = webClient.header("Content-Type",
"multipart/form-data").post(entity)) {
+ assertThat(response.getStatus(), equalTo(201));
+ assertThat(response.getHeaderString("Transfer-Encoding"),
equalTo(chunked ? "chunked" : null));
+ assertThat(response.getEntity(), not(equalTo(null)));
+ }
+ } finally {
+ webClient.close();
+ }
+ }
+
+ @Test
+ public void testStreamChunking() throws IOException {
+ final String url = "http://localhost:"; + PORT + "/file-store/stream";
+ final WebClient webClient = WebClient.create(url).query("chunked",
chunked);
+
+ final ClientConfiguration config = WebClient.getConfig(webClient);
+ config.getBus().setProperty(AsyncHTTPConduit.USE_ASYNC, true);
+ config.getHttpConduit().getClient().setAllowChunking(chunked);
+ configureLogging(config);
+
+ final byte[] bytes = new byte [32 * 1024];
+ final Random random = new Random();
+ random.nextBytes(bytes);
+
+ try (InputStream in = new ByteArrayInputStream(bytes)) {
+ final Entity<InputStream> entity = Entity.entity(in,
MediaType.APPLICATION_OCTET_STREAM);
+ try (Response response = webClient.post(entity)) {
+ assertThat(response.getStatus(), equalTo(200));
+ assertThat(response.getHeaderString("Transfer-Encoding"),
equalTo(chunked ? "chunked" : null));
+ assertThat(response.getEntity(), not(equalTo(null)));
+ }
+ } finally {
+ webClient.close();
+ }
+ }
+
+ private void configureLogging(final ClientConfiguration config) {
+ final LoggingOutInterceptor out = new LoggingOutInterceptor();
+ out.setShowMultipartContent(false);
+
+ final LoggingInInterceptor in = new LoggingInInterceptor();
+ in.setShowBinaryContent(false);
+
+ config.getInInterceptors().add(in);
+ config.getOutInterceptors().add(out);
+ }
+}
diff --git
a/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/jaxws/JAXWSAsyncClientTest.java
b/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/jaxws/JAXWSAsyncClientTest.java
index 87aeab9545..32806c92ac 100644
---
a/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/jaxws/JAXWSAsyncClientTest.java
+++
b/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/jaxws/JAXWSAsyncClientTest.java
@@ -25,6 +25,7 @@ import java.io.StringWriter;
import java.net.URL;
import java.util.ArrayList;
import java.util.List;
+import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
@@ -47,9 +48,12 @@ import jakarta.xml.ws.Service;
import jakarta.xml.ws.soap.SOAPBinding;
import jakarta.xml.ws.soap.SOAPFaultException;
import org.apache.cxf.endpoint.Client;
+import org.apache.cxf.frontend.ClientProxy;
import org.apache.cxf.greeter_control.AbstractGreeterImpl;
import org.apache.cxf.greeter_control.Greeter;
import org.apache.cxf.greeter_control.types.GreetMeResponse;
+import org.apache.cxf.interceptor.LoggingInInterceptor;
+import org.apache.cxf.interceptor.LoggingOutInterceptor;
import org.apache.cxf.jaxws.JaxWsProxyFactoryBean;
import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
import org.apache.cxf.testutil.common.AbstractBusTestServerBase;
@@ -60,6 +64,7 @@ import org.junit.BeforeClass;
import org.junit.Test;
import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.CoreMatchers.equalTo;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
@@ -143,6 +148,39 @@ public class JAXWSAsyncClientTest extends
AbstractBusClientServerTestBase {
}
assertTrue("Response still not received.", response.isDone());
+ assertThat(response.get().getResponseType(), equalTo("CXF"));
+ }
+
+ @Test
+ public void testAsyncClientChunking() throws Exception {
+ // setup the feature by using JAXWS front-end API
+ JaxWsProxyFactoryBean factory = new JaxWsProxyFactoryBean();
+ factory.setAddress("http://localhost:"; + PORT +
"/SoapContext/GreeterPort");
+ factory.getOutInterceptors().add(new LoggingOutInterceptor());
+ factory.getInInterceptors().add(new LoggingInInterceptor());
+ factory.setServiceClass(Greeter.class);
+ Greeter proxy = factory.create(Greeter.class);
+
+ Client client = ClientProxy.getClient(proxy);
+ HTTPConduit http = (HTTPConduit) client.getConduit();
+ http.getClient().setAllowChunking(true);
+
+ final char[] bytes = new char [32 * 1024];
+ final Random random = new Random();
+ for (int i = 0; i < bytes.length; ++i) {
+ bytes[i] = (char)(random.nextInt(26) + 'a');
+ }
+
+ final String greeting = new String(bytes);
+ Response<GreetMeResponse> response = proxy.greetMeAsync(greeting);
+ int waitCount = 0;
+ while (!response.isDone() && waitCount < 15) {
+ Thread.sleep(1000);
+ waitCount++;
+ }
+
+ assertTrue("Response still not received.", response.isDone());
+ assertThat(response.get().getResponseType(),
equalTo(greeting.toUpperCase()));
}
@Test
diff --git a/systests/transport-hc5/src/test/resources/logging.properties
b/systests/transport-hc5/src/test/resources/logging.properties
index b2e5a799c8..c89316aba9 100644
--- a/systests/transport-hc5/src/test/resources/logging.properties
+++ b/systests/transport-hc5/src/test/resources/logging.properties
@@ -53,6 +53,13 @@
# Describes specific configuration info for Handlers.
############################################################
+# "handlers" specifies a comma separated list of log Handler
+# classes. These handlers will be installed during VM startup.
+# Note that these classes must be on the system classpath.
+# By default we only configure a ConsoleHandler, which will only
+# show messages at the INFO and above levels.
+handlers= java.util.logging.ConsoleHandler
+
# default file output is in user's home directory.
java.util.logging.FileHandler.pattern = %h/java%u.log
java.util.logging.FileHandler.limit = 50000
