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

Add exchangeToMono and exchangeToFlux + deprecate exchange()

Browse Source 
See gh-25751
This commit is contained in:
Rossen Stoyanchev
2020-09-25 13:57:30 +01:00
parent 3e4ce9a348
commit 1404dd768f
12 changed files with 260 additions and 173 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
spring-webflux/src/main/java/org/springframework/web/reactive/function/client/ClientResponse.java
View File

@@ -45,30 +45,6 @@ import org.springframework.web.reactive.function.BodyExtractor;
* {@link ExchangeFunction}. Provides access to the response status and
* headers, and also methods to consume the response body.
*
* <p><strong>NOTE:</strong> When using a {@link ClientResponse}
* through the {@code WebClient}
* {@link WebClient.RequestHeadersSpec#exchange() exchange()} method,
* you have to make sure that the body is consumed or released by using
* one of the following methods:
* <ul>
* <li>{@link #body(BodyExtractor)}</li>
* <li>{@link #bodyToMono(Class)} or
* {@link #bodyToMono(ParameterizedTypeReference)}</li>
* <li>{@link #bodyToFlux(Class)} or
* {@link #bodyToFlux(ParameterizedTypeReference)}</li>
* <li>{@link #toEntity(Class)} or
* {@link #toEntity(ParameterizedTypeReference)}</li>
* <li>{@link #toEntityList(Class)} or
* {@link #toEntityList(ParameterizedTypeReference)}</li>
* <li>{@link #toBodilessEntity()}</li>
* <li>{@link #releaseBody()}</li>
* </ul>
* You can also use {@code bodyToMono(Void.class)} if no response content is
* expected. However keep in mind the connection will be closed, instead of
* being placed back in the pool, if any content does arrive. This is in
* contrast to {@link #releaseBody()} which does consume the full body and
* releases any content received.
*
* @author Brian Clozel
* @author Arjen Poutsma
* @since 5.0

110
spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java
View File

@@ -147,6 +147,14 @@ class DefaultWebClient implements WebClient {
return new DefaultWebClientBuilder(this.builder);
}
private static Mono<Void> releaseIfNotConsumed(ClientResponse response) {
return response.releaseBody().onErrorResume(ex2 -> Mono.empty());
}
private static <T> Mono<T> releaseIfNotConsumed(ClientResponse response, Throwable ex) {
return response.releaseBody().onErrorResume(ex2 -> Mono.empty()).then(Mono.error(ex));
}
private class DefaultRequestBodyUriSpec implements RequestBodyUriSpec {
@@ -342,6 +350,65 @@ class DefaultWebClient implements WebClient {
}
@Override
public ResponseSpec retrieve() {
return new DefaultResponseSpec(exchange(), this::createRequest);
}
private HttpRequest createRequest() {
return new HttpRequest() {
private final URI uri = initUri();
private final HttpHeaders headers = initHeaders();
@Override
public HttpMethod getMethod() {
return httpMethod;
}
@Override
public String getMethodValue() {
return httpMethod.name();
}
@Override
public URI getURI() {
return this.uri;
}
@Override
public HttpHeaders getHeaders() {
return this.headers;
}
};
}
@Override
public <V> Mono<V> exchangeToMono(Function<ClientResponse, ? extends Mono<V>> responseHandler) {
return exchange().flatMap(response -> {
try {
return responseHandler.apply(response)
.flatMap(value -> releaseIfNotConsumed(response).thenReturn(value))
.switchIfEmpty(Mono.defer(() -> releaseIfNotConsumed(response).then(Mono.empty())))
.onErrorResume(ex -> releaseIfNotConsumed(response, ex));
}
catch (Throwable ex) {
return releaseIfNotConsumed(response, ex);
}
});
}
@Override
public <V> Flux<V> exchangeToFlux(Function<ClientResponse, ? extends Flux<V>> responseHandler) {
return exchange().flatMapMany(response -> {
try {
return responseHandler.apply(response)
.concatWith(Flux.defer(() -> releaseIfNotConsumed(response).then(Mono.empty())))
.onErrorResume(ex -> releaseIfNotConsumed(response, ex));
}
catch (Throwable ex) {
return releaseIfNotConsumed(response, ex);
}
});
}
@Override
@SuppressWarnings("deprecation")
public Mono<ClientResponse> exchange() {
ClientRequest request = (this.inserter != null ?
initRequestBuilder().body(this.inserter).build() :
@@ -398,35 +465,6 @@ class DefaultWebClient implements WebClient {
return result;
}
}
@Override
public ResponseSpec retrieve() {
return new DefaultResponseSpec(exchange(), this::createRequest);
}
private HttpRequest createRequest() {
return new HttpRequest() {
private final URI uri = initUri();
private final HttpHeaders headers = initHeaders();
@Override
public HttpMethod getMethod() {
return httpMethod;
}
@Override
public String getMethodValue() {
return httpMethod.name();
}
@Override
public URI getURI() {
return this.uri;
}
@Override
public HttpHeaders getHeaders() {
return this.headers;
}
};
}
}
@@ -530,11 +568,11 @@ class DefaultWebClient implements WebClient {
Mono<? extends Throwable> exMono;
try {
exMono = handler.apply(response);
exMono = exMono.flatMap(ex -> drainBody(response, ex));
exMono = exMono.onErrorResume(ex -> drainBody(response, ex));
exMono = exMono.flatMap(ex -> releaseIfNotConsumed(response, ex));
exMono = exMono.onErrorResume(ex -> releaseIfNotConsumed(response, ex));
}
catch (Throwable ex2) {
exMono = drainBody(response, ex2);
exMono = releaseIfNotConsumed(response, ex2);
}
Mono<T> result = exMono.flatMap(Mono::error);
HttpRequest request = this.requestSupplier.get();
@@ -544,14 +582,6 @@ class DefaultWebClient implements WebClient {
return null;
}
@SuppressWarnings("unchecked")
private <T> Mono<T> drainBody(ClientResponse response, Throwable ex) {
// Ensure the body is drained, even if the StatusHandler didn't consume it,
// but ignore exception, in case the handler did consume.
return (Mono<T>) response.releaseBody()
.onErrorResume(ex2 -> Mono.empty()).thenReturn(ex);
}
private <T> Mono<T> insertCheckpoint(Mono<T> result, int statusCode, HttpRequest request) {
String httpMethod = request.getMethodValue();
URI uri = request.getURI();

97
spring-webflux/src/main/java/org/springframework/web/reactive/function/client/WebClient.java
View File

@@ -57,7 +57,8 @@ import org.springframework.web.util.UriBuilderFactory;
* <p>For examples with a response body see:
* <ul>
* <li>{@link RequestHeadersSpec#retrieve() retrieve()}
* <li>{@link RequestHeadersSpec#exchange() exchange()}
* <li>{@link RequestHeadersSpec#exchangeToMono(Function) exchangeToMono()}
* <li>{@link RequestHeadersSpec#exchangeToFlux(Function) exchangeToFlux()}
* </ul>
* <p>For examples with a request body see:
* <ul>
@@ -252,8 +253,7 @@ public interface WebClient {
Builder defaultCookies(Consumer<MultiValueMap<String, String>> cookiesConsumer);
/**
* Provide a consumer to modify every request being built just before the
* call to {@link RequestHeadersSpec#exchange() exchange()}.
* Provide a consumer to customize every request being built.
* @param defaultRequest the consumer to use for modifying requests
* @since 5.1
*/
@@ -483,21 +483,93 @@ public interface WebClient {
S httpRequest(Consumer<ClientHttpRequest> requestConsumer);
/**
* Perform the HTTP request and retrieve the response body:
* Proceed to declare how to extract the response. For example to extract
* a {@link ResponseEntity} with status, headers, and body:
* <p><pre>
* Mono&lt;Person&gt; bodyMono = client.get()
* Mono&lt;ResponseEntity&lt;Person&gt;&gt; entityMono = client.get()
* .uri("/persons/1")
* .accept(MediaType.APPLICATION_JSON)
* .retrieve()
* .toEntity(Person.class);
* </pre>
* <p>Or if interested only in the body:
* <p><pre>
* Mono&lt;Person&gt; entityMono = client.get()
* .uri("/persons/1")
* .accept(MediaType.APPLICATION_JSON)
* .retrieve()
* .bodyToMono(Person.class);
* </pre>
* <p>This method is a shortcut to using {@link #exchange()} and
* decoding the response body through {@link ClientResponse}.
* @return {@code ResponseSpec} to specify how to decode the body
* @see #exchange()
* <p>By default, 4xx and 5xx responses result in a
* {@link WebClientResponseException}. To customize error handling, use
* {@link ResponseSpec#onStatus(Predicate, Function) onStatus} handlers.
*/
ResponseSpec retrieve();
/**
* An alternative to {@link #retrieve()} that provides more control via
* access to the {@link ClientResponse}. This can be useful for advanced
* scenarios, for example to decode the response differently depending
* on the response status:
* <p><pre>
* Mono&lt;Object&gt; entityMono = client.get()
* .uri("/persons/1")
* .accept(MediaType.APPLICATION_JSON)
* .exchangeToMono(response -> {
* if (response.statusCode().equals(HttpStatus.OK)) {
* return response.bodyToMono(Person.class);
* }
* else if (response.statusCode().is4xxClientError()) {
* return response.bodyToMono(ErrorContainer.class);
* }
* else {
* return Mono.error(response.createException());
* }
* });
* </pre>
* <p><strong>Note:</strong> After the returned {@code Mono} completes,
* the response body is automatically released if it hasn't been consumed.
* If the response content is needed, the provided function must declare
* how to decode it.
* @param responseHandler the function to handle the response with
* @param <V> the type of Object the response will be transformed to
* @return a {@code Mono} produced from the response
* @since 5.3
*/
<V> Mono<V> exchangeToMono(Function<ClientResponse, ? extends Mono<V>> responseHandler);
/**
* An alternative to {@link #retrieve()} that provides more control via
* access to the {@link ClientResponse}. This can be useful for advanced
* scenarios, for example to decode the response differently depending
* on the response status:
* <p><pre>
* Mono&lt;Object&gt; entityMono = client.get()
* .uri("/persons")
* .accept(MediaType.APPLICATION_JSON)
* .exchangeToFlux(response -> {
* if (response.statusCode().equals(HttpStatus.OK)) {
* return response.bodyToFlux(Person.class);
* }
* else if (response.statusCode().is4xxClientError()) {
* return response.bodyToMono(ErrorContainer.class).flux();
* }
* else {
* return Flux.error(response.createException());
* }
* });
* </pre>
* <p><strong>Note:</strong> After the returned {@code Flux} completes,
* the response body is automatically released if it hasn't been consumed.
* If the response content is needed, the provided function must declare
* how to decode it.
* @param responseHandler the function to handle the response with
* @param <V> the type of Objects the response will be transformed to
* @return a {@code Flux} of Objects produced from the response
* @since 5.3
*/
<V> Flux<V> exchangeToFlux(Function<ClientResponse, ? extends Flux<V>> responseHandler);
/**
* Perform the HTTP request and return a {@link ClientResponse} with the
* response status and headers. You can then use methods of the response
@@ -526,7 +598,14 @@ public interface WebClient {
* if to consume the response.
* @return a {@code Mono} for the response
* @see #retrieve()
* @deprecated since 5.3 due to the possibility to leak memory and/or
* connections; please, use {@link #exchangeToMono(Function)},
* {@link #exchangeToFlux(Function)}; consider also using
* {@link #retrieve()} which provides access to the response status
* and headers via {@link ResponseEntity} along with error status
* handling.
*/
@Deprecated
Mono<ClientResponse> exchange();
}

3
spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/WebClientExtensions.kt
View File

@@ -17,8 +17,8 @@
package org.springframework.web.reactive.function.client
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.reactive.awaitSingle
import kotlinx.coroutines.reactive.asFlow
import kotlinx.coroutines.reactive.awaitSingle
import org.reactivestreams.Publisher
import org.springframework.core.ParameterizedTypeReference
import org.springframework.web.reactive.function.client.WebClient.RequestBodySpec
@@ -69,6 +69,7 @@ inline fun <reified T : Any> RequestBodySpec.body(producer: Any): RequestHeaders
* @author Sebastien Deleuze
* @since 5.2
*/
@Suppress("DEPRECATION")
suspend fun RequestHeadersSpec<out RequestHeadersSpec<*>>.awaitExchange(): ClientResponse =
exchange().awaitSingle()