CXF-8629: AsyncHTTPConduit (hc5) should support chunked request / res…

…ponse

systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/http/auth/DigestAuthSupplierSpringTest.java

@@ -22,6 +22,7 @@
 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));
 

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());
+                }
+
+            }
+        }
+    }
+}

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;
+    }
+}

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);
+    }
+}

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