From 1573f3206faedd95984a0e13debd626efb6bcdd3 Mon Sep 17 00:00:00 2001 From: daijithegeek Date: Thu, 4 Aug 2022 15:59:31 +0200 Subject: [PATCH 1/3] Enable data streaming possible by moving the response future and callback completion right after headers are received. All necessary information is present there to ensure proper Jersey handling of the content. This implementation generates an unwanted deadlock though. --- .../jetty/connector/JettyConnector.java | 45 ++-- .../jersey/jetty/connector/StreamingTest.java | 249 ++++++++++++++++++ 2 files changed, 269 insertions(+), 25 deletions(-) create mode 100644 connectors/jetty-connector/src/test/java/org/glassfish/jersey/jetty/connector/StreamingTest.java diff --git a/connectors/jetty-connector/src/main/java/org/glassfish/jersey/jetty/connector/JettyConnector.java b/connectors/jetty-connector/src/main/java/org/glassfish/jersey/jetty/connector/JettyConnector.java index 89a0c05eb7..0a11801ed5 100644 --- a/connectors/jetty-connector/src/main/java/org/glassfish/jersey/jetty/connector/JettyConnector.java +++ b/connectors/jetty-connector/src/main/java/org/glassfish/jersey/jetty/connector/JettyConnector.java @@ -34,33 +34,15 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicReference; import java.util.logging.Level; import java.util.logging.Logger; +import javax.net.ssl.SSLContext; import javax.ws.rs.ProcessingException; import javax.ws.rs.client.Client; import javax.ws.rs.core.Configuration; import javax.ws.rs.core.MultivaluedMap; -import javax.net.ssl.SSLContext; - -import org.eclipse.jetty.client.util.BasicAuthentication; -import org.eclipse.jetty.client.util.BytesContentProvider; -import org.eclipse.jetty.client.util.FutureResponseListener; -import org.eclipse.jetty.client.util.OutputStreamContentProvider; -import org.glassfish.jersey.client.ClientProperties; -import org.glassfish.jersey.client.ClientRequest; -import org.glassfish.jersey.client.ClientResponse; -import org.glassfish.jersey.client.innate.ClientProxy; -import org.glassfish.jersey.client.spi.AsyncConnectorCallback; -import org.glassfish.jersey.client.spi.Connector; -import org.glassfish.jersey.internal.util.collection.ByteBufferInputStream; -import org.glassfish.jersey.internal.util.collection.NonBlockingInputStream; -import org.glassfish.jersey.message.internal.HeaderUtils; -import org.glassfish.jersey.message.internal.OutboundMessageContext; -import org.glassfish.jersey.message.internal.Statuses; - import org.eclipse.jetty.client.HttpClient; import org.eclipse.jetty.client.HttpProxy; import org.eclipse.jetty.client.ProxyConfiguration; @@ -70,6 +52,10 @@ import org.eclipse.jetty.client.api.Request; import org.eclipse.jetty.client.api.Response; import org.eclipse.jetty.client.api.Result; +import org.eclipse.jetty.client.util.BasicAuthentication; +import org.eclipse.jetty.client.util.BytesContentProvider; +import org.eclipse.jetty.client.util.FutureResponseListener; +import org.eclipse.jetty.client.util.OutputStreamContentProvider; import org.eclipse.jetty.http.HttpField; import org.eclipse.jetty.http.HttpFields; import org.eclipse.jetty.http.HttpHeader; @@ -77,6 +63,17 @@ import org.eclipse.jetty.util.Jetty; import org.eclipse.jetty.util.ssl.SslContextFactory; import org.eclipse.jetty.util.thread.QueuedThreadPool; +import org.glassfish.jersey.client.ClientProperties; +import org.glassfish.jersey.client.ClientRequest; +import org.glassfish.jersey.client.ClientResponse; +import org.glassfish.jersey.client.innate.ClientProxy; +import org.glassfish.jersey.client.spi.AsyncConnectorCallback; +import org.glassfish.jersey.client.spi.Connector; +import org.glassfish.jersey.internal.util.collection.ByteBufferInputStream; +import org.glassfish.jersey.internal.util.collection.NonBlockingInputStream; +import org.glassfish.jersey.message.internal.HeaderUtils; +import org.glassfish.jersey.message.internal.OutboundMessageContext; +import org.glassfish.jersey.message.internal.Statuses; /** * A {@link Connector} that utilizes the Jetty HTTP Client to send and receive @@ -402,7 +399,6 @@ public Future apply(final ClientRequest jerseyRequest, final AsyncConnectorCa } }); - final AtomicReference jerseyResponse = new AtomicReference<>(); final ByteBufferInputStream entityStream = new ByteBufferInputStream(); jettyRequest.send(new Response.Listener.Adapter() { @@ -417,7 +413,10 @@ public void onHeaders(final Response jettyResponse) { } } final ClientResponse response = translateResponse(jerseyRequest, jettyResponse, entityStream); - jerseyResponse.set(response); + if (callbackInvoked.compareAndSet(false, true)) { + callback.response(response); + } + responseFuture.complete(response); } @Override @@ -447,10 +446,6 @@ public void onContent(final Response jettyResponse, final ByteBuffer content) { @Override public void onComplete(final Result result) { entityStream.closeQueue(); - if (!callbackInvoked.get()) { - callback.response(jerseyResponse.get()); - } - responseFuture.complete(jerseyResponse.get()); } @Override diff --git a/connectors/jetty-connector/src/test/java/org/glassfish/jersey/jetty/connector/StreamingTest.java b/connectors/jetty-connector/src/test/java/org/glassfish/jersey/jetty/connector/StreamingTest.java new file mode 100644 index 0000000000..d93fa20073 --- /dev/null +++ b/connectors/jetty-connector/src/test/java/org/glassfish/jersey/jetty/connector/StreamingTest.java @@ -0,0 +1,249 @@ +/* + * Copyright (c) 2013, 2022 Oracle and/or its affiliates. All rights reserved. + * + * This program and the accompanying materials are made available under the + * terms of the Eclipse Public License v. 2.0, which is available at + * http://www.eclipse.org/legal/epl-2.0. + * + * This Source Code may also be made available under the following Secondary + * Licenses when the conditions for such availability set forth in the + * Eclipse Public License v. 2.0 are satisfied: GNU General Public License, + * version 2 with the GNU Classpath Exception, which is available at + * https://www.gnu.org/software/classpath/license.html. + * + * SPDX-License-Identifier: EPL-2.0 OR GPL-2.0 WITH Classpath-exception-2.0 + */ + +package org.glassfish.jersey.jetty.connector; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.endsWith; +import static org.hamcrest.Matchers.equalTo; +import static org.hamcrest.Matchers.greaterThan; +import static org.hamcrest.Matchers.lessThan; + +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; +import java.util.logging.Level; +import java.util.logging.Logger; + +import javax.ws.rs.DefaultValue; +import javax.ws.rs.GET; +import javax.ws.rs.Path; +import javax.ws.rs.ProcessingException; +import javax.ws.rs.Produces; +import javax.ws.rs.QueryParam; +import javax.ws.rs.core.Application; +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.Response; +import javax.ws.rs.core.StreamingOutput; + +import org.glassfish.jersey.CommonProperties; +import org.glassfish.jersey.client.ClientConfig; +import org.glassfish.jersey.client.ClientProperties; +import org.glassfish.jersey.logging.LoggingFeature; +import org.glassfish.jersey.server.ResourceConfig; +import org.glassfish.jersey.test.JerseyTest; +import org.junit.Test; + +public class StreamingTest extends JerseyTest { + private static final Logger LOGGER = Logger.getLogger(StreamingTest.class.getName()); + + @Path("/test") + public static class StreamingResource { + + /** + * Long-running streaming request + * + * @param count number of packets send + * @param pauseMillis pause between each packets + */ + @GET + @Produces(MediaType.TEXT_PLAIN) + @Path("stream") + public Response streamsWithDelay(@QueryParam("start") @DefaultValue("0") int startMillis, @QueryParam("count") int count, + @QueryParam("pauseMillis") int pauseMillis) { + StreamingOutput streamingOutput = streamSlowly(startMillis, count, pauseMillis); + + return Response.ok(streamingOutput) + .build(); + } + + @GET + @Produces(MediaType.APPLICATION_JSON) + @Path("json") + public String json() { + return "great success"; + } + } + + private static StreamingOutput streamSlowly(int startMillis, int count, int pauseMillis) { + + return output -> { + try { + TimeUnit.MILLISECONDS.sleep(startMillis); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + output.write("begin\n".getBytes(StandardCharsets.UTF_8)); + output.flush(); + for (int i = 0; i < count; i++) { + try { + TimeUnit.MILLISECONDS.sleep(pauseMillis); + } + catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + + output.write(("message " + i + "\n").getBytes(StandardCharsets.UTF_8)); + output.flush(); + } + output.write("end".getBytes(StandardCharsets.UTF_8)); + }; + } + + @Override + protected Application configure() { + ResourceConfig config = new ResourceConfig(StreamingResource.class); + config.register(new LoggingFeature(LOGGER, LoggingFeature.Verbosity.PAYLOAD_ANY)); + return config; + } + + @Override + protected void configureClient(ClientConfig config) { + config.connectorProvider(new JettyConnectorProvider()); + } + + /** + * Test accessing an operation that is streaming slowly + * + * @throws ProcessingException in case of a test error. + */ + @Test + public void testDataStreamedASAP() throws Exception { + + int count = 5; + int pauseMillis = 1000; + + long start = System.currentTimeMillis(); + final Future future = target("test") + .property(CommonProperties.OUTBOUND_CONTENT_LENGTH_BUFFER_SERVER, "-1") + .path("stream") + .queryParam("count", count) + .queryParam("pauseMillis", pauseMillis) + .request() + .async() + .get(); + + Response response = future.get(); + StreamingStatistics stats = computeOutputStatistics(() -> response.readEntity(InputStream.class), start); + + assertThat("Listening the stream for bytes starts after headers are received", + stats.timeToStartReadingBytes, lessThan(500L)); + assertThat("The first bytes are forwarded ASAP", stats.timeToFirstByte, lessThan(500L)); + assertThat("Last bytes come way after the start due to the streaming pauses", stats.timeToLastByte, greaterThan(5000L)); + assertThat("Data should be complete", stats.data, endsWith("end")); + } + + /** + * Test accessing an operation that is streaming slowly + * + * @throws ProcessingException in case of a test error. + */ + @Test + public void testJettyThreadShouldNotDeadlock() throws Exception { + + /** + * This test fails due to a deadlock when reading the entity in org.glassfish.jersey.client.JerseyInvocation#translate + * The entity reading seems to be triggered by calling the response callback, but since this is done in the Jetty thread + * that is also pushing contents to the buffer, the result is a deadlock. + * + * The observed behavior is a timeout. + */ + AnObject result = target("test") + .property(ClientProperties.READ_TIMEOUT, "10000") // remove this timeout to deadlock indefinitely + .property(CommonProperties.OUTBOUND_CONTENT_LENGTH_BUFFER_SERVER, "-1") + .path("json") + .request() + .async() + .get(AnObject.class) + .get(); + + assertThat("Listening the stream for bytes starts after headers are received", result, equalTo("blabla")); + } + + private StreamingStatistics computeOutputStatistics(Supplier stream, long httpCallStart) throws IOException { + AtomicLong timeToStartReading = new AtomicLong(System.currentTimeMillis() - httpCallStart); + AtomicLong timeToFirstBytes = new AtomicLong(0); + AtomicLong timeToLastBytes = new AtomicLong(0); + AtomicReference data = new AtomicReference<>("none"); + + try (InputStream in = stream.get()) { + + byte[] buffer = new byte[4]; + StringBuffer stringBuffer = new StringBuffer(); + int consumed = -1; + + while ((consumed = in.read(buffer)) != -1) { + timeToFirstBytes.compareAndSet(0L, System.currentTimeMillis() - httpCallStart); // initialize on first iteration + String message = new String(buffer, 0, consumed, StandardCharsets.UTF_8); + stringBuffer.append(message); + LOGGER.log(Level.INFO, "got {0} after {1}ms", new Object[] {message, System.currentTimeMillis() - httpCallStart}); + } + + timeToLastBytes.compareAndSet(0L, System.currentTimeMillis() - httpCallStart); + data.set(stringBuffer.toString()); + } + + return new StreamingStatistics(timeToStartReading.get(), timeToFirstBytes.get(), timeToLastBytes.get(), data.get()); + } + + public static class StreamingStatistics { + + private final Long timeToStartReadingBytes; + private final Long timeToFirstByte; + + private final Long timeToLastByte; + private final String data; + + public StreamingStatistics(Long timeToReadingBytes, Long timeToFirstByte, Long timeToLastByte, String data) { + this.timeToStartReadingBytes = timeToReadingBytes; + this.timeToFirstByte = timeToFirstByte; + this.timeToLastByte = timeToLastByte; + this.data = data; + } + } + + public static class AnObject { + private String aField; + private int anotherField; + + public AnObject(String aField, int anotherField) { + this.aField = aField; + this.anotherField = anotherField; + } + + public String getaField() { + return aField; + } + + public void setaField(String aField) { + this.aField = aField; + } + + public int getAnotherField() { + return anotherField; + } + + public void setAnotherField(int anotherField) { + this.anotherField = anotherField; + } + } +} From 28b53de7947df4f010e7ebb80de69d590d8171b3 Mon Sep 17 00:00:00 2001 From: daijithegeek Date: Thu, 11 Aug 2022 10:06:32 +0200 Subject: [PATCH 2/3] Fix resource implementation. It should return AnObject instance. --- .../org/glassfish/jersey/jetty/connector/StreamingTest.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/connectors/jetty-connector/src/test/java/org/glassfish/jersey/jetty/connector/StreamingTest.java b/connectors/jetty-connector/src/test/java/org/glassfish/jersey/jetty/connector/StreamingTest.java index d93fa20073..8494c2570e 100644 --- a/connectors/jetty-connector/src/test/java/org/glassfish/jersey/jetty/connector/StreamingTest.java +++ b/connectors/jetty-connector/src/test/java/org/glassfish/jersey/jetty/connector/StreamingTest.java @@ -78,8 +78,8 @@ public Response streamsWithDelay(@QueryParam("start") @DefaultValue("0") int sta @GET @Produces(MediaType.APPLICATION_JSON) @Path("json") - public String json() { - return "great success"; + public AnObject json() { + return new AnObject("a field", 42); } } From 2711ad05221fdec523fc4760ac0947738e176bc2 Mon Sep 17 00:00:00 2001 From: daijithegeek Date: Thu, 18 Aug 2022 11:28:55 +0200 Subject: [PATCH 3/3] Double-checked that the StreamingTest behaves as expected with current JettyConnector implementation. --- .../glassfish/jersey/jetty/connector/StreamingTest.java | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/connectors/jetty-connector/src/test/java/org/glassfish/jersey/jetty/connector/StreamingTest.java b/connectors/jetty-connector/src/test/java/org/glassfish/jersey/jetty/connector/StreamingTest.java index 8494c2570e..6be5819c3a 100644 --- a/connectors/jetty-connector/src/test/java/org/glassfish/jersey/jetty/connector/StreamingTest.java +++ b/connectors/jetty-connector/src/test/java/org/glassfish/jersey/jetty/connector/StreamingTest.java @@ -54,6 +54,7 @@ public class StreamingTest extends JerseyTest { private static final Logger LOGGER = Logger.getLogger(StreamingTest.class.getName()); + public static final String FIELD_CONTENT = "a field"; @Path("/test") public static class StreamingResource { @@ -79,7 +80,7 @@ public Response streamsWithDelay(@QueryParam("start") @DefaultValue("0") int sta @Produces(MediaType.APPLICATION_JSON) @Path("json") public AnObject json() { - return new AnObject("a field", 42); + return new AnObject(FIELD_CONTENT, 42); } } @@ -176,7 +177,7 @@ public void testJettyThreadShouldNotDeadlock() throws Exception { .get(AnObject.class) .get(); - assertThat("Listening the stream for bytes starts after headers are received", result, equalTo("blabla")); + assertThat( result.getaField(), equalTo(FIELD_CONTENT)); } private StreamingStatistics computeOutputStatistics(Supplier stream, long httpCallStart) throws IOException { @@ -225,6 +226,10 @@ public static class AnObject { private String aField; private int anotherField; + public AnObject() { + //empty constructor for jackson + } + public AnObject(String aField, int anotherField) { this.aField = aField; this.anotherField = anotherField;