WebClientException should allow access to status code of the response

This commit changes the WebClient so that it now throws a
`WebClientResponseException` for `ResponseSpec.bodyTo`. This newly
introduces exception contains the status code, headers, and body of the
response message.

As a consequence of the above, we had to change `onStatus` so that the
`exceptionFunction` now returns a `Mono<Throwable>` rather than a
`Throwable`, which it was before. The Mono allows for asynchronous
operations, such as reading the contents of the body.

Issue: SPR-15824

spring-web/src/main/java/org/springframework/http/HttpHeaders.java

@@ -73,6 +73,10 @@ public class HttpHeaders implements MultiValueMap<String, String>, Serializable
 
 	private static final long serialVersionUID = -8578554704772377436L;
 
+	/**
+	 * The empty {@code HttpHeaders} instance (immutable).
+	 */
+	public static final HttpHeaders EMPTY = new HttpHeaders(new LinkedHashMap<>(0), true);
 	/**
 	 * The HTTP {@code Accept} header field name.
 	 * @see <a href="http://tools.ietf.org/html/rfc7231#section-5.3.2">Section 5.3.2 of RFC 7231</a>

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

@@ -18,6 +18,7 @@ package org.springframework.web.reactive.function.client;
 
 import java.net.URI;
 import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
@@ -26,7 +27,6 @@ import java.util.Arrays;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
@@ -36,6 +36,8 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.core.ParameterizedTypeReference;
+import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
@@ -46,6 +48,7 @@ import org.springframework.lang.Nullable;
 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.BodyExtractor;
 import org.springframework.web.reactive.function.BodyExtractors;
@@ -378,22 +381,11 @@ class DefaultWebClient implements WebClient {
 
 	private static class DefaultResponseSpec implements ResponseSpec {
 
-		private static final Function<ClientResponse, Optional<? extends Throwable>> DEFAULT_STATUS_HANDLER =
+		private static final StatusHandler DEFAULT_STATUS_HANDLER = new StatusHandler(HttpStatus::isError, DefaultResponseSpec::createResponseException);
-				clientResponse -> {
-					HttpStatus statusCode = clientResponse.statusCode();
-					if (statusCode.isError()) {
-						return Optional.of(new WebClientException(
-								"ClientResponse has erroneous status code: " + statusCode.value() +
-										" " + statusCode.getReasonPhrase()));
-					} else {
-						return Optional.empty();
-					}
-				};
 
 		private final Mono<ClientResponse> responseMono;
 
-		private List<Function<ClientResponse, Optional<? extends Throwable>>> statusHandlers =
+		private List<StatusHandler> statusHandlers = new ArrayList<>(1);
-				new ArrayList<>(1);
 
 
 		DefaultResponseSpec(Mono<ClientResponse> responseMono) {
@@ -403,7 +395,7 @@ class DefaultWebClient implements WebClient {
 
 		@Override
 		public ResponseSpec onStatus(Predicate<HttpStatus> statusPredicate,
-				Function<ClientResponse, ? extends Throwable> exceptionFunction) {
+				Function<ClientResponse, Mono<? extends Throwable>> exceptionFunction) {
 
 			Assert.notNull(statusPredicate, "'statusPredicate' must not be null");
 			Assert.notNull(exceptionFunction, "'exceptionFunction' must not be null");
@@ -412,60 +404,107 @@ class DefaultWebClient implements WebClient {
 				this.statusHandlers.clear();
 			}
 
-			Function<ClientResponse, Optional<? extends Throwable>> statusHandler =
+			this.statusHandlers.add(new StatusHandler(statusPredicate, exceptionFunction));
-					clientResponse -> {
-						if (statusPredicate.test(clientResponse.statusCode())) {
-							return Optional.of(exceptionFunction.apply(clientResponse));
-						}
-						else {
-							return Optional.empty();
-						}
-					};
-			this.statusHandlers.add(statusHandler);
 
 			return this;
 		}
 
 		@Override
+		@SuppressWarnings("unchecked")
 		public <T> Mono<T> bodyToMono(Class<T> bodyType) {
 			return this.responseMono.flatMap(
 					response -> bodyToPublisher(response, BodyExtractors.toMono(bodyType),
-							Mono::error));
+							this::monoThrowableToMono));
 		}
 
 		@Override
+		@SuppressWarnings("unchecked")
 		public <T> Mono<T> bodyToMono(ParameterizedTypeReference<T> typeReference) {
 			return this.responseMono.flatMap(
 					response -> bodyToPublisher(response, BodyExtractors.toMono(typeReference),
-							Mono::error));
+							mono -> (Mono<T>)mono));
 		}
 
-		@Override
+		private <T> Mono<T> monoThrowableToMono(Mono<? extends Throwable> mono) {
+			return mono.flatMap(Mono::error);
+		}
+
+ 		@Override
 		public <T> Flux<T> bodyToFlux(Class<T> elementType) {
 			return this.responseMono.flatMapMany(
 					response -> bodyToPublisher(response, BodyExtractors.toFlux(elementType),
-							Flux::error));
+							this::monoThrowableToFlux));
 		}
 
 		@Override
 		public <T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> typeReference) {
 			return this.responseMono.flatMapMany(
 					response -> bodyToPublisher(response, BodyExtractors.toFlux(typeReference),
-							Flux::error));
+							this::monoThrowableToFlux));
+		}
+
+		private <T> Flux<T> monoThrowableToFlux(Mono<? extends Throwable> mono) {
+			return mono.flatMapMany(Flux::error);
 		}
 
 		private <T extends Publisher<?>> T bodyToPublisher(ClientResponse response,
 				BodyExtractor<T, ? super ClientHttpResponse> extractor,
-				Function<Throwable, T> errorFunction) {
+				Function<Mono<? extends Throwable>, T> errorFunction) {
 
 			return this.statusHandlers.stream()
-					.map(statusHandler -> statusHandler.apply(response))
+					.filter(statusHandler -> statusHandler.test(response.statusCode()))
-					.filter(Optional::isPresent)
 					.findFirst()
-					.map(Optional::get)
+					.map(statusHandler -> statusHandler.apply(response))
 					.map(errorFunction::apply)
 					.orElse(response.body(extractor));
 		}
 
+		private static Mono<WebClientResponseException> createResponseException(ClientResponse response) {
+
+			return response.body(BodyExtractors.toDataBuffers())
+					.reduce(DataBuffer::write)
+					.map(dataBuffer -> {
+						byte[] bytes = new byte[dataBuffer.readableByteCount()];
+						dataBuffer.read(bytes);
+						DataBufferUtils.release(dataBuffer);
+						return bytes;
+					})
+					.map(bodyBytes -> {
+						String msg = String.format("ClientResponse has erroneous status code: %d %s", response.statusCode().value(),
+								response.statusCode().getReasonPhrase());
+						Charset charset = response.headers().contentType()
+								.map(MimeType::getCharset)
+								.orElse(StandardCharsets.ISO_8859_1);
+						return new WebClientResponseException(msg,
+								response.statusCode().value(),
+								response.statusCode().getReasonPhrase(),
+								response.headers().asHttpHeaders(),
+								bodyBytes,
+								charset
+								);
+					});
+		}
+
+		private static class StatusHandler {
+
+			private final Predicate<HttpStatus> predicate;
+
+			private final Function<ClientResponse, Mono<? extends Throwable>> exceptionFunction;
+
+			public StatusHandler(Predicate<HttpStatus> predicate,
+					Function<ClientResponse, Mono<? extends Throwable>> exceptionFunction) {
+				this.predicate = predicate;
+				this.exceptionFunction = exceptionFunction;
+			}
+
+			public boolean test(HttpStatus status) {
+				return this.predicate.test(status);
+			}
+
+			public Mono<? extends Throwable> apply(ClientResponse response) {
+				return this.exceptionFunction.apply(response);
+			}
+		}
+
 	}
 }

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

@@ -562,15 +562,15 @@ public interface WebClient {
 		 * Register a custom error function that gets invoked when the given {@link HttpStatus}
 		 * predicate applies. The exception returned from the function will be returned from
 		 * {@link #bodyToMono(Class)} and {@link #bodyToFlux(Class)}.
-		 * <p>By default, an error handler is register that throws a {@link WebClientException}
+		 * <p>By default, an error handler is register that throws a
-		 * when the response status code is 4xx or 5xx.
+		 * {@link WebClientResponseException} when the response status code is 4xx or 5xx.
 		 * @param statusPredicate a predicate that indicates whether {@code exceptionFunction}
 		 * applies
 		 * @param exceptionFunction the function that returns the exception
 		 * @return this builder
 		 */
 		ResponseSpec onStatus(Predicate<HttpStatus> statusPredicate,
-				Function<ClientResponse, ? extends Throwable> exceptionFunction);
+				Function<ClientResponse, Mono<? extends Throwable>> exceptionFunction);
 
 		/**
 		 * Extract the body to a {@code Mono}. By default, if the response has status code 4xx or
@@ -578,8 +578,8 @@ public interface WebClient {
 		 * with {@link #onStatus(Predicate, Function)}.
 		 * @param bodyType the expected response body type
 		 * @param <T> response body type
-		 * @return a mono containing the body, or a {@link WebClientException} if the status code is
+		 * @return a mono containing the body, or a {@link WebClientResponseException} if the
-		 * 4xx or 5xx
+		 * status code is 4xx or 5xx
 		 */
 		<T> Mono<T> bodyToMono(Class<T> bodyType);
 
