Wrap exceptions in WebClient

This commit makes sure that exceptions emitted by WebClient are wrapped
by WebClientExceptions:

- Exceptions emitted by the ClientHttpConnector are wrapped in a new
  WebClientRequestException.

- Exceptions emitted after a response is received are wrapped in a
  WebClientResponseException

Closes gh-23842

Author: poutsma

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

@@ -55,6 +55,9 @@
  */
 class DefaultClientResponse implements ClientResponse {
 
+	private static final byte[] EMPTY = new byte[0];
+
+
 	private final ClientHttpResponse response;
 
 	private final Headers headers;
@@ -200,7 +203,8 @@ public Mono<WebClientResponseException> createException() {
 					DataBufferUtils.release(dataBuffer);
 					return bytes;
 				})
-				.defaultIfEmpty(new byte[0])
+				.defaultIfEmpty(EMPTY)
+				.onErrorReturn(IllegalStateException.class::isInstance, EMPTY)
 				.map(bodyBytes -> {
 					HttpRequest request = this.requestSupplier.get();
 					Charset charset = headers().contentType()

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

@@ -488,11 +488,13 @@ public <T> Mono<T> bodyToMono(ParameterizedTypeReference<T> elementTypeRef) {
 
 		private <T> Mono<T> handleBodyMono(ClientResponse response, Mono<T> bodyPublisher) {
 			Mono<T> result = statusHandlers(response);
+			Mono<T> wrappedExceptions = bodyPublisher.onErrorResume(WebClientUtils::shouldWrapException,
+							t -> wrapException(t, response));
 			if (result != null) {
-				return result.switchIfEmpty(bodyPublisher);
+				return result.switchIfEmpty(wrappedExceptions);
 			}
 			else {
-				return bodyPublisher;
+				return wrappedExceptions;
 			}
 		}
 
@@ -510,11 +512,13 @@ public <T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> elementTypeRef) {
 
 		private <T> Publisher<T> handleBodyFlux(ClientResponse response, Flux<T> bodyPublisher) {
 			Mono<T> result = statusHandlers(response);
+			Flux<T> wrappedExceptions = bodyPublisher.onErrorResume(WebClientUtils::shouldWrapException,
+					t -> wrapException(t, response));
 			if (result != null) {
-				return result.flux().switchIfEmpty(bodyPublisher);
+				return result.flux().switchIfEmpty(wrappedExceptions);
 			}
 			else {
-				return bodyPublisher;
+				return wrappedExceptions;
 			}
 		}
 
@@ -555,6 +559,12 @@ private <T> Mono<T> insertCheckpoint(Mono<T> result, int statusCode, HttpRequest
 			return result.checkpoint(description);
 		}
 
+		private <T> Mono<T> wrapException(Throwable throwable, ClientResponse response) {
+			return response.createException()
+					.map(responseException -> responseException.initCause(throwable))
+					.flatMap(Mono::error);
+		}
+
 		@Override
 		public <T> Mono<ResponseEntity<T>> toEntity(Class<T> bodyClass) {
 			return this.responseMono.flatMap(response ->

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

@@ -104,6 +104,7 @@ public Mono<ClientResponse> exchange(ClientRequest clientRequest) {
 					.connect(httpMethod, url, httpRequest -> clientRequest.writeTo(httpRequest, this.strategies))
 					.doOnRequest(n -> logRequest(clientRequest))
 					.doOnCancel(() -> logger.debug(logPrefix + "Cancel signal (to close connection)"))
+					.onErrorResume(WebClientUtils::shouldWrapException, t -> wrapException(t, clientRequest))
 					.map(httpResponse -> {
 						logResponse(httpResponse, logPrefix);
 						return new DefaultClientResponse(
@@ -132,6 +133,10 @@ private String formatHeaders(HttpHeaders headers) {
 			return this.enableLoggingRequestDetails ? headers.toString() : headers.isEmpty() ? "{}" : "{masked}";
 		}
 
+		private <T> Mono<T> wrapException(Throwable t, ClientRequest r) {
+			return Mono.error(() -> new WebClientRequestException(t, r.method(), r.url(), r.headers()));
+		}
+
 		private HttpRequest createRequest(ClientRequest request) {
 			return new HttpRequest() {
 

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

@@ -0,0 +1,74 @@
+/*
+ * 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.
+ * You may obtain a copy of the License at
+ *
+ *      https://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.net.URI;
+
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpMethod;
+
+/**
+ * Exceptions that contain actual HTTP request data.
+ *
+ * @author Arjen Poutsma
+ * @since 5.3
+ */
+public class WebClientRequestException extends WebClientException {
+
+	private static final long serialVersionUID = -5139991985321385005L;
+
+
+	private final HttpMethod method;
+
+	private final URI uri;
+
+	private final HttpHeaders headers;
+
+
+	/**
+	 * Constructor for throwable.
+	 */
+	public WebClientRequestException(Throwable ex, HttpMethod method, URI uri, HttpHeaders headers) {
+		super(ex.getMessage(), ex);
+
+		this.method = method;
+		this.uri = uri;
+		this.headers = headers;
+	}
+
+	/**
+	 * Return the HTTP request method.
+	 */
+	public HttpMethod getMethod() {
+		return this.method;
+	}
+
+	/**
+	 * Return the request URI.
+	 */
+	public URI getUri() {
+		return this.uri;
+	}
+
+	/**
+	 * Return the HTTP request headers.
+	 */
+	public HttpHeaders getHeaders() {
+		return this.headers;
+	}
+
+}

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

@@ -22,6 +22,7 @@
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import org.springframework.core.codec.CodecException;
 import org.springframework.http.ResponseEntity;
 
 /**
@@ -56,4 +57,10 @@ public static <T> Mono<ResponseEntity<List<T>>> mapToEntityList(ClientResponse r
 						.body(list));
 	}
 
+	/**
+	 * Indicates whether the given exception should be wrapped.
+	 */
+	public static boolean shouldWrapException(Throwable t) {
+		return !(t instanceof WebClientException) && !(t instanceof CodecException);
+	}
 }

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

@@ -40,7 +40,6 @@
 import org.springframework.http.ResponseEntity;
 import org.springframework.http.client.reactive.ReactorClientHttpConnector;
 import org.springframework.http.client.reactive.ReactorResourceFactory;
-import org.springframework.web.reactive.function.UnsupportedMediaTypeException;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.TestInstance.Lifecycle.PER_CLASS;
@@ -127,7 +126,7 @@ void bodyToMonoVoidWithoutContentType(String displayName, DataBufferFactory buff
 				.retrieve()
 				.bodyToMono(new ParameterizedTypeReference<Map<String, String>>() {});
 
-		StepVerifier.create(mono).expectError(UnsupportedMediaTypeException.class).verify(Duration.ofSeconds(3));
+		StepVerifier.create(mono).expectError(WebClientResponseException.class).verify(Duration.ofSeconds(3));
 		assertThat(this.server.getRequestCount()).isEqualTo(1);
 	}
 

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

@@ -1013,7 +1013,12 @@ void exchangeWithRelativeUrl(ClientHttpConnector connector) {
 		Mono<ClientResponse> responseMono = WebClient.builder().build().get().uri(uri).exchange();
 
 		StepVerifier.create(responseMono)
-				.expectErrorMessage("URI is not absolute: " + uri)
+				.expectErrorSatisfies(throwable -> {
+					assertThat(throwable).isInstanceOf(WebClientRequestException.class);
+					WebClientRequestException ex = (WebClientRequestException) throwable;
+					assertThat(ex.getMethod()).isEqualTo(HttpMethod.GET);
+					assertThat(ex.getUri()).isEqualTo(URI.create(uri));
+				})
 				.verify(Duration.ofSeconds(5));
 	}
 
@@ -1126,6 +1131,25 @@ void exchangeResponseCookies(ClientHttpConnector connector) {
 		expectRequestCount(1);
 	}
 
+	@ParameterizedWebClientTest
+	void invalidDomain(ClientHttpConnector connector) {
+		startServer(connector);
+
+		String url = "http://example.invalid";
+		Mono<ClientResponse> result = this.webClient.get().
+				uri(url)
+				.exchange();
+
+		StepVerifier.create(result)
+				.expectErrorSatisfies(throwable -> {
+					assertThat(throwable).isInstanceOf(WebClientRequestException.class);
+					WebClientRequestException ex = (WebClientRequestException) throwable;
+					assertThat(ex.getMethod()).isEqualTo(HttpMethod.GET);
+					assertThat(ex.getUri()).isEqualTo(URI.create(url));
+				})
+				.verify();
+	}
+
 
 	private void prepareResponse(Consumer<MockResponse> consumer) {
 		MockResponse response = new MockResponse();