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

Add status code check to bodyTo[Mono|Flux]

Browse Source 
- Add 4xx/5xx status code check to ClientResponse.bodyToMono and bodyToFlux.
- Removed WebClient.retrieveMono and retrieveFlux.

Issue: SPR-14977
This commit is contained in:
Arjen Poutsma
2016-12-06 09:24:29 +01:00
parent 953f7d40cf
commit 079eca9f63
8 changed files with 75 additions and 327 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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 <T> the type of the body returned
* @return the extracted body
@@ -59,18 +62,22 @@ public interface ClientResponse {
<T> T body(BodyExtractor<T, ? super ClientHttpResponse> 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 <T> 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
*/
<T> Mono<T> bodyToMono(Class<? extends T> 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 <T> 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
*/
<T> Flux<T> bodyToFlux(Class<? extends T> elementClass);

22
spring-web/src/main/java/org/springframework/web/client/reactive/DefaultClientResponse.java
View File

@@ -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 <T> Mono<T> bodyToMono(Class<? extends T> elementClass) {
return body(BodyExtractors.toMono(elementClass));
return bodyToPublisher(BodyExtractors.toMono(elementClass), Mono::error);
}
@Override
public <T> Flux<T> bodyToFlux(Class<? extends T> elementClass) {
return body(BodyExtractors.toFlux(elementClass));
return bodyToPublisher(BodyExtractors.toFlux(elementClass), Flux::error);
}
private <T extends Publisher<?>> T bodyToPublisher(
BodyExtractor<T, ? super ClientHttpResponse> extractor,
Function<WebClientException, T> 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 {

31
spring-web/src/main/java/org/springframework/web/client/reactive/DefaultWebClientBuilder.java
View File

@@ -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 <T> Mono<T> retrieveMono(ClientRequest<?> request, Class<? extends T> 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 <T> Flux<T> retrieveFlux(ClientRequest<?> request, Class<? extends T> 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 <T> Mono<T> retrieve(ClientRequest<?> request,
BodyExtractor<T, ? super ClientHttpResponse> extractor) {
ExchangeFilterFunction errorFilter = ExchangeFilterFunctions.clientOrServerError();
return errorFilter.filter(request, this::exchange)
.map(clientResponse -> clientResponse.body(extractor));
}
@Override
public Mono<ClientResponse> exchange(ClientRequest<?> request) {
Assert.notNull(request, "'request' must not be null");

93
spring-web/src/main/java/org/springframework/web/client/reactive/ExchangeFilterFunctions.java
View File

@@ -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<HttpStatus> predicate) {
Function<ClientResponse, Optional<? extends Throwable>> 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<ClientResponse> predicate) {
Assert.notNull(predicate, "'predicate' must not be null");
Function<ClientResponse, Optional<? extends Throwable>> 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<ClientResponse>} 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<ClientResponse,
Optional<? extends Throwable>> mapper) {
Assert.notNull(mapper, "'mapper' must not be null");
return ExchangeFilterFunction.ofResponseProcessor(
clientResponse -> {
Optional<? extends Throwable> 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

41
spring-web/src/main/java/org/springframework/web/client/reactive/WebClient.java
View File

@@ -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.
*
* <p>For example:
* <pre class="code">
@@ -35,54 +32,24 @@ import org.springframework.util.Assert;
*
* Mono&lt;String&gt; result = client
* .exchange(request)
* .then(response -> response.body(BodyExtractors.toMono(String.class)));
* </pre>
* <p>or, by using {@link #retrieveMono(ClientRequest, Class)}:
* <pre class="code">
* Mono&lt;String&gt; result = client.retrieveMono(request, String.class);
* .then(response -> response.bodyToMono(String.class));
* </pre>
*
* @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.
* <p>Note that this method will <strong>not</strong> 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<ClientResponse> 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 <T> the element type
* @return the response body as a mono
* @see ExchangeFilterFunctions#clientOrServerError()
*/
<T> Mono<T> retrieveMono(ClientRequest<?> request, Class<? extends T> 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 <T> the element type
* @return the response body as a flux
* @see ExchangeFilterFunctions#clientOrServerError()
*/
<T> Flux<T> retrieveFlux(ClientRequest<?> request, Class<? extends T> elementClass);
/**
* Create a new instance of {@code WebClient} with the given connector. This method uses

38
spring-web/src/test/java/org/springframework/web/client/reactive/DefaultClientResponseTests.java
View File

@@ -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<HttpMessageReader<?>> 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<HttpMessageReader<?>> messageReaders = Collections
.singleton(new DecoderHttpMessageReader<String>(new StringDecoder()));
when(mockWebClientStrategies.messageReaders()).thenReturn(messageReaders::stream);
Mono<String> 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<HttpMessageReader<?>> 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<HttpMessageReader<?>> messageReaders = Collections
.singleton(new DecoderHttpMessageReader<String>(new StringDecoder()));
when(mockWebClientStrategies.messageReaders()).thenReturn(messageReaders::stream);
Flux<String> resultFlux = defaultClientResponse.bodyToFlux(String.class);
StepVerifier.create(resultFlux)
.expectError(WebClientException.class)
.verify();
}
}

74
spring-web/src/test/java/org/springframework/web/client/reactive/ExchangeFilterFunctionsTests.java
View File

@@ -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<Void> 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<ClientResponse> result = standardErrors.filter(request, exchange);
StepVerifier.create(result)
.expectNext(response)
.expectComplete()
.verify();
}
@Test
public void serverError() throws Exception {
ClientRequest<Void> 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<ClientResponse> result = standardErrors.filter(request, exchange);
StepVerifier.create(result)
.expectError(WebClientException.class)
.verify();
}
@Test
public void errorPredicate() throws Exception {
ClientRequest<Void> 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<ClientResponse> result = errorPredicate.filter(request, exchange);
StepVerifier.create(result)
.expectError(WebClientException.class)
.verify();
}
@Test
public void errorMapperFunction() throws Exception {
ClientRequest<Void> 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<ClientResponse> result = errorMapper.filter(request, exchange);
StepVerifier.create(result)
.expectError(IllegalStateException.class)
.verify();
}
@Test
public void basicAuthentication() throws Exception {
ClientRequest<Void> request = ClientRequest.GET("http://example.com").build();

84
spring-web/src/test/java/org/springframework/web/client/reactive/WebClientIntegrationTests.java
View File

@@ -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<Void> request = ClientRequest.GET(baseUrl.toString()).build();
Mono<String> 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<Void> request = ClientRequest.GET(baseUrl.toString()).build();
Flux<String> 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<Void> request = ClientRequest.GET(baseUrl.toString()).build();
Mono<String> 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<Void> request = ClientRequest.GET(baseUrl.toString()).build();
Mono<String> 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");