This is an automated email from the ASF dual-hosted git repository. reta pushed a commit to branch 3.5.x-fixes in repository https://gitbox.apache.org/repos/asf/cxf.git
commit 9c4fd11f88ec3bf17584842523d2aa51cfced20b Author: Andriy Redko <[email protected]> AuthorDate: Mon Oct 7 12:18:20 2024 -0400 CXF-8629: AsyncHTTPConduit (hc5) should support chunked request / response --- .../http/auth/DigestAuthSupplierSpringTest.java | 2 + .../apache/cxf/systest/hc5/jaxrs/FileStore.java | 145 +++++++++++++++++++++ .../cxf/systest/hc5/jaxrs/FileStoreServer.java | 47 +++++++ .../hc5/jaxrs/JAXRSAsyncClientChunkingTest.java | 142 ++++++++++++++++++++ .../systest/hc5/jaxws/JAXWSAsyncClientTest.java | 39 ++++++ .../src/test/resources/logging.properties | 7 + 6 files changed, 382 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 f0abaf5596..e634dae931 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 @@ -23,6 +23,7 @@ import javax.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; @@ -68,6 +69,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..753e70130a --- /dev/null +++ b/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/jaxrs/FileStore.java @@ -0,0 +1,145 @@ +/** + * 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 javax.activation.DataHandler; +import javax.ws.rs.Consumes; +import javax.ws.rs.POST; +import javax.ws.rs.Path; +import javax.ws.rs.QueryParam; +import javax.ws.rs.WebApplicationException; +import javax.ws.rs.container.AsyncResponse; +import javax.ws.rs.container.Suspended; +import javax.ws.rs.core.Context; +import javax.ws.rs.core.HttpHeaders; +import javax.ws.rs.core.Response; +import javax.ws.rs.core.Response.Status; +import javax.ws.rs.core.StreamingOutput; +import javax.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("*/*") + @SuppressWarnings("PMD.UseTryWithResources") + 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 { + 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(); + } + } finally { + if (in != null) { + in.close(); + } + } + } + + @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..230e9760ae --- /dev/null +++ b/systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/jaxrs/JAXRSAsyncClientChunkingTest.java @@ -0,0 +1,142 @@ +/** + * 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 javax.ws.rs.client.Entity; +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.MultivaluedMap; +import javax.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 b458714d2c..31d2579263 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 @@ -19,6 +19,7 @@ package org.apache.cxf.systest.hc5.jaxws; +import java.util.Random; import java.util.concurrent.ExecutionException; import javax.jws.WebService; @@ -26,9 +27,12 @@ import javax.xml.ws.Response; import javax.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; @@ -38,6 +42,8 @@ import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.Test; +import static org.hamcrest.CoreMatchers.equalTo; +import static org.hamcrest.MatcherAssert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -112,6 +118,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