@@ -589,8 +589,8 @@ public interface WebClient {
 		 * with {@link #onStatus(Predicate, Function)}.
 		 * @param typeReference a type reference describing the expected response body type
 		 * @param <T> response body type
-		 * @return a mono containing the body, or a {@link WebClientException} if the status code is
+		 * @return a mono containing the body, or a {@link WebClientResponseException} if the
-		 * 4xx or 5xx
+		 * status code is 4xx or 5xx
 		 */
 		<T> Mono<T> bodyToMono(ParameterizedTypeReference<T> typeReference);
 
@@ -600,8 +600,8 @@ public interface WebClient {
          * with {@link #onStatus(Predicate, Function)}.
 		 * @param elementType the type of element in the response
 		 * @param <T> the type of elements in the response
-		 * @return a flux containing the body, or a {@link WebClientException} if the status code is
+		 * @return a flux containing the body, or a {@link WebClientResponseException} if the
-		 * 4xx or 5xx
+		 * status code is 4xx or 5xx
 		 */
 		<T> Flux<T> bodyToFlux(Class<T> elementType);
 
@@ -611,8 +611,8 @@ public interface WebClient {
          * with {@link #onStatus(Predicate, Function)}.
 		 * @param typeReference a type reference describing the expected response body type
 		 * @param <T> the type of elements in the response
-		 * @return a flux containing the body, or a {@link WebClientException} if the status code is
+		 * @return a flux containing the body, or a {@link WebClientResponseException} if the
-		 * 4xx or 5xx
+		 * status code is 4xx or 5xx
 		 */
 		<T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> typeReference);
 

spring-webflux/src/main/java/org/springframework/web/reactive/function/client/WebClientException.java

@@ -19,13 +19,14 @@ package org.springframework.web.reactive.function.client;
 import org.springframework.core.NestedRuntimeException;
 
 /**
- * Exception published by {@link WebClient} in case of errors.
+ * Abstract base class for exception published by {@link WebClient} in case of errors.
-  *
+ *
  * @author Arjen Poutsma
  * @since 5.0
  */
-@SuppressWarnings("serial")
+public abstract class WebClientException extends NestedRuntimeException {
-public class WebClientException extends NestedRuntimeException {
+
+	private static final long serialVersionUID = 472776714118912855L;
 
 	/**
 	 * Construct a new instance of {@code WebClientException} with the given message.

spring-webflux/src/main/java/org/springframework/web/reactive/function/client/WebClientResponseException.java

@@ -0,0 +1,112 @@
+/*
+ * Copyright 2002-2017 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.web.reactive.function.client;
+
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
+import org.springframework.lang.Nullable;
+
+/**
+ * Exceptions that contain actual HTTP response data.
+ *
+ * @author Arjen Poutsma
+ * @since 5.0
+ */
+public class WebClientResponseException extends WebClientException {
+
+	private static final long serialVersionUID = 4127543205414951611L;
+
+
+	private final int statusCode;
+
+	private final String statusText;
+
+	private final byte[] responseBody;
+
+	private final HttpHeaders headers;
+
+	private final Charset responseCharset;
+
+
+	/**
+	 * Construct a new instance of with the given response data.
+	 * @param statusCode the raw status code value
+	 * @param statusText the status text
+	 * @param headers the response headers (may be {@code null})
+	 * @param responseBody the response body content (may be {@code null})
+	 * @param responseCharset the response body charset (may be {@code null})
+	 */
+	public WebClientResponseException(String message, int statusCode, String statusText,
+			@Nullable HttpHeaders headers, @Nullable byte[] responseBody,
+			@Nullable Charset responseCharset) {
+
+		super(message);
+
+		this.statusCode = statusCode;
+		this.statusText = statusText;
+		this.headers = (headers != null ? headers : HttpHeaders.EMPTY);
+		this.responseBody = (responseBody != null ? responseBody : new byte[0]);
+		this.responseCharset = (responseCharset != null ? responseCharset : StandardCharsets.ISO_8859_1);
+	}
+
+
+	/**
+	 * Return the HTTP status code value.
+	 */
+	public HttpStatus getStatusCode() {
+		return HttpStatus.valueOf(this.statusCode);
+	}
+
+	/**
+	 * Return the raw HTTP status code value.
+	 */
+	public int getRawStatusCode() {
+		return this.statusCode;
+	}
+
+	/**
+	 * Return the HTTP status text.
+	 */
+	public String getStatusText() {
+		return this.statusText;
+	}
+
+	/**
+	 * Return the HTTP response headers.
+	 */
+	public HttpHeaders getHeaders() {
+		return this.headers;
+	}
+
+	/**
+	 * Return the response body as a byte array.
+	 */
+	public byte[] getResponseBodyAsByteArray() {
+		return this.responseBody;
+	}
+
+	/**
+	 * Return the response body as a string.
+	 */
+	public String getResponseBodyAsString() {
+		return new String(this.responseBody, this.responseCharset);
+	}
+
+}

spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientIntegrationTests.java

@@ -419,8 +419,9 @@ public class WebClientIntegrationTests {
 
 	@Test
 	public void retrieveBodyToMonoInternalServerError() throws Exception {
+		String errorMessage = "Internal Server error";
 		this.server.enqueue(new MockResponse().setResponseCode(500)
-				.setHeader("Content-Type", "text/plain").setBody("Internal Server error"));
+				.setHeader("Content-Type", "text/plain").setBody(errorMessage));
 
 		Mono<String> result = this.webClient.get()
 				.uri("/greeting?name=Spring")
@@ -428,7 +429,14 @@ public class WebClientIntegrationTests {
 				.bodyToMono(String.class);
 
 		StepVerifier.create(result)
-				.expectError(WebClientException.class)
+				.expectErrorSatisfies(throwable -> {
+					assertTrue(throwable instanceof WebClientResponseException);
+					WebClientResponseException ex = (WebClientResponseException) throwable;
+
+					assertEquals(HttpStatus.INTERNAL_SERVER_ERROR, ex.getStatusCode());
+					assertEquals(MediaType.TEXT_PLAIN, ex.getHeaders().getContentType());
+					assertEquals(errorMessage, ex.getResponseBodyAsString());
+				})
 				.verify(Duration.ofSeconds(3));
 
 		RecordedRequest recordedRequest = server.takeRequest();
@@ -445,7 +453,7 @@ public class WebClientIntegrationTests {
 		Mono<String> result = this.webClient.get()
 				.uri("/greeting?name=Spring")
 				.retrieve()
-				.onStatus(HttpStatus::is5xxServerError, response -> new MyException("500 error!"))
+				.onStatus(HttpStatus::is5xxServerError, response -> Mono.just(new MyException("500 error!")))
 				.bodyToMono(String.class);
 
 		StepVerifier.create(result)