CXF-8629: AsyncHTTPConduit (hc5) should support chunked request / response. Add test cases with auto-redirect

reta · reta · commit dc196aa6fc7d · 2024-12-29T15:24:48.000-05:00
diff --git a/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/AsyncHTTPConduit.java b/rt/transports/http-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/AsyncHTTPConduit.java
@@ -696,7 +696,10 @@ protected synchronized HttpResponse getHttpResponse() throws IOException {
         }
 
         protected void handleResponseAsync() throws IOException {
-            isAsync = true;
+            // The response hasn't been handled yet, should be handled asynchronously
+            if (httpResponse == null) {
+                isAsync = true;
+            }
         }
 
         protected void closeInputStream() throws IOException {
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
@@ -29,6 +29,7 @@
 
 import jakarta.activation.DataHandler;
 import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.GET;
 import jakarta.ws.rs.POST;
 import jakarta.ws.rs.Path;
 import jakarta.ws.rs.QueryParam;
@@ -38,8 +39,10 @@
 import jakarta.ws.rs.core.Context;
 import jakarta.ws.rs.core.HttpHeaders;
 import jakarta.ws.rs.core.Response;
+import jakarta.ws.rs.core.Response.ResponseBuilder;
 import jakarta.ws.rs.core.Response.Status;
 import jakarta.ws.rs.core.StreamingOutput;
+import jakarta.ws.rs.core.UriBuilder;
 import jakarta.ws.rs.core.UriInfo;
 import org.apache.cxf.common.util.StringUtils;
 import org.apache.cxf.helpers.IOUtils;
@@ -136,4 +139,53 @@ public void write(OutputStream os) throws IOException, WebApplicationException {
             }
         }
     }
+    
+    @GET
+    @Consumes("multipart/form-data")
+    public void getBook(@QueryParam("chunked") boolean chunked, @QueryParam("filename") String source,
+            @Suspended final AsyncResponse response) {
+
+        if (StringUtils.isEmpty(source)) {
+            response.resume(Response.status(Status.BAD_REQUEST).build());
+            return;
+        }
+
+        try {
+            if (!store.containsKey(source)) {
+                response.resume(Response.status(Status.NOT_FOUND).build());
+                return;
+            }
+
+            final byte[] content = store.get(source);
+            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.ok().entity(stream).build());
+            }
+
+        } catch (final Exception ex) {
+            response.resume(Response.serverError().build());
+        }
+    }
+
+    @GET
+    @Path("/redirect")
+    public Response addBook(@Context UriInfo uriInfo) {
+        final UriBuilder builder = uriInfo.getBaseUriBuilder().path(getClass());
+        uriInfo.getQueryParameters(true).forEach((p, v) -> builder.queryParam(p, v.get(0)));
+
+        final ResponseBuilder response = Response.status(303).header("Location", builder.build());
+        return response.build();
+    }
 }
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
@@ -26,12 +26,20 @@
 import java.util.Collection;
 import java.util.List;
 import java.util.Random;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.logging.Logger;
 
 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.LoggingMessage;
 import org.apache.cxf.interceptor.LoggingOutInterceptor;
 import org.apache.cxf.jaxrs.client.ClientConfiguration;
 import org.apache.cxf.jaxrs.client.WebClient;
@@ -40,6 +48,7 @@
 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.message.Message;
 import org.apache.cxf.testutil.common.AbstractBusClientServerTestBase;
 import org.apache.cxf.transport.http.asyncclient.hc5.AsyncHTTPConduit;
 
@@ -50,16 +59,20 @@
 
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.not;
+import static org.hamcrest.CoreMatchers.startsWith;
 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;
+    private final Boolean autoRedirect;
+    private final ConcurrentMap<String, AtomicInteger> ids = new ConcurrentHashMap<>();
 
-    public JAXRSAsyncClientChunkingTest(Boolean chunked) {
+    public JAXRSAsyncClientChunkingTest(Boolean chunked, Boolean autoRedirect) {
         this.chunked = chunked;
+        this.autoRedirect = autoRedirect;
     }
 
     @BeforeClass
@@ -69,9 +82,14 @@ public static void startServers() throws Exception {
         createStaticBus();
     }
     
-    @Parameters(name = "{0}")
-    public static Collection<Boolean> data() {
-        return Arrays.asList(new Boolean[] {Boolean.FALSE, Boolean.TRUE});
+    @Parameters(name = "chunked {0}, auto-redirect {1}")
+    public static Collection<Boolean[]> data() {
+        return Arrays.asList(new Boolean[][] {
+            {Boolean.FALSE /* chunked */, Boolean.FALSE /* autoredirect */},
+            {Boolean.FALSE /* chunked */, Boolean.TRUE /* autoredirect */},
+            {Boolean.TRUE /* chunked */, Boolean.FALSE /* autoredirect */},
+            {Boolean.TRUE /* chunked */, Boolean.TRUE /* autoredirect */},
+        });
     }
 
     @Test
@@ -82,24 +100,61 @@ public void testMultipartChunking() {
         final ClientConfiguration config = WebClient.getConfig(webClient);
         config.getBus().setProperty(AsyncHTTPConduit.USE_ASYNC, true);
         config.getHttpConduit().getClient().setAllowChunking(chunked);
+        config.getHttpConduit().getClient().setAutoRedirect(autoRedirect);
         configureLogging(config);
 
+        final String filename = "keymanagers.jks";
         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);
+            headers.add("Content-Disposition", "attachment; filename=" + chunked + "_" + autoRedirect + "_" + 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)) {
+            try (Response response = webClient.header("Content-Type", MediaType.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();
         }
+
+        assertRedirect(chunked + "_" + autoRedirect + "_" + filename);
+    }
+
+    @Test
+    public void testMultipartChunkingAsync() throws InterruptedException, ExecutionException, TimeoutException {
+        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);
+        config.getHttpConduit().getClient().setAutoRedirect(autoRedirect);
+        configureLogging(config);
+
+        final String filename = "keymanagers.jks";
+        try {
+            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
+                +  "_" + autoRedirect + "_async_" + filename);
+            final Attachment att = new Attachment(getClass().getResourceAsStream("/" + filename), headers);
+            final Entity<MultipartBody> entity = Entity.entity(new MultipartBody(att), 
+                    MediaType.MULTIPART_FORM_DATA_TYPE);
+            try (Response response = webClient.header("Content-Type", MediaType.MULTIPART_FORM_DATA).async()
+                    .post(entity).get(10, TimeUnit.SECONDS)) {
+                assertThat(response.getStatus(), equalTo(201));
+                assertThat(response.getHeaderString("Transfer-Encoding"), equalTo(chunked ? "chunked" : null));
+                assertThat(response.getEntity(), not(equalTo(null)));
+            }
+        } finally {
+            webClient.close();
+        }
+
+        assertRedirect(chunked + "_" + autoRedirect + "_" + filename);
     }
 
     @Test
