Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch '5.2.x' into master

Browse Source
This commit is contained in:
Rossen Stoyanchev
2020-10-02 14:39:27 +01:00
parent 444c8185b6 2533ba5253
commit d3087537d5
2 changed files with 21 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -21,7 +21,9 @@ import java.util.function.Consumer;
import java.util.function.Function;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.client.api.Request;
import org.eclipse.jetty.reactive.client.ContentChunk;
import org.eclipse.jetty.reactive.client.ReactiveRequest;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
@@ -81,7 +83,8 @@ public class JettyClientHttpConnector implements ClientHttpConnector {
* Constructor with an {@link JettyResourceFactory} that will manage shared resources.
* @param resourceFactory the {@link JettyResourceFactory} to use
* @param customizer the lambda used to customize the {@link HttpClient}
* @deprecated as of 5.2, in favor of {@link JettyClientHttpConnector#JettyClientHttpConnector(HttpClient, JettyResourceFactory)}
* @deprecated as of 5.2, in favor of
* {@link JettyClientHttpConnector#JettyClientHttpConnector(HttpClient, JettyResourceFactory)}
*/
@Deprecated
public JettyClientHttpConnector(JettyResourceFactory resourceFactory, @Nullable Consumer<HttpClient> customizer) {
@@ -117,14 +120,14 @@ public class JettyClientHttpConnector implements ClientHttpConnector {
}
}
JettyClientHttpRequest clientHttpRequest = new JettyClientHttpRequest(
this.httpClient.newRequest(uri).method(method.toString()), this.bufferFactory);
Request request = this.httpClient.newRequest(uri).method(method.toString());
return requestCallback.apply(clientHttpRequest).then(Mono.from(
clientHttpRequest.getReactiveRequest().response((response, chunks) -> {
Flux<DataBuffer> content = Flux.from(chunks).map(this::toDataBuffer);
return Mono.just(new JettyClientHttpResponse(response, content));
})));
return requestCallback.apply(new JettyClientHttpRequest(request, this.bufferFactory))
.then(Mono.fromDirect(ReactiveRequest.newBuilder(request).build()
.response((reactiveResponse, chunkPublisher) -> {
Flux<DataBuffer> content = Flux.from(chunkPublisher).map(this::toDataBuffer);
return Mono.just(new JettyClientHttpResponse(reactiveResponse, content));
})));
}
private DataBuffer toDataBuffer(ContentChunk chunk) {

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

@@ -24,6 +24,7 @@ import java.util.function.Function;
import org.eclipse.jetty.client.api.Request;
import org.eclipse.jetty.reactive.client.ContentChunk;
import org.eclipse.jetty.reactive.client.ReactiveRequest;
import org.eclipse.jetty.reactive.client.internal.PublisherContentProvider;
import org.eclipse.jetty.util.Callback;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
@@ -33,10 +34,10 @@ import reactor.core.publisher.MonoSink;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferFactory;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.PooledDataBuffer;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpMethod;
import org.springframework.http.MediaType;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
/**
@@ -52,9 +53,6 @@ class JettyClientHttpRequest extends AbstractClientHttpRequest {
private final DataBufferFactory bufferFactory;
@Nullable
private ReactiveRequest reactiveRequest;
public JettyClientHttpRequest(Request jettyRequest, DataBufferFactory bufferFactory) {
this.jettyRequest = jettyRequest;
@@ -93,17 +91,20 @@ class JettyClientHttpRequest extends AbstractClientHttpRequest {
@Override
public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
return Mono.<Void>create(sink -> {
Flux<ContentChunk> chunks = Flux.from(body).map(buffer -> toContentChunk(buffer, sink));
ReactiveRequest.Content content = ReactiveRequest.Content.fromPublisher(chunks, getContentType());
this.reactiveRequest = ReactiveRequest.newBuilder(this.jettyRequest).content(content).build();
ReactiveRequest.Content content = Flux.from(body)
.map(buffer -> toContentChunk(buffer, sink))
.as(chunks -> ReactiveRequest.Content.fromPublisher(chunks, getContentType()));
this.jettyRequest.content(new PublisherContentProvider(content));
sink.success();
})
.then(doCommit(this::completes));
.then(doCommit(this::completes));
}
@Override
public Mono<Void> writeAndFlushWith(Publisher<? extends Publisher<? extends DataBuffer>> body) {
return writeWith(Flux.from(body).flatMap(Function.identity()));
return writeWith(Flux.from(body)
.flatMap(Function.identity())
.doOnDiscard(PooledDataBuffer.class, DataBufferUtils::release));
}
private String getContentType() {
@@ -147,11 +148,4 @@ class JettyClientHttpRequest extends AbstractClientHttpRequest {
}
}
ReactiveRequest getReactiveRequest() {
if (this.reactiveRequest == null) {
this.reactiveRequest = ReactiveRequest.newBuilder(this.jettyRequest).build();
}
return this.reactiveRequest;
}
}