diff --git a/spring-web/src/main/java/org/springframework/web/client/reactive/ClientResponse.java b/spring-web/src/main/java/org/springframework/web/client/reactive/ClientResponse.java index 4d7994a979..af1f26a347 100644 --- a/spring-web/src/main/java/org/springframework/web/client/reactive/ClientResponse.java +++ b/spring-web/src/main/java/org/springframework/web/client/reactive/ClientResponse.java @@ -32,7 +32,8 @@ import org.springframework.http.codec.BodyExtractor; /** * Represents an HTTP response, as returned by the {@link WebClient}. * Access to headers and body is offered by {@link Headers} and - * {@link #body(BodyExtractor)} respectively. + * {@link #body(BodyExtractor)}, {@link #bodyToMono(Class)}, {@link #bodyToFlux(Class)} + * respectively. * * @author Brian Clozel * @author Arjen Poutsma @@ -51,7 +52,9 @@ public interface ClientResponse { Headers headers(); /** - * Extract the body with the given {@code BodyExtractor}. + * Extract the body with the given {@code BodyExtractor}. Unlike {@link #bodyToMono(Class)} and + * {@link #bodyToFlux(Class)}; this method does not check for a 4xx or 5xx status code before + * extracting the body. * @param extractor the {@code BodyExtractor} that reads from the response * @param the type of the body returned * @return the extracted body @@ -59,18 +62,22 @@ public interface ClientResponse { T body(BodyExtractor extractor); /** - * Extract the body to a {@code Mono}. + * Extract the body to a {@code Mono}. If the response has status code 4xx or 5xx, the + * {@code Mono} will contain a {@link WebClientException}. * @param elementClass the class of element in the {@code Mono} * @param the element type - * @return the body as a mono + * @return a mono containing the body, or a {@link WebClientException} if the status code is + * 4xx or 5xx */ Mono bodyToMono(Class elementClass); /** - * Extract the body to a {@code Flux}. + * Extract the body to a {@code Flux}. If the response has status code 4xx or 5xx, the + * {@code Flux} will contain a {@link WebClientException}. * @param elementClass the class of element in the {@code Flux} * @param the element type - * @return the body as a flux + * @return a flux containing the body, or a {@link WebClientException} if the status code is + * 4xx or 5xx */ Flux bodyToFlux(Class elementClass); diff --git a/spring-web/src/main/java/org/springframework/web/client/reactive/DefaultClientResponse.java b/spring-web/src/main/java/org/springframework/web/client/reactive/DefaultClientResponse.java index 5f48dd4b58..741758e4ae 100644 --- a/spring-web/src/main/java/org/springframework/web/client/reactive/DefaultClientResponse.java +++ b/spring-web/src/main/java/org/springframework/web/client/reactive/DefaultClientResponse.java @@ -20,9 +20,11 @@ import java.util.Collections; import java.util.List; import java.util.Optional; import java.util.OptionalLong; +import java.util.function.Function; import java.util.function.Supplier; import java.util.stream.Stream; +import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @@ -77,12 +79,28 @@ class DefaultClientResponse implements ClientResponse { @Override public Mono bodyToMono(Class elementClass) { - return body(BodyExtractors.toMono(elementClass)); + return bodyToPublisher(BodyExtractors.toMono(elementClass), Mono::error); } @Override public Flux bodyToFlux(Class elementClass) { - return body(BodyExtractors.toFlux(elementClass)); + return bodyToPublisher(BodyExtractors.toFlux(elementClass), Flux::error); + } + + private > T bodyToPublisher( + BodyExtractor extractor, + Function errorFunction) { + + HttpStatus status = statusCode(); + if (status.is4xxClientError() || status.is5xxServerError()) { + WebClientException ex = new WebClientException( + "ClientResponse has erroneous status code: " + status.value() + + " " + status.getReasonPhrase()); + return errorFunction.apply(ex); + } + else { + return body(extractor); + } } private class DefaultHeaders implements Headers { diff --git a/spring-web/src/main/java/org/springframework/web/client/reactive/DefaultWebClientBuilder.java b/spring-web/src/main/java/org/springframework/web/client/reactive/DefaultWebClientBuilder.java index d9bf642757..26876ce000 100644 --- a/spring-web/src/main/java/org/springframework/web/client/reactive/DefaultWebClientBuilder.java +++ b/spring-web/src/main/java/org/springframework/web/client/reactive/DefaultWebClientBuilder.java @@ -18,13 +18,9 @@ package org.springframework.web.client.reactive; import java.util.logging.Level; -import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import org.springframework.http.client.reactive.ClientHttpConnector; -import org.springframework.http.client.reactive.ClientHttpResponse; -import org.springframework.http.codec.BodyExtractor; -import org.springframework.http.codec.BodyExtractors; import org.springframework.util.Assert; /** @@ -82,33 +78,6 @@ class DefaultWebClientBuilder implements WebClient.Builder { this.filter = filter; } - @Override - public Mono retrieveMono(ClientRequest request, Class elementClass) { - Assert.notNull(request, "'request' must not be null"); - Assert.notNull(elementClass, "'elementClass' must not be null"); - - return retrieve(request, BodyExtractors.toMono(elementClass)) - .then(m -> m); - } - - @Override - public Flux retrieveFlux(ClientRequest request, Class elementClass) { - Assert.notNull(request, "'request' must not be null"); - Assert.notNull(elementClass, "'elementClass' must not be null"); - - return retrieve(request, BodyExtractors.toFlux(elementClass)) - .flatMap(flux -> flux); - } - - private Mono retrieve(ClientRequest request, - BodyExtractor extractor) { - - ExchangeFilterFunction errorFilter = ExchangeFilterFunctions.clientOrServerError(); - - return errorFilter.filter(request, this::exchange) - .map(clientResponse -> clientResponse.body(extractor)); - } - @Override public Mono exchange(ClientRequest request) { Assert.notNull(request, "'request' must not be null"); diff --git a/spring-web/src/main/java/org/springframework/web/client/reactive/ExchangeFilterFunctions.java b/spring-web/src/main/java/org/springframework/web/client/reactive/ExchangeFilterFunctions.java index 43a612a192..818ab1e072 100644 --- a/spring-web/src/main/java/org/springframework/web/client/reactive/ExchangeFilterFunctions.java +++ b/spring-web/src/main/java/org/springframework/web/client/reactive/ExchangeFilterFunctions.java @@ -18,14 +18,10 @@ package org.springframework.web.client.reactive; import java.nio.charset.StandardCharsets; import java.util.Base64; -import java.util.Optional; -import java.util.function.Function; -import java.util.function.Predicate; import reactor.core.publisher.Mono; import org.springframework.http.HttpHeaders; -import org.springframework.http.HttpStatus; import org.springframework.util.Assert; /** @@ -38,95 +34,6 @@ import org.springframework.util.Assert; */ public abstract class ExchangeFilterFunctions { - /** - * Return a filter that will publish a {@link WebClientException} when the - * {@code ClientResponse} has a 4xx status code. - * @return the {@code ExchangeFilterFunction} that publishes a {@code WebClientException} when - * the response has a client error - */ - public static ExchangeFilterFunction clientError() { - return statusError(HttpStatus::is4xxClientError); - } - - /** - * Return a filter that will publish a {@link WebClientException} if the - * {@code ClientResponse} has a 5xx status code. - * @return the {@code ExchangeFilterFunction} that publishes a {@code WebClientException} when - * the response has a server error - */ - public static ExchangeFilterFunction serverError() { - return statusError(HttpStatus::is5xxServerError); - } - - /** - * Return a filter that will publish a {@link WebClientException} if the - * {@code ClientResponse} has a 4xx or 5xx status code. - * @return the {@code ExchangeFilterFunction} that publishes a {@code WebClientException} when - * the response has a client or server error - */ - public static ExchangeFilterFunction clientOrServerError() { - return clientError().andThen(serverError()); - } - - private static ExchangeFilterFunction statusError(Predicate predicate) { - Function> mapper = - clientResponse -> { - HttpStatus status = clientResponse.statusCode(); - if (predicate.test(status)) { - return Optional.of(new WebClientException( - "ClientResponse has invalid status code: " + status.value() + - " " + status.getReasonPhrase())); - } - else { - return Optional.empty(); - } - }; - - return errorMapper(mapper); - } - - /** - * Return a filter that will publish a {@link WebClientException} if the response satisfies - * the given {@code predicate} function. - * @param predicate the predicate to test the response with - * @return the {@code ExchangeFilterFunction} that publishes a {@code WebClientException} when - * {@code predicate} returns {@code true} - */ - public static ExchangeFilterFunction errorPredicate(Predicate predicate) { - Assert.notNull(predicate, "'predicate' must not be null"); - - Function> mapper = - clientResponse -> { - if (predicate.test(clientResponse)) { - return Optional.of(new WebClientException( - "ClientResponse does not satisfy predicate : " + predicate)); - } - else { - return Optional.empty(); - } - }; - - return errorMapper(mapper); - } - - /** - * Return a filter that maps the response to a potential error. Exceptions returned by - * {@code mapper} will be published as signal in the {@code Mono} return value. - * @param mapper the function that maps from response to optional error - * @return the {@code ExchangeFilterFunction} that propagates the errors provided by - * {@code mapper} - */ - public static ExchangeFilterFunction errorMapper(Function> mapper) { - - Assert.notNull(mapper, "'mapper' must not be null"); - return ExchangeFilterFunction.ofResponseProcessor( - clientResponse -> { - Optional error = mapper.apply(clientResponse); - return error.isPresent() ? Mono.error(error.get()) : Mono.just(clientResponse); - }); - } - /** * Return a filter that adds an Authorization header for HTTP Basic Authentication. * @param username the username to use diff --git a/spring-web/src/main/java/org/springframework/web/client/reactive/WebClient.java b/spring-web/src/main/java/org/springframework/web/client/reactive/WebClient.java index b77f73803e..afdef85611 100644 --- a/spring-web/src/main/java/org/springframework/web/client/reactive/WebClient.java +++ b/spring-web/src/main/java/org/springframework/web/client/reactive/WebClient.java @@ -16,7 +16,6 @@ package org.springframework.web.client.reactive; -import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import org.springframework.http.client.reactive.ClientHttpConnector; @@ -24,9 +23,7 @@ import org.springframework.util.Assert; /** * Reactive Web client supporting the HTTP/1.1 protocol. Main entry point is through the - * {@link #exchange(ClientRequest)} method, or through the - * {@link #retrieveMono(ClientRequest, Class)} and {@link #retrieveFlux(ClientRequest, Class)} - * convenience methods. + * {@link #exchange(ClientRequest)} method. * *

For example: * 

@@ -35,54 +32,24 @@ import org.springframework.util.Assert;
  *
  * Mono<String> result = client
  *   .exchange(request)
- *   .then(response -> response.body(BodyExtractors.toMono(String.class)));
- *
- *

or, by using {@link #retrieveMono(ClientRequest, Class)}: - * 

- * Mono<String> result = client.retrieveMono(request, String.class);
+ *   .then(response -> response.bodyToMono(String.class));
  *
* * @author Brian Clozel * @author Arjen Poutsma * @since 5.0 */ -public interface WebClient { +public interface WebClient extends ExchangeFunction { /** * Exchange the given request for a response mono. Invoking this method performs the actual * HTTP request/response exchange. - *

Note that this method will not publish an exception if the response - * has a 4xx or 5xx status code; as opposed to {@link #retrieveMono(ClientRequest, Class)} and - * {@link #retrieveFlux(ClientRequest, Class)}. * @param request the request to exchange * @return the response, wrapped in a {@code Mono} */ + @Override Mono exchange(ClientRequest request); - /** - * Retrieve the body of the response as a {@code Mono}. A 4xx or 5xx status - * code in the response will result in a {@link WebClientException} published in the returned - * {@code Mono}. - * @param request the request to exchange - * @param elementClass the class of element in the {@code Mono} - * @param the element type - * @return the response body as a mono - * @see ExchangeFilterFunctions#clientOrServerError() - */ - Mono retrieveMono(ClientRequest request, Class elementClass); - - /** - * Retrieve the body of the response as a {@code Flux}. A 4xx or 5xx status - * code in the response will result in a {@link WebClientException} published in the returned - * {@code Flux}. - * @param request the request to exchange - * @param elementClass the class of element in the {@code Flux} - * @param the element type - * @return the response body as a flux - * @see ExchangeFilterFunctions#clientOrServerError() - */ - Flux retrieveFlux(ClientRequest request, Class elementClass); - /** * Create a new instance of {@code WebClient} with the given connector. This method uses diff --git a/spring-web/src/test/java/org/springframework/web/client/reactive/DefaultClientResponseTests.java b/spring-web/src/test/java/org/springframework/web/client/reactive/DefaultClientResponseTests.java index d58c6c71d1..0f2e594308 100644 --- a/spring-web/src/test/java/org/springframework/web/client/reactive/DefaultClientResponseTests.java +++ b/spring-web/src/test/java/org/springframework/web/client/reactive/DefaultClientResponseTests.java @@ -29,6 +29,7 @@ import org.junit.Before; import org.junit.Test; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; import org.springframework.core.codec.StringDecoder; import org.springframework.core.io.buffer.DataBuffer; @@ -125,6 +126,7 @@ public class DefaultClientResponseTests { HttpHeaders httpHeaders = new HttpHeaders(); httpHeaders.setContentType(MediaType.TEXT_PLAIN); when(mockResponse.getHeaders()).thenReturn(httpHeaders); + when(mockResponse.getStatusCode()).thenReturn(HttpStatus.OK); when(mockResponse.getBody()).thenReturn(body); Set> messageReaders = Collections @@ -135,6 +137,24 @@ public class DefaultClientResponseTests { assertEquals("foo", resultMono.block()); } + @Test + public void bodyToMonoError() throws Exception { + HttpHeaders httpHeaders = new HttpHeaders(); + httpHeaders.setContentType(MediaType.TEXT_PLAIN); + when(mockResponse.getHeaders()).thenReturn(httpHeaders); + when(mockResponse.getStatusCode()).thenReturn(HttpStatus.NOT_FOUND); + + Set> messageReaders = Collections + .singleton(new DecoderHttpMessageReader(new StringDecoder())); + when(mockWebClientStrategies.messageReaders()).thenReturn(messageReaders::stream); + + Mono resultMono = defaultClientResponse.bodyToMono(String.class); + + StepVerifier.create(resultMono) + .expectError(WebClientException.class) + .verify(); + } + @Test public void bodyToFlux() throws Exception { DefaultDataBufferFactory factory = new DefaultDataBufferFactory(); @@ -145,6 +165,7 @@ public class DefaultClientResponseTests { HttpHeaders httpHeaders = new HttpHeaders(); httpHeaders.setContentType(MediaType.TEXT_PLAIN); when(mockResponse.getHeaders()).thenReturn(httpHeaders); + when(mockResponse.getStatusCode()).thenReturn(HttpStatus.OK); when(mockResponse.getBody()).thenReturn(body); Set> messageReaders = Collections @@ -156,5 +177,22 @@ public class DefaultClientResponseTests { assertEquals(Collections.singletonList("foo"), result.block()); } + @Test + public void bodyToFluxError() throws Exception { + HttpHeaders httpHeaders = new HttpHeaders(); + httpHeaders.setContentType(MediaType.TEXT_PLAIN); + when(mockResponse.getHeaders()).thenReturn(httpHeaders); + when(mockResponse.getStatusCode()).thenReturn(HttpStatus.INTERNAL_SERVER_ERROR); + + Set> messageReaders = Collections + .singleton(new DecoderHttpMessageReader(new StringDecoder())); + when(mockWebClientStrategies.messageReaders()).thenReturn(messageReaders::stream); + + Flux resultFlux = defaultClientResponse.bodyToFlux(String.class); + StepVerifier.create(resultFlux) + .expectError(WebClientException.class) + .verify(); + } + } \ No newline at end of file diff --git a/spring-web/src/test/java/org/springframework/web/client/reactive/ExchangeFilterFunctionsTests.java b/spring-web/src/test/java/org/springframework/web/client/reactive/ExchangeFilterFunctionsTests.java index 946545460d..9666ca04ee 100644 --- a/spring-web/src/test/java/org/springframework/web/client/reactive/ExchangeFilterFunctionsTests.java +++ b/spring-web/src/test/java/org/springframework/web/client/reactive/ExchangeFilterFunctionsTests.java @@ -16,20 +16,15 @@ package org.springframework.web.client.reactive; -import java.util.Optional; - import org.junit.Test; import reactor.core.publisher.Mono; -import reactor.test.StepVerifier; import org.springframework.http.HttpHeaders; -import org.springframework.http.HttpStatus; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; /** * @author Arjen Poutsma @@ -85,75 +80,6 @@ public class ExchangeFilterFunctionsTests { assertTrue(filterInvoked[0]); } - - @Test - public void clientNoError() throws Exception { - ClientRequest request = ClientRequest.GET("http://example.com").build(); - ClientResponse response = mock(ClientResponse.class); - when(response.statusCode()).thenReturn(HttpStatus.OK); - ExchangeFunction exchange = r -> Mono.just(response); - - ExchangeFilterFunction standardErrors = ExchangeFilterFunctions.clientError(); - - Mono result = standardErrors.filter(request, exchange); - - StepVerifier.create(result) - .expectNext(response) - .expectComplete() - .verify(); - } - - @Test - public void serverError() throws Exception { - ClientRequest request = ClientRequest.GET("http://example.com").build(); - ClientResponse response = mock(ClientResponse.class); - when(response.statusCode()).thenReturn(HttpStatus.INTERNAL_SERVER_ERROR); - ExchangeFunction exchange = r -> Mono.just(response); - - ExchangeFilterFunction standardErrors = ExchangeFilterFunctions.serverError(); - - Mono result = standardErrors.filter(request, exchange); - - StepVerifier.create(result) - .expectError(WebClientException.class) - .verify(); - } - - @Test - public void errorPredicate() throws Exception { - ClientRequest request = ClientRequest.GET("http://example.com").build(); - ClientResponse response = mock(ClientResponse.class); - when(response.statusCode()).thenReturn(HttpStatus.NOT_FOUND); - ExchangeFunction exchange = r -> Mono.just(response); - - ExchangeFilterFunction errorPredicate = ExchangeFilterFunctions - .errorPredicate(clientResponse -> clientResponse.statusCode().is4xxClientError()); - - Mono result = errorPredicate.filter(request, exchange); - - StepVerifier.create(result) - .expectError(WebClientException.class) - .verify(); - } - - - @Test - public void errorMapperFunction() throws Exception { - ClientRequest request = ClientRequest.GET("http://example.com").build(); - ClientResponse response = mock(ClientResponse.class); - ExchangeFunction exchange = r -> Mono.just(response); - - ExchangeFilterFunction errorMapper = ExchangeFilterFunctions - .errorMapper(clientResponse -> Optional.of(new IllegalStateException())); - - Mono result = errorMapper.filter(request, exchange); - - StepVerifier.create(result) - .expectError(IllegalStateException.class) - .verify(); - } - - @Test public void basicAuthentication() throws Exception { ClientRequest request = ClientRequest.GET("http://example.com").build(); diff --git a/spring-web/src/test/java/org/springframework/web/client/reactive/WebClientIntegrationTests.java b/spring-web/src/test/java/org/springframework/web/client/reactive/WebClientIntegrationTests.java index 137809c56d..1597c628c0 100644 --- a/spring-web/src/test/java/org/springframework/web/client/reactive/WebClientIntegrationTests.java +++ b/spring-web/src/test/java/org/springframework/web/client/reactive/WebClientIntegrationTests.java @@ -110,48 +110,6 @@ public class WebClientIntegrationTests { assertEquals("/greeting?name=Spring", recordedRequest.getPath()); } - @Test - public void retrieveMono() throws Exception { - HttpUrl baseUrl = server.url("/greeting?name=Spring"); - this.server.enqueue(new MockResponse().setBody("Hello Spring!")); - - ClientRequest request = ClientRequest.GET(baseUrl.toString()).build(); - - Mono result = this.webClient - .retrieveMono(request, String.class); - - StepVerifier.create(result) - .expectNext("Hello Spring!") - .expectComplete() - .verify(); - - RecordedRequest recordedRequest = server.takeRequest(); - assertEquals(1, server.getRequestCount()); - assertEquals("*/*", recordedRequest.getHeader(HttpHeaders.ACCEPT)); - assertEquals("/greeting?name=Spring", recordedRequest.getPath()); - } - - @Test - public void retrieveFlux() throws Exception { - HttpUrl baseUrl = server.url("/greeting?name=Spring"); - this.server.enqueue(new MockResponse().setBody("Hello Spring!")); - - ClientRequest request = ClientRequest.GET(baseUrl.toString()).build(); - - Flux result = this.webClient - .retrieveFlux(request, String.class); - - StepVerifier.create(result) - .expectNext("Hello Spring!") - .expectComplete() - .verify(); - - RecordedRequest recordedRequest = server.takeRequest(); - assertEquals(1, server.getRequestCount()); - assertEquals("*/*", recordedRequest.getHeader(HttpHeaders.ACCEPT)); - assertEquals("/greeting?name=Spring", recordedRequest.getPath()); - } - @Test public void jsonString() throws Exception { HttpUrl baseUrl = server.url("/json"); @@ -309,48 +267,6 @@ public class WebClientIntegrationTests { assertEquals("/greeting?name=Spring", recordedRequest.getPath()); } - @Test - public void retrieveNotFound() throws Exception { - HttpUrl baseUrl = server.url("/greeting?name=Spring"); - this.server.enqueue(new MockResponse().setResponseCode(404) - .setHeader("Content-Type", "text/plain").setBody("Not Found")); - - ClientRequest request = ClientRequest.GET(baseUrl.toString()).build(); - - Mono result = this.webClient - .retrieveMono(request, String.class); - - StepVerifier.create(result) - .expectError(WebClientException.class) - .verify(Duration.ofSeconds(3)); - - RecordedRequest recordedRequest = server.takeRequest(); - assertEquals(1, server.getRequestCount()); - assertEquals("*/*", recordedRequest.getHeader(HttpHeaders.ACCEPT)); - assertEquals("/greeting?name=Spring", recordedRequest.getPath()); - } - - @Test - public void retrieveServerError() throws Exception { - HttpUrl baseUrl = server.url("/greeting?name=Spring"); - this.server.enqueue(new MockResponse().setResponseCode(500) - .setHeader("Content-Type", "text/plain").setBody("Not Found")); - - ClientRequest request = ClientRequest.GET(baseUrl.toString()).build(); - - Mono result = this.webClient - .retrieveMono(request, String.class); - - StepVerifier.create(result) - .expectError(WebClientException.class) - .verify(Duration.ofSeconds(3)); - - RecordedRequest recordedRequest = server.takeRequest(); - assertEquals(1, server.getRequestCount()); - assertEquals("*/*", recordedRequest.getHeader(HttpHeaders.ACCEPT)); - assertEquals("/greeting?name=Spring", recordedRequest.getPath()); - } - @Test public void filter() throws Exception { HttpUrl baseUrl = server.url("/greeting?name=Spring");