Skip to content

Commit

Permalink
CXF-8629: AsyncHTTPConduit (hc5) should support chunked request / res…
Browse files Browse the repository at this point in the history 
…ponse. Add test cases with auto-redirect
  • Loading branch information
@reta
reta committed Dec 29, 2024
1 parent 9164d67 commit dc196aa
Show file tree
Hide file tree
Showing 3 changed files with 197 additions and 10 deletions.
5 changes: 4 additions & 1 deletion ...ttp-hc5/src/main/java/org/apache/cxf/transport/http/asyncclient/hc5/AsyncHTTPConduit.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down
52 changes: 52 additions & 0 deletions systests/transport-hc5/src/test/java/org/apache/cxf/systest/hc5/jaxrs/FileStore.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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();
}
}
150 changes: 141 additions & 9 deletions ...port-hc5/src/test/java/org/apache/cxf/systest/hc5/jaxrs/JAXRSAsyncClientChunkingTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;

Expand All @@ -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
Expand All @@ -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
Expand All @@ -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
Expand All @@ -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];
Expand All @@ -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);
Expand Down

0 comments on commit dc196aa

Please sign in to comment.