Add support for Jetty Reactive Streams HTTP client #1783

sdeleuze · 2018-04-10T12:36:35Z

Ready to be reviewed before merging in master since I would like to begin having feedback with real applications using Spring Framework snapshots.

Notice that this PR is implemented with buffer copy instead of optimized buffer wrapping because the latter hangs since Callback#succeeded doesn't allow releasing the buffer and requesting more data at different times (required for Mono<DataBuffer> for example).

Simone is aware of that issue and will see if it can be changed on Jetty side (not sure yet). That could be changed with an additional commit if that becomes possible in the future.

Issue: SPR-15092

sbordet · 2018-04-10T16:33:59Z

spring-web/spring-web.gradle

@@ -36,6 +36,7 @@ dependencies {
 	optional("io.netty:netty-all")
 	optional("io.projectreactor.ipc:reactor-netty")
 	optional("org.apache.tomcat.embed:tomcat-embed-core:${tomcatVersion}")
+	optional("org.eclipse.jetty:jetty-reactive-httpclient:0.9.2")


Version 1.0.0 with RS TCK implemented is available in Maven Central.

Updated thanks

poutsma

Looks good overall. Some things that need to change and some things I need more info on during our meeting.

poutsma · 2018-04-12T09:47:28Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpConnector.java

+ * @author Sebastien Deleuze
+ * @since 5.1
+ */
+public class JettyClientHttpConnector implements ClientHttpConnector {


I think this class should implement Spring's SmartLifecycle instead of lazily starting the HttpClient in connect. This way, we make sure the started is started on app context refresh. We should also call httpClient.stop() as part of implementing Lifecycle, because it will clean up the resources that are currently leaking.

poutsma · 2018-04-12T09:50:35Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpConnector.java

+	public Mono<ClientHttpResponse> connect(HttpMethod method, URI uri,
+			Function<? super ClientHttpRequest, Mono<Void>> requestCallback) {
+
+		if (!httpClient.isStarted()) {


Though I suggest to implement SmartLifecycle above, I think we should keep this check. Not everybody will configure their WebClient in an app context, so the client might not yet be started when connect is called.

poutsma · 2018-04-12T09:56:02Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpRequest.java

+
+	private final Request jettyRequest;
+
+	private final DefaultDataBufferFactory bufferFactory = new DefaultDataBufferFactory();


I think the data buffer factory should be exposed via a setter in the JettyClientHttpConnector, and passed along in the constructor. That way, people can configure data buffer usage (i.e. direct vs heap-based, etc.)

poutsma · 2018-04-12T10:16:54Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpRequest.java

+		String contentType = jettyRequest.getHeaders().contains(HttpHeader.CONTENT_TYPE) ?
+				jettyRequest.getHeaders().getField(HttpHeader.CONTENT_TYPE).getValue():
+				MediaType.APPLICATION_OCTET_STREAM_VALUE;
+		Publisher<ContentChunk> chunks = Flux.from(publisher).map(DataBuffer::asByteBuffer).map(ContentChunk::new);


Even though this class is using DefaultDataBuffer, which does not require releasing, I still think it would be a good idea to release them here anyway, especially if we make the data buffer factory configurable.

I think the release can be implemented as the ContentChunk callback (I am assuming that the callback would be called after the data has been written, not sure about that). I see that below you said that releasing in the callback does not work for the response, but perhaps it does for the request?

poutsma · 2018-04-12T10:17:28Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpRequest.java

+		ReactiveRequest.Content content = ReactiveRequest.Content.fromPublisher(Flux
+				.from(body)
+				.flatMap(publisher -> publisher)
+				.map(buffer -> new ContentChunk(buffer.asByteBuffer())), contentType);


Release needs to happen here as well.

poutsma · 2018-04-12T10:25:29Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpRequest.java

+	@Override
+	public HttpMethod getMethod() {
+		HttpMethod method = HttpMethod.resolve(this.jettyRequest.getMethod());
+		Assert.notNull(method, "Method must not be null");


Seems like Assert.state is more suitable here, because notNull results in an IllegalArgumentException.

poutsma · 2018-04-12T10:26:39Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpRequest.java

+ * @author Sebastien Deleuze
+ * @since 5.1
+ */
+public class JettyClientHttpRequest extends AbstractClientHttpRequest {


Could be package-protected instead, no need to expose this class.

poutsma · 2018-04-12T10:26:50Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpResponse.java

+ * @author Sebastien Deleuze
+ * @since 5.1
+ */
+public class JettyClientHttpResponse implements ClientHttpResponse {


Could be package-protected instead, no need to expose it.

poutsma · 2018-04-12T10:28:07Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpResponse.java

+	private final Flux<DataBuffer> content;
+
+
+	public JettyClientHttpResponse(ReactiveResponse reactiveResponse, Publisher<DataBuffer> content) {


The Publisher<DataBuffer> parameter should be Flux<DataBuffer>, since the only place you're creating them is internally, from JettyClientHttpRequest.

poutsma · 2018-04-12T10:28:24Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpResponse.java

+		Assert.notNull(reactiveResponse, "reactiveResponse should not be null");
+		Assert.notNull(content, "content should not be null");
+		this.reactiveResponse = reactiveResponse;
+		this.content = Flux.from(content);


See above, drop the Flux.from.

sdeleuze · 2018-04-12T12:22:45Z

I have taken in account your feedback @poutsma thanks.

rstoyanchev · 2018-07-05T19:31:13Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpConnector.java

+	public Mono<ClientHttpResponse> connect(HttpMethod method, URI uri,
+			Function<? super ClientHttpRequest, Mono<Void>> requestCallback) {
+
+		if (!httpClient.isStarted()) {


Really minor, but perhaps delegate to isRunning()?

I don't think so, isStarted() is different than isRunning().

rstoyanchev · 2018-07-05T19:46:12Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpConnector.java

+	public int getPhase() {
+		return Integer.MAX_VALUE;
+	}
+}


For ordering of SmartLifecycle methods, in the very least:

stop(Runnable) should be after stop()

getPhase and isAutoStartup up ahead among accessors

As lifecycle methods, personally I like to see start/stop/isRunning ahead of runtime methods like connect, and it's something I've been following in a number of classes, but I'm not sure if there is a broader precedent.

rstoyanchev · 2018-07-05T19:48:28Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpConnector.java

+	@Override
+	public void start() {
+		try {
+			this.httpClient.start();


A code level comment that explains HttpClient is internally synchronized and protected with state checks around here could be useful.

rstoyanchev · 2018-07-05T19:50:04Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpConnector.java

+
+		if (!uri.isAbsolute()) {
+			return Mono.error(new IllegalArgumentException("URI is not absolute: " + uri));
+		}


No practical difference but maybe this check first (before starting the client above) ?

rstoyanchev · 2018-07-05T19:57:10Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpRequest.java

+		headers.entrySet().forEach(e -> e.getValue().forEach(v -> this.jettyRequest.header(e.getKey(), v)));
+		if (!headers.containsKey(HttpHeaders.ACCEPT)) {
+			this.jettyRequest.header(HttpHeaders.ACCEPT, "*/*");
+		}


Is it necessary to add "*/*" explicitly? It should be the same as not having it.

rstoyanchev · 2018-07-05T19:59:59Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpRequest.java

+	public Mono<ClientHttpResponse> getResponse() {
+		return this.response;
+	}
+


The overall method ordering should be: accessors, public contract methods, protected contract (base class) methods (e.g. see ReactorHttpRequest).

rstoyanchev · 2018-07-05T20:16:20Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpRequest.java

+	public Mono<Void> writeWith(Publisher<? extends DataBuffer> publisher) {
+		String contentType = jettyRequest.getHeaders().contains(HttpHeader.CONTENT_TYPE) ?
+				jettyRequest.getHeaders().getField(HttpHeader.CONTENT_TYPE).getValue():
+				MediaType.APPLICATION_OCTET_STREAM_VALUE;


Is it not enough to look up via getHeaders().getContentType() ?

rstoyanchev · 2018-07-05T20:18:25Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpRequest.java

+				DataBufferUtils.release(buffer);
+				throw Exceptions.propagate(x);
+			}
+		}));


This could be extracted:

private ContentChunk toContentChunk(DataBuffer buffer) { return new ContentChunk(buffer.asByteBuffer(), new Callback() { @Override public void succeeded() { DataBufferUtils.release(buffer); } @Override public void failed(Throwable x) { DataBufferUtils.release(buffer); throw Exceptions.propagate(x); } }); }

Then it becomes:

Publisher<ContentChunk> chunks = Flux.from(publisher).map(this::toContentChunk);

And the writeAndFlushWith method below does the same AFAICS.

rstoyanchev · 2018-07-05T21:33:31Z

spring-web/src/main/java/org/springframework/http/client/reactive/JettyClientHttpRequest.java

+
+	public Mono<ClientHttpResponse> getResponse() {
+		return this.response;
+	}


It's not obvious at first why a request object should have a method that returns the response, and a public method for that. It must be called at the right time too, or else the response field may not have been initialized, but I can see why the method is here, based on the way the Jetty API works.

I think it would help clarity if the method called completes simply stored ReactiveRequest, returned Mono.empty(), and nothing more. The method already returns Mono.empty() anyway, so the part about creating the response is completely independent and could be moved out of the request. In other words, the connector would access the ReactiveRequest (via package private method) and then create the response. Then would make it a little more instantly obvious what's going on.

The other point is that the Mono<Void> from writeWith is supposed to reflect the completion of the writing of the request body, per contract, but probably not a practical issue at this time. If we keep all this package private we have the option to evolve it in the future.

rstoyanchev · 2018-07-05T21:42:32Z

Sorry for the delayed response, but my comments above. Let's get this ready for master, and continue there as needed.

Implemented with buffer copy instead of optimized buffer wrapping because the latter hangs since Callback#succeeded doesn't allow releasing the buffer and requesting more data at different times (required for Mono<DataBuffer> for example). See jetty/jetty.project#2429. Issue: SPR-15092

sdeleuze · 2018-07-07T16:46:04Z

@rstoyanchev Thanks for your feedback, I think I have taken your remarks in account. I have not yet merged it in master because I have a strange remaining issue: SseIntegrationTests and WebClientIntegrationTests fail on command line with ./gradlew clean test (AssertionError, likely timeouts) but succeed in IntelliJ IDEA. Any idea?

sdeleuze · 2018-07-18T12:56:57Z

Merged via a87764f.

sdeleuze self-assigned this Apr 10, 2018

sdeleuze requested review from bclozel, poutsma and rstoyanchev April 10, 2018 12:36

sbordet mentioned this pull request Apr 10, 2018

Review HttpClient backpressure semantic jetty/jetty.project#2429

Closed

sbordet reviewed Apr 10, 2018

View reviewed changes

sdeleuze force-pushed the SPR-15092 branch 4 times, most recently from 08ae9e3 to f12fbf1 Compare April 11, 2018 08:42

poutsma reviewed Apr 12, 2018

View reviewed changes

sdeleuze force-pushed the SPR-15092 branch 2 times, most recently from 8cae9a1 to 82481a9 Compare April 12, 2018 12:19

rstoyanchev reviewed Jul 5, 2018

View reviewed changes

sdeleuze force-pushed the SPR-15092 branch 2 times, most recently from 151e245 to de8c1e2 Compare July 7, 2018 16:28

sdeleuze force-pushed the SPR-15092 branch from de8c1e2 to 85a35c2 Compare July 7, 2018 16:41

sdeleuze closed this Jul 18, 2018

spring-projects-issues mentioned this pull request Jan 11, 2019

Jetty HTTP client integration with WebClient [SPR-15092] #19658

Closed

sdeleuze deleted the SPR-15092 branch April 28, 2021 09:57


		private final Request jettyRequest;

		private final DefaultDataBufferFactory bufferFactory = new DefaultDataBufferFactory();

		private final Flux<DataBuffer> content;


		public JettyClientHttpResponse(ReactiveResponse reactiveResponse, Publisher<DataBuffer> content) {

Add support for Jetty Reactive Streams HTTP client #1783

Add support for Jetty Reactive Streams HTTP client #1783

Uh oh!

Conversation

sdeleuze commented Apr 10, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sdeleuze Apr 11, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

poutsma left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sdeleuze commented Apr 12, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rstoyanchev Jul 5, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rstoyanchev commented Jul 5, 2018

Uh oh!

sdeleuze commented Jul 7, 2018

Uh oh!

sdeleuze commented Jul 18, 2018

Uh oh!

Uh oh!

sdeleuze Apr 11, 2018 •

edited

Loading

rstoyanchev Jul 5, 2018 •

edited

Loading