From 1404dd768fd02391233d69519459df32581347fc Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Fri, 25 Sep 2020 13:57:30 +0100 Subject: [PATCH 1/3] Add exchangeToMono and exchangeToFlux + deprecate exchange() See gh-25751 --- .../function/client/ClientResponse.java | 24 ---- .../function/client/DefaultWebClient.java | 110 +++++++++++------- .../reactive/function/client/WebClient.java | 97 +++++++++++++-- .../function/client/WebClientExtensions.kt | 3 +- .../function/MultipartIntegrationTests.java | 16 +-- .../client/DefaultWebClientTests.java | 39 ++++--- .../WebClientDataBufferAllocatingTests.java | 7 +- .../client/WebClientIntegrationTests.java | 45 +++---- ...LocaleContextResolverIntegrationTests.java | 14 +-- .../annotation/MultipartIntegrationTests.java | 11 +- .../annotation/ProtobufIntegrationTests.java | 52 +++++---- ...LocaleContextResolverIntegrationTests.java | 15 +-- 12 files changed, 260 insertions(+), 173 deletions(-) diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/ClientResponse.java b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/ClientResponse.java index 1cb8790d370..897a1a28b5d 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/ClientResponse.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/ClientResponse.java @@ -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. * - *

NOTE: 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: - *