@@ -110,6 +165,7 @@ public void testStreamChunking() throws IOException {
         final ClientConfiguration config = WebClient.getConfig(webClient);
         config.getBus().setProperty(AsyncHTTPConduit.USE_ASYNC, true);
         config.getHttpConduit().getClient().setAllowChunking(chunked);
+        config.getHttpConduit().getClient().setAutoRedirect(autoRedirect);
         configureLogging(config);
 
         final byte[] bytes = new byte [32 * 1024];
@@ -126,13 +182,89 @@ public void testStreamChunking() throws IOException {
         } finally {
             webClient.close();
         }
+
+        assertNoDuplicateLogging();
     }
-    
+
+    @Test
+    public void testStreamChunkingAsync() throws IOException, InterruptedException,
+            ExecutionException, TimeoutException {
+        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);
+        config.getHttpConduit().getClient().setAutoRedirect(autoRedirect);
+        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.async().post(entity).get(10, TimeUnit.SECONDS)) {
+                assertThat(response.getStatus(), equalTo(200));
+                assertThat(response.getHeaderString("Transfer-Encoding"), equalTo(chunked ? "chunked" : null));
+                assertThat(response.getEntity(), not(equalTo(null)));
+            }
+        } finally {
+            webClient.close();
+        }
+
+        assertNoDuplicateLogging();
+    }
+
+    private void assertRedirect(String filename) {
+        final String url = "http://localhost:" + PORT + "/file-store/redirect";
+
+        final WebClient webClient = WebClient.create(url, List.of(new MultipartProvider()))
+                .query("chunked", chunked)
+                .query("filename", filename);
+
+        final ClientConfiguration config = WebClient.getConfig(webClient);
+        config.getBus().setProperty(AsyncHTTPConduit.USE_ASYNC, true);
+        config.getHttpConduit().getClient().setAllowChunking(chunked);
+        config.getHttpConduit().getClient().setAutoRedirect(autoRedirect);
+        configureLogging(config);
+
+        try {
+            try (Response response = webClient.get()) {
+                if (autoRedirect) {
+                    assertThat(response.getStatus(), equalTo(200));
+                    assertThat(response.getHeaderString("Transfer-Encoding"), equalTo(chunked ? "chunked" : null));
+                    assertThat(response.getEntity(), not(equalTo(null)));
+                } else {
+                    assertThat(response.getStatus(), equalTo(303));
+                    assertThat(response.getHeaderString("Location"),
+                        startsWith("http://localhost:" + PORT + "/file-store"));
+                }
+            }
+        } finally {
+            webClient.close();
+        }
+
+        assertNoDuplicateLogging();
+    }
+
+    private void assertNoDuplicateLogging() {
+        ids.forEach((id, counter) -> assertThat("Duplicate client logging for message " + id,
+            counter.get(), equalTo(1)));
+    }
+
     private void configureLogging(final ClientConfiguration config) {
         final LoggingOutInterceptor out = new LoggingOutInterceptor();
         out.setShowMultipartContent(false);
 
-        final LoggingInInterceptor in = new LoggingInInterceptor();
+        final LoggingInInterceptor in = new LoggingInInterceptor() {
+            @Override
+            protected void logging(Logger logger, Message message) {
+                super.logging(logger, message);
+                final String id = (String) message.get(LoggingMessage.ID_KEY);
+                ids.computeIfAbsent(id, key -> new AtomicInteger()).incrementAndGet(); 
+            }
+        };
         in.setShowBinaryContent(false);
 
         config.getInInterceptors().add(in);

Original file line number	Diff line number	Diff line change
`@@ -696,7 +696,10 @@ protected synchronized HttpResponse getHttpResponse() throws IOException {`
`696`	`696`	`}`
`697`	`697`
`698`	`698`	`protected void handleResponseAsync() throws IOException {`
`699`		`- isAsync = true;`
	`699`	`+ // The response hasn't been handled yet, should be handled asynchronously`
	`700`	`+ if (httpResponse == null) {`
	`701`	`+ isAsync = true;`
	`702`	`+ }`
`700`	`703`	`}`
`701`	`704`
`702`	`705`	`protected void closeInputStream() throws IOException {`