- * 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 diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java index b7f7e4fb0f0..b8c849c29af 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java @@ -147,6 +147,14 @@ class DefaultWebClient implements WebClient { return new DefaultWebClientBuilder(this.builder); } + private static Mono releaseIfNotConsumed(ClientResponse response) { + return response.releaseBody().onErrorResume(ex2 -> Mono.empty()); + } + + private static Mono 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 Mono exchangeToMono(Function> 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 Flux exchangeToFlux(Function> 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 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 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 result = exMono.flatMap(Mono::error); HttpRequest request = this.requestSupplier.get(); @@ -544,14 +582,6 @@ class DefaultWebClient implements WebClient { return null; } - @SuppressWarnings("unchecked") - private Mono 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) response.releaseBody() - .onErrorResume(ex2 -> Mono.empty()).thenReturn(ex); - } - private Mono insertCheckpoint(Mono result, int statusCode, HttpRequest request) { String httpMethod = request.getMethodValue(); URI uri = request.getURI(); diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/WebClient.java b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/WebClient.java index c7843a2d995..adc044752ff 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/WebClient.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/WebClient.java @@ -57,7 +57,8 @@ import org.springframework.web.util.UriBuilderFactory; *

For examples with a response body see: *

    *
  • {@link RequestHeadersSpec#retrieve() retrieve()} - *
  • {@link RequestHeadersSpec#exchange() exchange()} + *
  • {@link RequestHeadersSpec#exchangeToMono(Function) exchangeToMono()} + *
  • {@link RequestHeadersSpec#exchangeToFlux(Function) exchangeToFlux()} *
*

For examples with a request body see: *

    @@ -252,8 +253,7 @@ public interface WebClient { Builder defaultCookies(Consumer> 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 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: * 

    -		 * Mono<Person> bodyMono = client.get()
+		 * Mono<ResponseEntity<Person>> entityMono = client.get()
+		 *     .uri("/persons/1")
+		 *     .accept(MediaType.APPLICATION_JSON)
+		 *     .retrieve()
+		 *     .toEntity(Person.class);
+		 *
    + *

    Or if interested only in the body: + * 

    +		 * Mono<Person> entityMono = client.get()
 		 *     .uri("/persons/1")
 		 *     .accept(MediaType.APPLICATION_JSON)
 		 *     .retrieve()
 		 *     .bodyToMono(Person.class);
 		 *
    - *

    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() + *

    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: + * 

    +		 * Mono<Object> 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());
+		 *         }
+		 *     });
+		 *
    + *

    Note: 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 the type of Object the response will be transformed to + * @return a {@code Mono} produced from the response + * @since 5.3 + */ + Mono exchangeToMono(Function> 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: + * 

    +		 * Mono<Object> 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());
+		 *         }
+		 *     });
+		 *
    + *

    Note: 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 the type of Objects the response will be transformed to + * @return a {@code Flux} of Objects produced from the response + * @since 5.3 + */ + Flux exchangeToFlux(Function> 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 exchange(); } diff --git a/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/WebClientExtensions.kt b/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/WebClientExtensions.kt index 486cfb0bebb..ec1d10a740a 100644 --- a/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/WebClientExtensions.kt +++ b/spring-webflux/src/main/kotlin/org/springframework/web/reactive/function/client/WebClientExtensions.kt @@ -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 RequestBodySpec.body(producer: Any): RequestHeaders * @author Sebastien Deleuze * @since 5.2 */ +@Suppress("DEPRECATION") suspend fun RequestHeadersSpec>.awaitExchange(): ClientResponse = exchange().awaitSingle() diff --git a/spring-webflux/src/test/java/org/springframework/web/reactive/function/MultipartIntegrationTests.java b/spring-webflux/src/test/java/org/springframework/web/reactive/function/MultipartIntegrationTests.java index 0441d3fe7cc..a1f28b1b63f 100644 --- a/spring-webflux/src/test/java/org/springframework/web/reactive/function/MultipartIntegrationTests.java +++ b/spring-webflux/src/test/java/org/springframework/web/reactive/function/MultipartIntegrationTests.java @@ -29,13 +29,13 @@ import reactor.test.StepVerifier; import org.springframework.core.io.ClassPathResource; import org.springframework.http.HttpEntity; import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; import org.springframework.http.client.MultipartBodyBuilder; import org.springframework.http.codec.multipart.FilePart; import org.springframework.http.codec.multipart.FormFieldPart; import org.springframework.http.codec.multipart.Part; import org.springframework.util.FileCopyUtils; import org.springframework.util.MultiValueMap; -import org.springframework.web.reactive.function.client.ClientResponse; import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.reactive.function.server.AbstractRouterFunctionIntegrationTests; import org.springframework.web.reactive.function.server.RouterFunction; @@ -62,15 +62,16 @@ class MultipartIntegrationTests extends AbstractRouterFunctionIntegrationTests { void multipartData(HttpServer httpServer) throws Exception { startServer(httpServer); - Mono result = webClient + Mono> result = webClient .post() .uri("http://localhost:" + this.port + "/multipartData") .bodyValue(generateBody()) - .exchange(); + .retrieve() + .toEntity(Void.class); StepVerifier .create(result) - .consumeNextWith(response -> assertThat(response.statusCode()).isEqualTo(HttpStatus.OK)) + .consumeNextWith(entity -> assertThat(entity.getStatusCode()).isEqualTo(HttpStatus.OK)) .verifyComplete(); } @@ -78,15 +79,16 @@ class MultipartIntegrationTests extends AbstractRouterFunctionIntegrationTests { void parts(HttpServer httpServer) throws Exception { startServer(httpServer); - Mono result = webClient + Mono> result = webClient .post() .uri("http://localhost:" + this.port + "/parts") .bodyValue(generateBody()) - .exchange(); + .retrieve() + .toEntity(Void.class); StepVerifier .create(result) - .consumeNextWith(response -> assertThat(response.statusCode()).isEqualTo(HttpStatus.OK)) + .consumeNextWith(entity -> assertThat(entity.getStatusCode()).isEqualTo(HttpStatus.OK)) .verifyComplete(); } diff --git a/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/DefaultWebClientTests.java b/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/DefaultWebClientTests.java index a40f386d419..e5f2df9e252 100644 --- a/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/DefaultWebClientTests.java +++ b/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/DefaultWebClientTests.java @@ -47,6 +47,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.verifyNoMoreInteractions; +import static org.mockito.Mockito.when; /** * Unit tests for {@link DefaultWebClient}. @@ -69,6 +70,7 @@ public class DefaultWebClientTests { @BeforeEach public void setup() { ClientResponse mockResponse = mock(ClientResponse.class); + when(mockResponse.bodyToMono(Void.class)).thenReturn(Mono.empty()); given(this.exchangeFunction.exchange(this.captor.capture())).willReturn(Mono.just(mockResponse)); this.builder = WebClient.builder().baseUrl("/base").exchangeFunction(this.exchangeFunction); } @@ -77,7 +79,7 @@ public class DefaultWebClientTests { @Test public void basic() { this.builder.build().get().uri("/path") - .exchange().block(Duration.ofSeconds(10)); + .retrieve().bodyToMono(Void.class).block(Duration.ofSeconds(10)); ClientRequest request = verifyAndGetRequest(); assertThat(request.url().toString()).isEqualTo("/base/path"); @@ -89,7 +91,7 @@ public class DefaultWebClientTests { public void uriBuilder() { this.builder.build().get() .uri(builder -> builder.path("/path").queryParam("q", "12").build()) - .exchange().block(Duration.ofSeconds(10)); + .retrieve().bodyToMono(Void.class).block(Duration.ofSeconds(10)); ClientRequest request = verifyAndGetRequest(); assertThat(request.url().toString()).isEqualTo("/base/path?q=12"); @@ -98,8 +100,8 @@ public class DefaultWebClientTests { @Test // gh-22705 public void uriBuilderWithUriTemplate() { this.builder.build().get() - .uri("/path/{id}", builder -> builder.queryParam("q", "12").build("identifier")) - .exchange().block(Duration.ofSeconds(10)); + .uri("/path/{id}", builder -> builder.queryParam("q", "12").build("identifier")) + .retrieve().bodyToMono(Void.class).block(Duration.ofSeconds(10)); ClientRequest request = verifyAndGetRequest(); assertThat(request.url().toString()).isEqualTo("/base/path/identifier?q=12"); @@ -110,7 +112,7 @@ public class DefaultWebClientTests { public void uriBuilderWithPathOverride() { this.builder.build().get() .uri(builder -> builder.replacePath("/path").build()) - .exchange().block(Duration.ofSeconds(10)); + .retrieve().bodyToMono(Void.class).block(Duration.ofSeconds(10)); ClientRequest request = verifyAndGetRequest(); assertThat(request.url().toString()).isEqualTo("/path"); @@ -120,7 +122,7 @@ public class DefaultWebClientTests { public void requestHeaderAndCookie() { this.builder.build().get().uri("/path").accept(MediaType.APPLICATION_JSON) .cookies(cookies -> cookies.add("id", "123")) // SPR-16178 - .exchange().block(Duration.ofSeconds(10)); + .retrieve().bodyToMono(Void.class).block(Duration.ofSeconds(10)); ClientRequest request = verifyAndGetRequest(); assertThat(request.headers().getFirst("Accept")).isEqualTo("application/json"); @@ -131,7 +133,7 @@ public class DefaultWebClientTests { public void httpRequest() { this.builder.build().get().uri("/path") .httpRequest(httpRequest -> {}) - .exchange().block(Duration.ofSeconds(10)); + .retrieve().bodyToMono(Void.class).block(Duration.ofSeconds(10)); ClientRequest request = verifyAndGetRequest(); assertThat(request.httpRequest()).isNotNull(); @@ -143,7 +145,8 @@ public class DefaultWebClientTests { .defaultHeader("Accept", "application/json").defaultCookie("id", "123") .build(); - client.get().uri("/path").exchange().block(Duration.ofSeconds(10)); + client.get().uri("/path") + .retrieve().bodyToMono(Void.class).block(Duration.ofSeconds(10)); ClientRequest request = verifyAndGetRequest(); assertThat(request.headers().getFirst("Accept")).isEqualTo("application/json"); @@ -160,7 +163,7 @@ public class DefaultWebClientTests { client.get().uri("/path") .header("Accept", "application/xml") .cookie("id", "456") - .exchange().block(Duration.ofSeconds(10)); + .retrieve().bodyToMono(Void.class).block(Duration.ofSeconds(10)); ClientRequest request = verifyAndGetRequest(); assertThat(request.headers().getFirst("Accept")).isEqualTo("application/xml"); @@ -185,7 +188,7 @@ public class DefaultWebClientTests { try { context.set("bar"); client.get().uri("/path").attribute("foo", "bar") - .exchange().block(Duration.ofSeconds(10)); + .retrieve().bodyToMono(Void.class).block(Duration.ofSeconds(10)); } finally { context.remove(); @@ -271,7 +274,7 @@ public class DefaultWebClientTests { this.builder.filter(filter).build() .get().uri("/path") .attribute("foo", "bar") - .exchange().block(Duration.ofSeconds(10)); + .retrieve().bodyToMono(Void.class).block(Duration.ofSeconds(10)); assertThat(actual.get("foo")).isEqualTo("bar"); @@ -290,7 +293,7 @@ public class DefaultWebClientTests { this.builder.filter(filter).build() .get().uri("/path") .attribute("foo", null) - .exchange().block(Duration.ofSeconds(10)); + .retrieve().bodyToMono(Void.class).block(Duration.ofSeconds(10)); assertThat(actual.get("foo")).isNull(); @@ -306,7 +309,7 @@ public class DefaultWebClientTests { .defaultCookie("id", "123")) .build(); - client.get().uri("/path").exchange().block(Duration.ofSeconds(10)); + client.get().uri("/path").retrieve().bodyToMono(Void.class).block(Duration.ofSeconds(10)); ClientRequest request = verifyAndGetRequest(); assertThat(request.headers().getFirst("Accept")).isEqualTo("application/json"); @@ -317,8 +320,8 @@ public class DefaultWebClientTests { public void switchToErrorOnEmptyClientResponseMono() { ExchangeFunction exchangeFunction = mock(ExchangeFunction.class); given(exchangeFunction.exchange(any())).willReturn(Mono.empty()); - WebClient.Builder builder = WebClient.builder().baseUrl("/base").exchangeFunction(exchangeFunction); - StepVerifier.create(builder.build().get().uri("/path").exchange()) + WebClient client = WebClient.builder().baseUrl("/base").exchangeFunction(exchangeFunction).build(); + StepVerifier.create(client.get().uri("/path").retrieve().bodyToMono(Void.class)) .expectErrorMessage("The underlying HTTP client completed without emitting a response.") .verify(Duration.ofSeconds(5)); } @@ -333,9 +336,11 @@ public class DefaultWebClientTests { .build()) ) .build(); - Mono exchange = client.get().uri("/path").exchange(); + + Mono result = client.get().uri("/path").retrieve().bodyToMono(Void.class); + verifyNoInteractions(this.exchangeFunction); - exchange.block(Duration.ofSeconds(10)); + result.block(Duration.ofSeconds(10)); ClientRequest request = verifyAndGetRequest(); assertThat(request.headers().getFirst("Custom")).isEqualTo("value"); } diff --git a/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientDataBufferAllocatingTests.java b/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientDataBufferAllocatingTests.java index 4b5ec631b93..33202d48f46 100644 --- a/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientDataBufferAllocatingTests.java +++ b/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientDataBufferAllocatingTests.java @@ -182,9 +182,7 @@ class WebClientDataBufferAllocatingTests extends AbstractDataBufferAllocatingTes .setBody("foo bar")); Mono result = this.webClient.get() - .exchange() - .flatMap(ClientResponse::releaseBody); - + .exchangeToMono(ClientResponse::releaseBody); StepVerifier.create(result) .expectComplete() @@ -201,8 +199,7 @@ class WebClientDataBufferAllocatingTests extends AbstractDataBufferAllocatingTes .setBody("foo bar")); Mono> result = this.webClient.get() - .exchange() - .flatMap(ClientResponse::toBodilessEntity); + .exchangeToMono(ClientResponse::toBodilessEntity); StepVerifier.create(result) .assertNext(entity -> { diff --git a/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientIntegrationTests.java b/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientIntegrationTests.java index 50a9b9f65b3..5c61261fb28 100644 --- a/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientIntegrationTests.java +++ b/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientIntegrationTests.java @@ -837,8 +837,7 @@ class WebClientIntegrationTests { Mono result = this.webClient.get() .uri("/greeting") .header("X-Test-Header", "testvalue") - .exchange() - .flatMap(response -> response.bodyToMono(String.class)); + .retrieve().bodyToMono(String.class); StepVerifier.create(result) .expectNext("Hello Spring!") @@ -862,8 +861,7 @@ class WebClientIntegrationTests { Mono> result = this.webClient.get() .uri("/json").accept(MediaType.APPLICATION_JSON) - .exchange() - .flatMap(response -> response.toEntity(Pojo.class)); + .retrieve().toEntity(Pojo.class); StepVerifier.create(result) .consumeNextWith(entity -> { @@ -890,8 +888,7 @@ class WebClientIntegrationTests { Mono> result = this.webClient.get() .uri("/json").accept(MediaType.APPLICATION_JSON) - .exchange() - .flatMap(ClientResponse::toBodilessEntity); + .retrieve().toBodilessEntity(); StepVerifier.create(result) .consumeNextWith(entity -> { @@ -919,8 +916,7 @@ class WebClientIntegrationTests { Mono>> result = this.webClient.get() .uri("/json").accept(MediaType.APPLICATION_JSON) - .exchange() - .flatMap(response -> response.toEntityList(Pojo.class)); + .retrieve().toEntityList(Pojo.class); StepVerifier.create(result) .consumeNextWith(entity -> { @@ -948,8 +944,7 @@ class WebClientIntegrationTests { Mono> result = this.webClient.get() .uri("/noContent") - .exchange() - .flatMap(response -> response.toEntity(Void.class)); + .retrieve().toBodilessEntity(); StepVerifier.create(result) .assertNext(r -> assertThat(r.getStatusCode().is2xxSuccessful()).isTrue()) @@ -963,10 +958,11 @@ class WebClientIntegrationTests { prepareResponse(response -> response.setResponseCode(404) .setHeader("Content-Type", "text/plain").setBody("Not Found")); - Mono result = this.webClient.get().uri("/greeting").exchange(); + Mono> result = this.webClient.get().uri("/greeting") + .exchangeToMono(ClientResponse::toBodilessEntity); StepVerifier.create(result) - .consumeNextWith(response -> assertThat(response.statusCode()).isEqualTo(HttpStatus.NOT_FOUND)) + .consumeNextWith(entity -> assertThat(entity.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND)) .expectComplete() .verify(Duration.ofSeconds(3)); @@ -987,12 +983,12 @@ class WebClientIntegrationTests { prepareResponse(response -> response.setResponseCode(errorStatus) .setHeader("Content-Type", "text/plain").setBody(errorMessage)); - Mono result = this.webClient.get() + Mono> result = this.webClient.get() .uri("/unknownPage") - .exchange(); + .exchangeToMono(ClientResponse::toBodilessEntity); StepVerifier.create(result) - .consumeNextWith(response -> assertThat(response.rawStatusCode()).isEqualTo(555)) + .consumeNextWith(entity -> assertThat(entity.getStatusCodeValue()).isEqualTo(555)) .expectComplete() .verify(Duration.ofSeconds(3)); @@ -1008,7 +1004,8 @@ class WebClientIntegrationTests { startServer(connector); String uri = "/api/v4/groups/1"; - Mono responseMono = WebClient.builder().build().get().uri(uri).exchange(); + Mono> responseMono = WebClient.builder().build().get().uri(uri) + .retrieve().toBodilessEntity(); StepVerifier.create(responseMono) .expectErrorSatisfies(throwable -> { @@ -1103,12 +1100,9 @@ class WebClientIntegrationTests { .addHeader("Set-Cookie", "testkey2=testvalue2; Max-Age=42; HttpOnly; SameSite=Lax; Secure") .setBody("test")); - Mono result = this.webClient.get() + this.webClient.get() .uri("/test") - .exchange(); - - StepVerifier.create(result) - .consumeNextWith(response -> { + .exchangeToMono(response -> { assertThat(response.cookies()).containsOnlyKeys("testkey1", "testkey2"); ResponseCookie cookie1 = response.cookies().get("testkey1").get(0); @@ -1123,9 +1117,10 @@ class WebClientIntegrationTests { assertThat(cookie2.isHttpOnly()).isTrue(); assertThat(cookie2.getSameSite()).isEqualTo("Lax"); assertThat(cookie2.getMaxAge().getSeconds()).isEqualTo(42); + + return response.releaseBody(); }) - .expectComplete() - .verify(Duration.ofSeconds(3)); + .block(Duration.ofSeconds(3)); expectRequestCount(1); } @@ -1135,9 +1130,7 @@ class WebClientIntegrationTests { startServer(connector); String url = "http://example.invalid"; - Mono result = this.webClient.get(). - uri(url) - .exchange(); + Mono result = this.webClient.get().uri(url).retrieve().bodyToMono(Void.class); StepVerifier.create(result) .expectErrorSatisfies(throwable -> { diff --git a/spring-webflux/src/test/java/org/springframework/web/reactive/function/server/LocaleContextResolverIntegrationTests.java b/spring-webflux/src/test/java/org/springframework/web/reactive/function/server/LocaleContextResolverIntegrationTests.java index efc8cf8c1e2..b5a4e2bd4e6 100644 --- a/spring-webflux/src/test/java/org/springframework/web/reactive/function/server/LocaleContextResolverIntegrationTests.java +++ b/spring-webflux/src/test/java/org/springframework/web/reactive/function/server/LocaleContextResolverIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -26,8 +26,8 @@ import reactor.test.StepVerifier; import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; import org.springframework.lang.Nullable; -import org.springframework.web.reactive.function.client.ClientResponse; import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.reactive.result.view.View; import org.springframework.web.reactive.result.view.ViewResolver; @@ -66,16 +66,16 @@ class LocaleContextResolverIntegrationTests extends AbstractRouterFunctionIntegr void fixedLocale(HttpServer httpServer) throws Exception { startServer(httpServer); - Mono result = webClient + Mono> result = webClient .get() .uri("http://localhost:" + this.port + "/") - .exchange(); + .retrieve().toBodilessEntity(); StepVerifier .create(result) - .consumeNextWith(response -> { - assertThat(response.statusCode()).isEqualTo(HttpStatus.OK); - assertThat(response.headers().asHttpHeaders().getContentLanguage()).isEqualTo(Locale.GERMANY); + .consumeNextWith(entity -> { + assertThat(entity.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(entity.getHeaders().getContentLanguage()).isEqualTo(Locale.GERMANY); }) .verifyComplete(); } diff --git a/spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/MultipartIntegrationTests.java b/spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/MultipartIntegrationTests.java index 34558d112b4..1b49442ce2b 100644 --- a/spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/MultipartIntegrationTests.java +++ b/spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/MultipartIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -37,6 +37,7 @@ import org.springframework.core.io.ClassPathResource; import org.springframework.core.io.Resource; import org.springframework.http.HttpEntity; import org.springframework.http.HttpStatus; +import org.springframework.http.ResponseEntity; import org.springframework.http.client.MultipartBodyBuilder; import org.springframework.http.codec.multipart.FilePart; import org.springframework.http.codec.multipart.FormFieldPart; @@ -52,7 +53,6 @@ import org.springframework.web.bind.annotation.RequestPart; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.reactive.DispatcherHandler; import org.springframework.web.reactive.config.EnableWebFlux; -import org.springframework.web.reactive.function.client.ClientResponse; import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.server.adapter.WebHttpHandlerBuilder; import org.springframework.web.testfixture.http.server.reactive.bootstrap.AbstractHttpHandlerIntegrationTests; @@ -85,15 +85,16 @@ class MultipartIntegrationTests extends AbstractHttpHandlerIntegrationTests { void requestPart(HttpServer httpServer) throws Exception { startServer(httpServer); - Mono result = webClient + Mono> result = webClient .post() .uri("/requestPart") .bodyValue(generateBody()) - .exchange(); + .retrieve() + .toBodilessEntity(); StepVerifier .create(result) - .consumeNextWith(response -> assertThat(response.statusCode()).isEqualTo(HttpStatus.OK)) + .consumeNextWith(entity -> assertThat(entity.getStatusCode()).isEqualTo(HttpStatus.OK)) .verifyComplete(); } diff --git a/spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/ProtobufIntegrationTests.java b/spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/ProtobufIntegrationTests.java index e64264e8371..f06955d7dc4 100644 --- a/spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/ProtobufIntegrationTests.java +++ b/spring-webflux/src/test/java/org/springframework/web/reactive/result/method/annotation/ProtobufIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,6 +17,7 @@ package org.springframework.web.reactive.result.method.annotation; import java.time.Duration; +import java.util.List; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -26,6 +27,8 @@ import org.springframework.context.ApplicationContext; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; +import org.springframework.http.HttpHeaders; +import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.reactive.config.EnableWebFlux; @@ -66,18 +69,19 @@ class ProtobufIntegrationTests extends AbstractRequestMappingIntegrationTests { void value(HttpServer httpServer) throws Exception { startServer(httpServer); - Mono result = this.webClient.get() + Mono> result = this.webClient.get() .uri("/message") - .exchange() - .doOnNext(response -> { - assertThat(response.headers().contentType().get().getParameters().containsKey("delimited")).isFalse(); - assertThat(response.headers().header("X-Protobuf-Schema").get(0)).isEqualTo("sample.proto"); - assertThat(response.headers().header("X-Protobuf-Message").get(0)).isEqualTo("Msg"); - }) - .flatMap(response -> response.bodyToMono(Msg.class)); + .retrieve() + .toEntity(Msg.class); StepVerifier.create(result) - .expectNext(TEST_MSG) + .consumeNextWith(entity -> { + HttpHeaders headers = entity.getHeaders(); + assertThat(headers.getContentType().getParameters().containsKey("delimited")).isFalse(); + assertThat(headers.getFirst("X-Protobuf-Schema")).isEqualTo("sample.proto"); + assertThat(headers.getFirst("X-Protobuf-Message")).isEqualTo("Msg"); + assertThat(entity.getBody()).isEqualTo(TEST_MSG); + }) .verifyComplete(); } @@ -85,20 +89,19 @@ class ProtobufIntegrationTests extends AbstractRequestMappingIntegrationTests { void values(HttpServer httpServer) throws Exception { startServer(httpServer); - Flux result = this.webClient.get() + Mono>> result = this.webClient.get() .uri("/messages") - .exchange() - .doOnNext(response -> { - assertThat(response.headers().contentType().get().getParameters().get("delimited")).isEqualTo("true"); - assertThat(response.headers().header("X-Protobuf-Schema").get(0)).isEqualTo("sample.proto"); - assertThat(response.headers().header("X-Protobuf-Message").get(0)).isEqualTo("Msg"); - }) - .flatMapMany(response -> response.bodyToFlux(Msg.class)); + .retrieve() + .toEntityList(Msg.class); StepVerifier.create(result) - .expectNext(TEST_MSG) - .expectNext(TEST_MSG) - .expectNext(TEST_MSG) + .consumeNextWith(entity -> { + HttpHeaders headers = entity.getHeaders(); + assertThat(headers.getContentType().getParameters().get("delimited")).isEqualTo("true"); + assertThat(headers.getFirst("X-Protobuf-Schema")).isEqualTo("sample.proto"); + assertThat(headers.getFirst("X-Protobuf-Message")).isEqualTo("Msg"); + assertThat(entity.getBody()).containsExactly(TEST_MSG, TEST_MSG, TEST_MSG); + }) .verifyComplete(); } @@ -108,13 +111,12 @@ class ProtobufIntegrationTests extends AbstractRequestMappingIntegrationTests { Flux result = this.webClient.get() .uri("/message-stream") - .exchange() - .doOnNext(response -> { + .exchangeToFlux(response -> { assertThat(response.headers().contentType().get().getParameters().get("delimited")).isEqualTo("true"); assertThat(response.headers().header("X-Protobuf-Schema").get(0)).isEqualTo("sample.proto"); assertThat(response.headers().header("X-Protobuf-Message").get(0)).isEqualTo("Msg"); - }) - .flatMapMany(response -> response.bodyToFlux(Msg.class)); + return response.bodyToFlux(Msg.class); + }); StepVerifier.create(result) .expectNext(Msg.newBuilder().setFoo("Foo").setBlah(SecondMsg.newBuilder().setBlah(0).build()).build()) diff --git a/spring-webflux/src/test/java/org/springframework/web/reactive/result/view/LocaleContextResolverIntegrationTests.java b/spring-webflux/src/test/java/org/springframework/web/reactive/result/view/LocaleContextResolverIntegrationTests.java index b21b68cd5b7..bf0e326396e 100644 --- a/spring-webflux/src/test/java/org/springframework/web/reactive/result/view/LocaleContextResolverIntegrationTests.java +++ b/spring-webflux/src/test/java/org/springframework/web/reactive/result/view/LocaleContextResolverIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -30,12 +30,12 @@ import org.springframework.context.annotation.ComponentScan; import org.springframework.context.annotation.Configuration; import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; import org.springframework.lang.Nullable; import org.springframework.stereotype.Controller; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.reactive.config.ViewResolverRegistry; import org.springframework.web.reactive.config.WebFluxConfigurationSupport; -import org.springframework.web.reactive.function.client.ClientResponse; import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.reactive.result.method.annotation.AbstractRequestMappingIntegrationTests; import org.springframework.web.server.ServerWebExchange; @@ -66,15 +66,16 @@ class LocaleContextResolverIntegrationTests extends AbstractRequestMappingIntegr void fixedLocale(HttpServer httpServer) throws Exception { startServer(httpServer); - Mono result = webClient + Mono> result = webClient .get() .uri("http://localhost:" + this.port + "/") - .exchange(); + .retrieve() + .toBodilessEntity(); StepVerifier.create(result) - .consumeNextWith(response -> { - assertThat(response.statusCode()).isEqualTo(HttpStatus.OK); - assertThat(response.headers().asHttpHeaders().getContentLanguage()).isEqualTo(Locale.GERMANY); + .consumeNextWith(entity -> { + assertThat(entity.getStatusCode()).isEqualTo(HttpStatus.OK); + assertThat(entity.getHeaders().getContentLanguage()).isEqualTo(Locale.GERMANY); }) .verifyComplete(); } From db9190e0e61c481e1062fb6b78f5c6ebdbfdb086 Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Fri, 25 Sep 2020 16:41:26 +0100 Subject: [PATCH 2/3] Remove use of WebClient#exchange from WebTestClient The exchange() method is now deprecated because it is not safe for general use but that doesn't apply to the WebTestClient because it exposes a different higher level API for response handling that ensures the response is consumed. Nevertheless WebTestClient cannot call WebClient.exchange() any more. To fix this WebTestClient no longer delegates to WebClient and thus gains direct access to the underlying ExchangeFunction. This is not a big deal because WebClient and WebTestClient do the same only when it comes to gathering builder options and request input. See gh-25751 --- .../reactive/server/DefaultWebTestClient.java | 201 ++++++++++----- .../server/DefaultWebTestClientBuilder.java | 229 ++++++++++++++---- .../function/client/DefaultWebClient.java | 19 +- .../client/DefaultWebClientBuilder.java | 14 +- .../client/DefaultWebClientTests.java | 2 +- .../client/WebClientExtensionsTests.kt | 1 + 6 files changed, 345 insertions(+), 121 deletions(-) diff --git a/spring-test/src/main/java/org/springframework/test/web/reactive/server/DefaultWebTestClient.java b/spring-test/src/main/java/org/springframework/test/web/reactive/server/DefaultWebTestClient.java index 98bf4361bcc..ef6bde25d26 100644 --- a/spring-test/src/main/java/org/springframework/test/web/reactive/server/DefaultWebTestClient.java +++ b/spring-test/src/main/java/org/springframework/test/web/reactive/server/DefaultWebTestClient.java @@ -22,6 +22,7 @@ import java.nio.charset.StandardCharsets; import java.time.Duration; import java.time.ZonedDateTime; import java.util.Arrays; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Optional; @@ -46,12 +47,17 @@ import org.springframework.test.util.AssertionErrors; import org.springframework.test.util.JsonExpectationsHelper; import org.springframework.test.util.XmlExpectationsHelper; import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; +import org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MimeType; import org.springframework.util.MultiValueMap; import org.springframework.web.reactive.function.BodyInserter; +import org.springframework.web.reactive.function.BodyInserters; +import org.springframework.web.reactive.function.client.ClientRequest; import org.springframework.web.reactive.function.client.ClientResponse; -import org.springframework.web.reactive.function.client.WebClient; +import org.springframework.web.reactive.function.client.ExchangeFunction; import org.springframework.web.util.UriBuilder; +import org.springframework.web.util.UriBuilderFactory; /** * Default implementation of {@link WebTestClient}. @@ -61,30 +67,42 @@ import org.springframework.web.util.UriBuilder; */ class DefaultWebTestClient implements WebTestClient { - private final WebClient webClient; - private final WiretapConnector wiretapConnector; - private final Duration timeout; + private final ExchangeFunction exchangeFunction; + + private final UriBuilderFactory uriBuilderFactory; + + @Nullable + private final HttpHeaders defaultHeaders; + + @Nullable + private final MultiValueMap defaultCookies; + + private final Duration responseTimeout; private final DefaultWebTestClientBuilder builder; private final AtomicLong requestIndex = new AtomicLong(); - DefaultWebTestClient(WebClient.Builder clientBuilder, ClientHttpConnector connector, - @Nullable Duration timeout, DefaultWebTestClientBuilder webTestClientBuilder) { + DefaultWebTestClient(ClientHttpConnector connector, + Function exchangeFactory, UriBuilderFactory uriBuilderFactory, + @Nullable HttpHeaders headers, @Nullable MultiValueMap cookies, + @Nullable Duration responseTimeout, DefaultWebTestClientBuilder clientBuilder) { - Assert.notNull(clientBuilder, "WebClient.Builder is required"); this.wiretapConnector = new WiretapConnector(connector); - this.webClient = clientBuilder.clientConnector(this.wiretapConnector).build(); - this.timeout = (timeout != null ? timeout : Duration.ofSeconds(5)); - this.builder = webTestClientBuilder; + this.exchangeFunction = exchangeFactory.apply(this.wiretapConnector); + this.uriBuilderFactory = uriBuilderFactory; + this.defaultHeaders = headers; + this.defaultCookies = cookies; + this.responseTimeout = (responseTimeout != null ? responseTimeout : Duration.ofSeconds(5)); + this.builder = clientBuilder; } - private Duration getTimeout() { - return this.timeout; + private Duration getResponseTimeout() { + return this.responseTimeout; } @@ -124,12 +142,12 @@ class DefaultWebTestClient implements WebTestClient { } @Override - public RequestBodyUriSpec method(HttpMethod method) { - return methodInternal(method); + public RequestBodyUriSpec method(HttpMethod httpMethod) { + return methodInternal(httpMethod); } - private RequestBodyUriSpec methodInternal(HttpMethod method) { - return new DefaultRequestBodyUriSpec(this.webClient.method(method)); + private RequestBodyUriSpec methodInternal(HttpMethod httpMethod) { + return new DefaultRequestBodyUriSpec(httpMethod); } @Override @@ -145,154 +163,180 @@ class DefaultWebTestClient implements WebTestClient { private class DefaultRequestBodyUriSpec implements RequestBodyUriSpec { - private final WebClient.RequestBodyUriSpec bodySpec; + private final HttpMethod httpMethod; + + @Nullable + private URI uri; + + private final HttpHeaders headers; + + @Nullable + private MultiValueMap cookies; + + @Nullable + private BodyInserter inserter; + + private final Map attributes = new LinkedHashMap<>(4); + + @Nullable + private Consumer httpRequestConsumer; @Nullable private String uriTemplate; private final String requestId; - DefaultRequestBodyUriSpec(WebClient.RequestBodyUriSpec spec) { - this.bodySpec = spec; + DefaultRequestBodyUriSpec(HttpMethod httpMethod) { + this.httpMethod = httpMethod; this.requestId = String.valueOf(requestIndex.incrementAndGet()); - this.bodySpec.header(WebTestClient.WEBTESTCLIENT_REQUEST_ID, this.requestId); + this.headers = new HttpHeaders(); + this.headers.add(WebTestClient.WEBTESTCLIENT_REQUEST_ID, this.requestId); } @Override public RequestBodySpec uri(String uriTemplate, Object... uriVariables) { - this.bodySpec.uri(uriTemplate, uriVariables); this.uriTemplate = uriTemplate; - return this; + return uri(uriBuilderFactory.expand(uriTemplate, uriVariables)); } @Override public RequestBodySpec uri(String uriTemplate, Map uriVariables) { - this.bodySpec.uri(uriTemplate, uriVariables); this.uriTemplate = uriTemplate; - return this; + return uri(uriBuilderFactory.expand(uriTemplate, uriVariables)); } @Override public RequestBodySpec uri(Function uriFunction) { - this.bodySpec.uri(uriFunction); this.uriTemplate = null; - return this; + return uri(uriFunction.apply(uriBuilderFactory.builder())); } @Override public RequestBodySpec uri(URI uri) { - this.bodySpec.uri(uri); this.uriTemplate = null; + this.uri = uri; return this; } + private HttpHeaders getHeaders() { + return this.headers; + } + + private MultiValueMap getCookies() { + if (this.cookies == null) { + this.cookies = new LinkedMultiValueMap<>(3); + } + return this.cookies; + } + @Override public RequestBodySpec header(String headerName, String... headerValues) { - this.bodySpec.header(headerName, headerValues); + for (String headerValue : headerValues) { + getHeaders().add(headerName, headerValue); + } return this; } @Override public RequestBodySpec headers(Consumer headersConsumer) { - this.bodySpec.headers(headersConsumer); + headersConsumer.accept(getHeaders()); return this; } @Override public RequestBodySpec attribute(String name, Object value) { - this.bodySpec.attribute(name, value); + this.attributes.put(name, value); return this; } @Override - public RequestBodySpec attributes( - Consumer> attributesConsumer) { - this.bodySpec.attributes(attributesConsumer); + public RequestBodySpec attributes(Consumer> attributesConsumer) { + attributesConsumer.accept(this.attributes); return this; } @Override public RequestBodySpec accept(MediaType... acceptableMediaTypes) { - this.bodySpec.accept(acceptableMediaTypes); + getHeaders().setAccept(Arrays.asList(acceptableMediaTypes)); return this; } @Override public RequestBodySpec acceptCharset(Charset... acceptableCharsets) { - this.bodySpec.acceptCharset(acceptableCharsets); + getHeaders().setAcceptCharset(Arrays.asList(acceptableCharsets)); return this; } @Override public RequestBodySpec contentType(MediaType contentType) { - this.bodySpec.contentType(contentType); + getHeaders().setContentType(contentType); return this; } @Override public RequestBodySpec contentLength(long contentLength) { - this.bodySpec.contentLength(contentLength); + getHeaders().setContentLength(contentLength); return this; } @Override public RequestBodySpec cookie(String name, String value) { - this.bodySpec.cookie(name, value); + getCookies().add(name, value); return this; } @Override - public RequestBodySpec cookies( - Consumer> cookiesConsumer) { - this.bodySpec.cookies(cookiesConsumer); + public RequestBodySpec cookies(Consumer> cookiesConsumer) { + cookiesConsumer.accept(getCookies()); return this; } @Override public RequestBodySpec ifModifiedSince(ZonedDateTime ifModifiedSince) { - this.bodySpec.ifModifiedSince(ifModifiedSince); + getHeaders().setIfModifiedSince(ifModifiedSince); return this; } @Override public RequestBodySpec ifNoneMatch(String... ifNoneMatches) { - this.bodySpec.ifNoneMatch(ifNoneMatches); + getHeaders().setIfNoneMatch(Arrays.asList(ifNoneMatches)); return this; } @Override public RequestHeadersSpec bodyValue(Object body) { - this.bodySpec.bodyValue(body); + this.inserter = BodyInserters.fromValue(body); return this; } @Override - public > RequestHeadersSpec body(S publisher, Class elementClass) { - this.bodySpec.body(publisher, elementClass); + public > RequestHeadersSpec body( + P publisher, ParameterizedTypeReference elementTypeRef) { + this.inserter = BodyInserters.fromPublisher(publisher, elementTypeRef); return this; } @Override - public > RequestHeadersSpec body(S publisher, ParameterizedTypeReference elementTypeRef) { - this.bodySpec.body(publisher, elementTypeRef); + public > RequestHeadersSpec body(P publisher, Class elementClass) { + this.inserter = BodyInserters.fromPublisher(publisher, elementClass); return this; } @Override public RequestHeadersSpec body(Object producer, Class elementClass) { - this.bodySpec.body(producer, elementClass); + this.inserter = BodyInserters.fromProducer(producer, elementClass); return this; } @Override public RequestHeadersSpec body(Object producer, ParameterizedTypeReference elementTypeRef) { - this.bodySpec.body(producer, elementTypeRef); + this.inserter = BodyInserters.fromProducer(producer, elementTypeRef); return this; } @Override public RequestHeadersSpec body(BodyInserter inserter) { - this.bodySpec.body(inserter); + this.inserter = inserter; return this; } @@ -304,10 +348,57 @@ class DefaultWebTestClient implements WebTestClient { @Override public ResponseSpec exchange() { - ClientResponse clientResponse = this.bodySpec.exchange().block(getTimeout()); - Assert.state(clientResponse != null, "No ClientResponse"); - ExchangeResult result = wiretapConnector.getExchangeResult(this.requestId, this.uriTemplate, getTimeout()); - return new DefaultResponseSpec(result, clientResponse, getTimeout()); + ClientRequest request = (this.inserter != null ? + initRequestBuilder().body(this.inserter).build() : + initRequestBuilder().build()); + + ClientResponse response = exchangeFunction.exchange(request).block(getResponseTimeout()); + Assert.state(response != null, "No ClientResponse"); + + ExchangeResult result = wiretapConnector.getExchangeResult( + this.requestId, this.uriTemplate, getResponseTimeout()); + + return new DefaultResponseSpec(result, response, getResponseTimeout()); + } + + private ClientRequest.Builder initRequestBuilder() { + ClientRequest.Builder builder = ClientRequest.create(this.httpMethod, initUri()) + .headers(headers -> headers.addAll(initHeaders())) + .cookies(cookies -> cookies.addAll(initCookies())) + .attributes(attributes -> attributes.putAll(this.attributes)); + if (this.httpRequestConsumer != null) { + builder.httpRequest(this.httpRequestConsumer); + } + return builder; + } + + private URI initUri() { + return (this.uri != null ? this.uri : uriBuilderFactory.expand("")); + } + + private HttpHeaders initHeaders() { + if (CollectionUtils.isEmpty(defaultHeaders)) { + return this.headers; + } + HttpHeaders result = new HttpHeaders(); + result.putAll(defaultHeaders); + result.putAll(this.headers); + return result; + } + + private MultiValueMap initCookies() { + if (CollectionUtils.isEmpty(this.cookies)) { + return (defaultCookies != null ? defaultCookies : new LinkedMultiValueMap<>()); + } + else if (CollectionUtils.isEmpty(defaultCookies)) { + return this.cookies; + } + else { + MultiValueMap result = new LinkedMultiValueMap<>(); + result.putAll(defaultCookies); + result.putAll(this.cookies); + return result; + } } } diff --git a/spring-test/src/main/java/org/springframework/test/web/reactive/server/DefaultWebTestClientBuilder.java b/spring-test/src/main/java/org/springframework/test/web/reactive/server/DefaultWebTestClientBuilder.java index 69279499fe7..8e1904f3114 100644 --- a/spring-test/src/main/java/org/springframework/test/web/reactive/server/DefaultWebTestClientBuilder.java +++ b/spring-test/src/main/java/org/springframework/test/web/reactive/server/DefaultWebTestClientBuilder.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -17,20 +17,30 @@ package org.springframework.test.web.reactive.server; import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; import java.util.List; import java.util.function.Consumer; +import java.util.function.Function; import org.springframework.http.HttpHeaders; import org.springframework.http.client.reactive.ClientHttpConnector; +import org.springframework.http.client.reactive.HttpComponentsClientHttpConnector; +import org.springframework.http.client.reactive.JettyClientHttpConnector; import org.springframework.http.client.reactive.ReactorClientHttpConnector; import org.springframework.http.codec.ClientCodecConfigurer; import org.springframework.lang.Nullable; import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; +import org.springframework.util.CollectionUtils; +import org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; import org.springframework.web.reactive.function.client.ExchangeFilterFunction; +import org.springframework.web.reactive.function.client.ExchangeFunction; +import org.springframework.web.reactive.function.client.ExchangeFunctions; import org.springframework.web.reactive.function.client.ExchangeStrategies; -import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.server.adapter.WebHttpHandlerBuilder; +import org.springframework.web.util.DefaultUriBuilderFactory; import org.springframework.web.util.UriBuilderFactory; /** @@ -41,7 +51,21 @@ import org.springframework.web.util.UriBuilderFactory; */ class DefaultWebTestClientBuilder implements WebTestClient.Builder { - private final WebClient.Builder webClientBuilder; + private static final boolean reactorClientPresent; + + private static final boolean jettyClientPresent; + + private static final boolean httpComponentsClientPresent; + + static { + ClassLoader loader = DefaultWebTestClientBuilder.class.getClassLoader(); + reactorClientPresent = ClassUtils.isPresent("reactor.netty.http.client.HttpClient", loader); + jettyClientPresent = ClassUtils.isPresent("org.eclipse.jetty.client.HttpClient", loader); + httpComponentsClientPresent = + ClassUtils.isPresent("org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient", loader) && + ClassUtils.isPresent("org.apache.hc.core5.reactive.ReactiveDataConsumer", loader); + } + @Nullable private final WebHttpHandlerBuilder httpHandlerBuilder; @@ -49,136 +73,243 @@ class DefaultWebTestClientBuilder implements WebTestClient.Builder { @Nullable private final ClientHttpConnector connector; + @Nullable + private String baseUrl; + + @Nullable + private UriBuilderFactory uriBuilderFactory; + + @Nullable + private HttpHeaders defaultHeaders; + + @Nullable + private MultiValueMap defaultCookies; + + @Nullable + private List filters; + + @Nullable + private ExchangeStrategies strategies; + + @Nullable + private List> strategiesConfigurers; + @Nullable private Duration responseTimeout; - /** Connect to server via Reactor Netty. */ + /** Determine connector via classpath detection. */ DefaultWebTestClientBuilder() { - this(new ReactorClientHttpConnector()); - } - - /** Connect to server through the given connector. */ - DefaultWebTestClientBuilder(ClientHttpConnector connector) { - this(null, null, connector, null); + this(null, null); } - /** Connect to given mock server with mock request and response. */ + /** Use HttpHandlerConnector with mock server. */ DefaultWebTestClientBuilder(WebHttpHandlerBuilder httpHandlerBuilder) { - this(null, httpHandlerBuilder, null, null); + this(httpHandlerBuilder, null); } - /** Copy constructor. */ - DefaultWebTestClientBuilder(DefaultWebTestClientBuilder other) { - this(other.webClientBuilder.clone(), other.httpHandlerBuilder, other.connector, - other.responseTimeout); + /** Use given connector. */ + DefaultWebTestClientBuilder(ClientHttpConnector connector) { + this(null, connector); } - private DefaultWebTestClientBuilder(@Nullable WebClient.Builder webClientBuilder, - @Nullable WebHttpHandlerBuilder httpHandlerBuilder, @Nullable ClientHttpConnector connector, - @Nullable Duration responseTimeout) { + DefaultWebTestClientBuilder( + @Nullable WebHttpHandlerBuilder httpHandlerBuilder, @Nullable ClientHttpConnector connector) { - Assert.isTrue(httpHandlerBuilder != null || connector != null, - "Either WebHttpHandlerBuilder or ClientHttpConnector must be provided"); + Assert.isTrue(httpHandlerBuilder == null || connector == null, + "Expected WebHttpHandlerBuilder or ClientHttpConnector but not both."); - this.webClientBuilder = (webClientBuilder != null ? webClientBuilder : WebClient.builder()); - this.httpHandlerBuilder = (httpHandlerBuilder != null ? httpHandlerBuilder.clone() : null); this.connector = connector; - this.responseTimeout = responseTimeout; + this.httpHandlerBuilder = (httpHandlerBuilder != null ? httpHandlerBuilder.clone() : null); + } + + /** Copy constructor. */ + DefaultWebTestClientBuilder(DefaultWebTestClientBuilder other) { + this.httpHandlerBuilder = (other.httpHandlerBuilder != null ? other.httpHandlerBuilder.clone() : null); + this.connector = other.connector; + this.responseTimeout = other.responseTimeout; + + this.baseUrl = other.baseUrl; + this.uriBuilderFactory = other.uriBuilderFactory; + if (other.defaultHeaders != null) { + this.defaultHeaders = new HttpHeaders(); + this.defaultHeaders.putAll(other.defaultHeaders); + } + else { + this.defaultHeaders = null; + } + this.defaultCookies = (other.defaultCookies != null ? + new LinkedMultiValueMap<>(other.defaultCookies) : null); + this.filters = (other.filters != null ? new ArrayList<>(other.filters) : null); + this.strategies = other.strategies; + this.strategiesConfigurers = (other.strategiesConfigurers != null ? + new ArrayList<>(other.strategiesConfigurers) : null); } @Override public WebTestClient.Builder baseUrl(String baseUrl) { - this.webClientBuilder.baseUrl(baseUrl); + this.baseUrl = baseUrl; return this; } @Override public WebTestClient.Builder uriBuilderFactory(UriBuilderFactory uriBuilderFactory) { - this.webClientBuilder.uriBuilderFactory(uriBuilderFactory); + this.uriBuilderFactory = uriBuilderFactory; return this; } @Override - public WebTestClient.Builder defaultHeader(String headerName, String... headerValues) { - this.webClientBuilder.defaultHeader(headerName, headerValues); + public WebTestClient.Builder defaultHeader(String header, String... values) { + initHeaders().put(header, Arrays.asList(values)); return this; } @Override public WebTestClient.Builder defaultHeaders(Consumer headersConsumer) { - this.webClientBuilder.defaultHeaders(headersConsumer); + headersConsumer.accept(initHeaders()); return this; } + private HttpHeaders initHeaders() { + if (this.defaultHeaders == null) { + this.defaultHeaders = new HttpHeaders(); + } + return this.defaultHeaders; + } + @Override - public WebTestClient.Builder defaultCookie(String cookieName, String... cookieValues) { - this.webClientBuilder.defaultCookie(cookieName, cookieValues); + public WebTestClient.Builder defaultCookie(String cookie, String... values) { + initCookies().addAll(cookie, Arrays.asList(values)); return this; } @Override - public WebTestClient.Builder defaultCookies( - Consumer> cookiesConsumer) { - this.webClientBuilder.defaultCookies(cookiesConsumer); + public WebTestClient.Builder defaultCookies(Consumer> cookiesConsumer) { + cookiesConsumer.accept(initCookies()); return this; } + private MultiValueMap initCookies() { + if (this.defaultCookies == null) { + this.defaultCookies = new LinkedMultiValueMap<>(3); + } + return this.defaultCookies; + } + @Override public WebTestClient.Builder filter(ExchangeFilterFunction filter) { - this.webClientBuilder.filter(filter); + Assert.notNull(filter, "ExchangeFilterFunction must not be null"); + initFilters().add(filter); return this; } @Override public WebTestClient.Builder filters(Consumer> filtersConsumer) { - this.webClientBuilder.filters(filtersConsumer); + filtersConsumer.accept(initFilters()); return this; } + private List initFilters() { + if (this.filters == null) { + this.filters = new ArrayList<>(); + } + return this.filters; + } + @Override public WebTestClient.Builder codecs(Consumer configurer) { - this.webClientBuilder.codecs(configurer); + if (this.strategiesConfigurers == null) { + this.strategiesConfigurers = new ArrayList<>(4); + } + this.strategiesConfigurers.add(builder -> builder.codecs(configurer)); return this; } @Override public WebTestClient.Builder exchangeStrategies(ExchangeStrategies strategies) { - this.webClientBuilder.exchangeStrategies(strategies); + this.strategies = strategies; return this; } - @SuppressWarnings("deprecation") @Override + @SuppressWarnings("deprecation") public WebTestClient.Builder exchangeStrategies(Consumer configurer) { - this.webClientBuilder.exchangeStrategies(configurer); + if (this.strategiesConfigurers == null) { + this.strategiesConfigurers = new ArrayList<>(4); + } + this.strategiesConfigurers.add(configurer); return this; } @Override - public WebTestClient.Builder responseTimeout(Duration timeout) { - this.responseTimeout = timeout; + public WebTestClient.Builder apply(WebTestClientConfigurer configurer) { + configurer.afterConfigurerAdded(this, this.httpHandlerBuilder, this.connector); return this; } @Override - public WebTestClient.Builder apply(WebTestClientConfigurer configurer) { - configurer.afterConfigurerAdded(this, this.httpHandlerBuilder, this.connector); + public WebTestClient.Builder responseTimeout(Duration timeout) { + this.responseTimeout = timeout; return this; } - @Override public WebTestClient build() { ClientHttpConnector connectorToUse = this.connector; if (connectorToUse == null) { - Assert.state(this.httpHandlerBuilder != null, "No WebHttpHandlerBuilder available"); - connectorToUse = new HttpHandlerConnector(this.httpHandlerBuilder.build()); + if (this.httpHandlerBuilder != null) { + connectorToUse = new HttpHandlerConnector(this.httpHandlerBuilder.build()); + } + } + if (connectorToUse == null) { + connectorToUse = initConnector(); + } + Function exchangeFactory = connector -> { + ExchangeFunction exchange = ExchangeFunctions.create(connector, initExchangeStrategies()); + if (CollectionUtils.isEmpty(this.filters)) { + return exchange; + } + return this.filters.stream() + .reduce(ExchangeFilterFunction::andThen) + .map(filter -> filter.apply(exchange)) + .orElse(exchange); + + }; + return new DefaultWebTestClient(connectorToUse, exchangeFactory, initUriBuilderFactory(), + this.defaultHeaders != null ? HttpHeaders.readOnlyHttpHeaders(this.defaultHeaders) : null, + this.defaultCookies != null ? CollectionUtils.unmodifiableMultiValueMap(this.defaultCookies) : null, + this.responseTimeout, new DefaultWebTestClientBuilder(this)); + } + + private static ClientHttpConnector initConnector() { + if (reactorClientPresent) { + return new ReactorClientHttpConnector(); + } + else if (jettyClientPresent) { + return new JettyClientHttpConnector(); + } + else if (httpComponentsClientPresent) { + return new HttpComponentsClientHttpConnector(); } + throw new IllegalStateException("No suitable default ClientHttpConnector found"); + } - return new DefaultWebTestClient(this.webClientBuilder, - connectorToUse, this.responseTimeout, new DefaultWebTestClientBuilder(this)); + private ExchangeStrategies initExchangeStrategies() { + if (CollectionUtils.isEmpty(this.strategiesConfigurers)) { + return (this.strategies != null ? this.strategies : ExchangeStrategies.withDefaults()); + } + ExchangeStrategies.Builder builder = + (this.strategies != null ? this.strategies.mutate() : ExchangeStrategies.builder()); + this.strategiesConfigurers.forEach(configurer -> configurer.accept(builder)); + return builder.build(); } + private UriBuilderFactory initUriBuilderFactory() { + if (this.uriBuilderFactory != null) { + return this.uriBuilderFactory; + } + return (this.baseUrl != null ? + new DefaultUriBuilderFactory(this.baseUrl) : new DefaultUriBuilderFactory()); + } } diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java index b8c849c29af..9431b263b4b 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java @@ -49,7 +49,6 @@ import org.springframework.util.LinkedMultiValueMap; import org.springframework.util.MultiValueMap; import org.springframework.web.reactive.function.BodyInserter; import org.springframework.web.reactive.function.BodyInserters; -import org.springframework.web.util.DefaultUriBuilderFactory; import org.springframework.web.util.UriBuilder; import org.springframework.web.util.UriBuilderFactory; @@ -85,12 +84,12 @@ class DefaultWebClient implements WebClient { private final DefaultWebClientBuilder builder; - DefaultWebClient(ExchangeFunction exchangeFunction, @Nullable UriBuilderFactory factory, + DefaultWebClient(ExchangeFunction exchangeFunction, UriBuilderFactory uriBuilderFactory, @Nullable HttpHeaders defaultHeaders, @Nullable MultiValueMap defaultCookies, @Nullable Consumer> defaultRequest, DefaultWebClientBuilder builder) { this.exchangeFunction = exchangeFunction; - this.uriBuilderFactory = (factory != null ? factory : new DefaultUriBuilderFactory()); + this.uriBuilderFactory = uriBuilderFactory; this.defaultHeaders = defaultHeaders; this.defaultCookies = defaultCookies; this.defaultRequest = defaultRequest; @@ -251,13 +250,6 @@ class DefaultWebClient implements WebClient { return this; } - @Override - public RequestBodySpec httpRequest(Consumer requestConsumer) { - this.httpRequestConsumer = (this.httpRequestConsumer != null ? - this.httpRequestConsumer.andThen(requestConsumer) : requestConsumer); - return this; - } - @Override public DefaultRequestBodyUriSpec accept(MediaType... acceptableMediaTypes) { getHeaders().setAccept(Arrays.asList(acceptableMediaTypes)); @@ -306,6 +298,13 @@ class DefaultWebClient implements WebClient { return this; } + @Override + public RequestBodySpec httpRequest(Consumer requestConsumer) { + this.httpRequestConsumer = (this.httpRequestConsumer != null ? + this.httpRequestConsumer.andThen(requestConsumer) : requestConsumer); + return this; + } + @Override public RequestHeadersSpec bodyValue(Object body) { this.inserter = BodyInserters.fromValue(body); diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClientBuilder.java b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClientBuilder.java index 80e79090b0e..c8c0727cadf 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClientBuilder.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClientBuilder.java @@ -262,24 +262,26 @@ final class DefaultWebClientBuilder implements WebClient.Builder { @Override public WebClient build() { + ClientHttpConnector connectorToUse = + (this.connector != null ? this.connector : initConnector()); + ExchangeFunction exchange = (this.exchangeFunction == null ? - ExchangeFunctions.create(getOrInitConnector(), initExchangeStrategies()) : + ExchangeFunctions.create(connectorToUse, initExchangeStrategies()) : this.exchangeFunction); + ExchangeFunction filteredExchange = (this.filters != null ? this.filters.stream() .reduce(ExchangeFilterFunction::andThen) .map(filter -> filter.apply(exchange)) .orElse(exchange) : exchange); + return new DefaultWebClient(filteredExchange, initUriBuilderFactory(), this.defaultHeaders != null ? HttpHeaders.readOnlyHttpHeaders(this.defaultHeaders) : null, this.defaultCookies != null ? CollectionUtils.unmodifiableMultiValueMap(this.defaultCookies) : null, this.defaultRequest, new DefaultWebClientBuilder(this)); } - private ClientHttpConnector getOrInitConnector() { - if (this.connector != null) { - return this.connector; - } - else if (reactorClientPresent) { + private ClientHttpConnector initConnector() { + if (reactorClientPresent) { return new ReactorClientHttpConnector(); } else if (jettyClientPresent) { diff --git a/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/DefaultWebClientTests.java b/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/DefaultWebClientTests.java index e5f2df9e252..db58a16560a 100644 --- a/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/DefaultWebClientTests.java +++ b/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/DefaultWebClientTests.java @@ -43,11 +43,11 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; import static org.mockito.ArgumentMatchers.any; import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.when; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.verifyNoMoreInteractions; -import static org.mockito.Mockito.when; /** * Unit tests for {@link DefaultWebClient}. diff --git a/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/WebClientExtensionsTests.kt b/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/WebClientExtensionsTests.kt index c5ec1002b5c..2811cf6cc5c 100644 --- a/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/WebClientExtensionsTests.kt +++ b/spring-webflux/src/test/kotlin/org/springframework/web/reactive/function/client/WebClientExtensionsTests.kt @@ -80,6 +80,7 @@ class WebClientExtensionsTests { } @Test + @Suppress("DEPRECATION") fun awaitExchange() { val response = mockk() every { requestBodySpec.exchange() } returns Mono.just(response) From 10c5f85a9f8fab72f4b9abbd288832573ecbc4ff Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Fri, 25 Sep 2020 21:14:57 +0100 Subject: [PATCH 3/3] WebTestClient documentation updates See gh-25751 --- src/docs/asciidoc/web/webflux-webclient.adoc | 171 +++++++++---------- 1 file changed, 77 insertions(+), 94 deletions(-) diff --git a/src/docs/asciidoc/web/webflux-webclient.adoc b/src/docs/asciidoc/web/webflux-webclient.adoc index 3da96340a4b..376f99d885b 100644 --- a/src/docs/asciidoc/web/webflux-webclient.adoc +++ b/src/docs/asciidoc/web/webflux-webclient.adoc @@ -1,16 +1,20 @@ [[webflux-client]] = WebClient -Spring WebFlux includes a reactive, non-blocking `WebClient` for HTTP requests. The client -has a functional, fluent API with reactive types for declarative composition, see -<>. WebFlux client and server rely on the -same non-blocking <> to encode and decode request -and response content. +Spring WebFlux includes a client to perform HTTP requests with. `WebClient` has a +functional, fluent API based on Reactor, see <>, +which enables declarative composition of asynchronous logic without the need to deal with +threads or concurrency. It is fully non-blocking, it supports streaming, and relies on +the same <> that are also used to encode and +decode request and response content on the server side. -Internally `WebClient` delegates to an HTTP client library. By default, it uses -https://github.com/reactor/reactor-netty[Reactor Netty], there is built-in support for -the Jetty https://github.com/jetty-project/jetty-reactive-httpclient[reactive HttpClient], -and others can be plugged in through a `ClientHttpConnector`. +`WebClient` needs an HTTP client library to perform requests with. There is built-in +support for the following: + +* https://github.com/reactor/reactor-netty[Reactor Netty] +* https://github.com/jetty-project/jetty-reactive-httpclient[Jetty Reactive HttpClient] +* https://hc.apache.org/index.html[Apache HttpComponents] +* Others can be plugged via `ClientHttpConnector`. @@ -23,12 +27,10 @@ The simplest way to create a `WebClient` is through one of the static factory me * `WebClient.create()` * `WebClient.create(String baseUrl)` -The above methods use the Reactor Netty `HttpClient` with default settings and expect -`io.projectreactor.netty:reactor-netty` to be on the classpath. - You can also use `WebClient.builder()` with further options: * `uriBuilderFactory`: Customized `UriBuilderFactory` to use as a base URL. +* `defaultUriVariables`: default values to use when expanding URI templates. * `defaultHeader`: Headers for every request. * `defaultCookie`: Cookies for every request. * `defaultRequest`: `Consumer` to customize every request. @@ -36,33 +38,25 @@ You can also use `WebClient.builder()` with further options: * `exchangeStrategies`: HTTP message reader/writer customizations. * `clientConnector`: HTTP client library settings. -The following example configures <>: +For example: [source,java,indent=0,subs="verbatim,quotes",role="primary"] .Java ---- WebClient client = WebClient.builder() - .exchangeStrategies(builder -> { - return builder.codecs(codecConfigurer -> { - //... - }); - }) + .codecs(configurer -> ... ) .build(); ---- [source,kotlin,indent=0,subs="verbatim,quotes",role="secondary"] .Kotlin ---- val webClient = WebClient.builder() - .exchangeStrategies { strategies -> - strategies.codecs { - //... - } - } + .codecs { configurer -> ... } .build() ---- -Once built, a `WebClient` instance is immutable. However, you can clone it and build a -modified copy without affecting the original instance, as the following example shows: +Once built, a `WebClient` is immutable. However, you can clone it and build a +modified copy as follows: [source,java,indent=0,subs="verbatim,quotes",role="primary"] .Java @@ -94,37 +88,29 @@ modified copy without affecting the original instance, as the following example [[webflux-client-builder-maxinmemorysize]] === MaxInMemorySize -Spring WebFlux configures <> for buffering -data in-memory in codec to avoid application memory issues. By the default this is -configured to 256KB and if that's not enough for your use case, you'll see the following: +Codecs have <> for buffering data in +memory to avoid application memory issues. By the default those are set to 256KB. +If that's not enough you'll get the following error: ---- org.springframework.core.io.buffer.DataBufferLimitException: Exceeded limit on max bytes to buffer ---- -You can configure this limit on all default codecs with the following code sample: +To change the limit for default codecs, use the following: [source,java,indent=0,subs="verbatim,quotes",role="primary"] .Java ---- WebClient webClient = WebClient.builder() - .exchangeStrategies(builder -> - builder.codecs(codecs -> - codecs.defaultCodecs().maxInMemorySize(2 * 1024 * 1024) - ) - ) + .codecs(configurer -> configurer.defaultCodecs().maxInMemorySize(2 * 1024 * 1024)); .build(); ---- [source,kotlin,indent=0,subs="verbatim,quotes",role="secondary"] .Kotlin ---- val webClient = WebClient.builder() - .exchangeStrategies { builder -> - builder.codecs { - it.defaultCodecs().maxInMemorySize(2 * 1024 * 1024) - } - } - .build() + .codecs { configurer -> configurer.defaultCodecs().maxInMemorySize(2 * 1024 * 1024) } + .build() ---- @@ -132,7 +118,7 @@ You can configure this limit on all default codecs with the following code sampl [[webflux-client-builder-reactor]] === Reactor Netty -To customize Reactor Netty settings, simple provide a pre-configured `HttpClient`: +To customize Reactor Netty settings, provide a pre-configured `HttpClient`: [source,java,indent=0,subs="verbatim,quotes",role="primary"] .Java @@ -457,8 +443,30 @@ The following example shows how to customize Apache HttpComponents `HttpClient` [[webflux-client-retrieve]] == `retrieve()` -The `retrieve()` method is the easiest way to get a response body and decode it. -The following example shows how to do so: +The `retrieve()` method can be used to declare how to extract the response. For example: + +[source,java,indent=0,subs="verbatim,quotes",role="primary"] +.Java +---- + WebClient client = WebClient.create("https://example.org"); + + Mono> result = client.get() + .uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON) + .retrieve() + .toEntity(Person.class); +---- +[source,kotlin,indent=0,subs="verbatim,quotes",role="secondary"] +.Kotlin +---- + val client = WebClient.create("https://example.org") + + val result = client.get() + .uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON) + .retrieve() + .toEntity().awaitSingle() +---- + +Or to get only the body: [source,java,indent=0,subs="verbatim,quotes",role="primary"] .Java @@ -481,7 +489,7 @@ The following example shows how to do so: .awaitBody() ---- -You can also get a stream of objects decoded from the response, as the following example shows: +To get a stream of decoded objects: [source,java,indent=0,subs="verbatim,quotes",role="primary"] .Java @@ -500,11 +508,9 @@ You can also get a stream of objects decoded from the response, as the following .bodyToFlow() ---- -By default, responses with 4xx or 5xx status codes result in an -`WebClientResponseException` or one of its HTTP status specific sub-classes, such as -`WebClientResponseException.BadRequest`, `WebClientResponseException.NotFound`, and others. -You can also use the `onStatus` method to customize the resulting exception, -as the following example shows: +By default, 4xx or 5xx responses result in an `WebClientResponseException`, including +sub-classes for specific HTTP status codes. To customize the handling of error +responses, use `onStatus` handlers as follows: [source,java,indent=0,subs="verbatim,quotes",role="primary"] .Java @@ -527,67 +533,44 @@ as the following example shows: .awaitBody() ---- -When `onStatus` is used, if the response is expected to have content, then the `onStatus` -callback should consume it. If not, the content will be automatically drained to ensure -resources are released. - [[webflux-client-exchange]] -== `exchange()` +== `exchangeToMono()` -The `exchange()` method provides more control than the `retrieve` method. The following example is equivalent -to `retrieve()` but also provides access to the `ClientResponse`: +The `exchangeToMono()` and `exchangeToFlux()` methods are useful for more advanced +cases that require more control, such as to decode the response differently +depending on the response status: [source,java,indent=0,subs="verbatim,quotes",role="primary"] .Java ---- - Mono result = client.get() - .uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON) - .exchange() - .flatMap(response -> response.bodyToMono(Person.class)); + Mono 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()); + } + }); ---- [source,kotlin,indent=0,subs="verbatim,quotes",role="secondary"] .Kotlin ---- - val result = client.get() - .uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON) - .awaitExchange() - .awaitBody() ---- -At this level, you can also create a full `ResponseEntity`: +When using the above, after the returned `Mono` or `Flux` completes, the response body +is checked and if not consumed it is released to prevent memory and connection leaks. +Therefore the response cannot be decoded further downstream. It is up to the provided +function to declare how to decode the response if needed. -[source,java,indent=0,subs="verbatim,quotes",role="primary"] -.Java ----- - Mono> result = client.get() - .uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON) - .exchange() - .flatMap(response -> response.toEntity(Person.class)); ----- -[source,kotlin,indent=0,subs="verbatim,quotes",role="secondary"] -.Kotlin ----- - val result = client.get() - .uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON) - .awaitExchange() - .toEntity() ----- - -Note that (unlike `retrieve()`), with `exchange()`, there are no automatic error signals for -4xx and 5xx responses. You have to check the status code and decide how to proceed. - -[CAUTION] -==== -Unlike `retrieve()`, when using `exchange()`, it is the responsibility of the application -to consume any response content regardless of the scenario (success, error, unexpected -data, etc). Not doing so can cause a memory leak. The Javadoc for `ClientResponse` lists -all the available options for consuming the body. Generally prefer using `retrieve()` -unless you have a good reason for using `exchange()` which does allow to check the -response status and headers before deciding how to or if to consume the response. -====