Add ClientResponse::createException

This commit adds the createException() method to ClientResponse,
returning a delayed WebClientResponseException based on the status code,
headers, and body as well as the corresponding request.

Closes gh-22825

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2018 the original author or authors.
+ * Copyright 2002-2019 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.
@@ -28,6 +28,7 @@
 import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpRequest;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseCookie;
@@ -162,6 +163,14 @@ public interface ClientResponse {
 	 */
 	<T> Mono<ResponseEntity<List<T>>> toEntityList(ParameterizedTypeReference<T> elementTypeRef);
 
+	/**
+	 * Creates a {@link WebClientResponseException} based on the status code,
+	 * headers, and body of this response as well as the corresponding request.
+	 *
+	 * @return a {@code Mono} with a {@code WebClientResponseException} based on this response
+	 */
+	Mono<WebClientResponseException> createException();
+
 
 	// Static builder methods
 
@@ -317,6 +326,13 @@ interface Builder {
 		 */
 		Builder body(String body);
 
+		/**
+		 * Set the request associated with the response.
+		 * @param request the request
+		 * @return this builder
+		 */
+		Builder request(HttpRequest request);
+
 		/**
 		 * Build the response.
 		 */

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

@@ -16,25 +16,31 @@
 
 package org.springframework.web.reactive.function.client;
 
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.OptionalLong;
+import java.util.function.Supplier;
 
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.core.codec.Hints;
+import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpRequest;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseCookie;
 import org.springframework.http.ResponseEntity;
 import org.springframework.http.client.reactive.ClientHttpResponse;
 import org.springframework.http.codec.HttpMessageReader;
 import org.springframework.http.server.reactive.ServerHttpResponse;
+import org.springframework.util.MimeType;
 import org.springframework.util.MultiValueMap;
 import org.springframework.web.reactive.function.BodyExtractor;
 import org.springframework.web.reactive.function.BodyExtractors;
@@ -58,15 +64,18 @@ class DefaultClientResponse implements ClientResponse {
 
 	private final String requestDescription;
 
+	private final Supplier<HttpRequest> requestSupplier;
+
 
 	public DefaultClientResponse(ClientHttpResponse response, ExchangeStrategies strategies,
-			String logPrefix, String requestDescription) {
+			String logPrefix, String requestDescription, Supplier<HttpRequest> requestSupplier) {
 
 		this.response = response;
 		this.strategies = strategies;
 		this.headers = new DefaultHeaders();
 		this.logPrefix = logPrefix;
 		this.requestDescription = requestDescription;
+		this.requestSupplier = requestSupplier;
 	}
 
 
@@ -175,6 +184,46 @@ public <T> Mono<ResponseEntity<List<T>>> toEntityList(ParameterizedTypeReference
 		return toEntityListInternal(bodyToFlux(elementTypeRef));
 	}
 
+	@Override
+	public Mono<WebClientResponseException> createException() {
+		return DataBufferUtils.join(body(BodyExtractors.toDataBuffers()))
+				.map(dataBuffer -> {
+					byte[] bytes = new byte[dataBuffer.readableByteCount()];
+					dataBuffer.read(bytes);
+					DataBufferUtils.release(dataBuffer);
+					return bytes;
+				})
+				.defaultIfEmpty(new byte[0])
+				.map(bodyBytes -> {
+					HttpRequest request = this.requestSupplier.get();
+					Charset charset = headers().contentType()
+							.map(MimeType::getCharset)
+							.orElse(StandardCharsets.ISO_8859_1);
+					if (HttpStatus.resolve(rawStatusCode()) != null) {
+						return WebClientResponseException.create(
+								statusCode().value(),
+								statusCode().getReasonPhrase(),
+								headers().asHttpHeaders(),
+								bodyBytes,
+								charset,
+								request);
+					}
+					else {
+						return new UnknownHttpStatusCodeException(
+								rawStatusCode(),
+								headers().asHttpHeaders(),
+								bodyBytes,
+								charset,
+								request);
+					}
+				});
+	}
+
+	// Used by DefaultClientResponseBuilder
+	HttpRequest request() {
+		return this.requestSupplier.get();
+	}
+
 	private <T> Mono<ResponseEntity<List<T>>> toEntityListInternal(Flux<T> bodyFlux) {
 		HttpHeaders headers = headers().asHttpHeaders();
 		int status = rawStatusCode();

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

@@ -26,9 +26,11 @@
 import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpRequest;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.ResponseCookie;
 import org.springframework.http.client.reactive.ClientHttpResponse;
+import org.springframework.lang.Nullable;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.LinkedMultiValueMap;
@@ -52,6 +54,9 @@ final class DefaultClientResponseBuilder implements ClientResponse.Builder {
 
 	private Flux<DataBuffer> body = Flux.empty();
 
+	@Nullable
+	private HttpRequest request;
+
 
 	public DefaultClientResponseBuilder(ExchangeStrategies strategies) {
 		Assert.notNull(strategies, "ExchangeStrategies must not be null");
@@ -64,6 +69,9 @@ public DefaultClientResponseBuilder(ClientResponse other) {
 		statusCode(other.statusCode());
 		headers(headers -> headers.addAll(other.headers().asHttpHeaders()));
 		cookies(cookies -> cookies.addAll(other.cookies()));
+		if (other instanceof DefaultClientResponse) {
+			this.request = ((DefaultClientResponse) other).request();
+		}
 	}
 
 
@@ -127,6 +135,13 @@ private void releaseBody() {
 		this.body.subscribe(DataBufferUtils.releaseConsumer());
 	}
 
+	@Override
+	public ClientResponse.Builder request(HttpRequest request) {
+		Assert.notNull(request, "Request must not be null");
+		this.request = request;
+		return this;
+	}
+
 	@Override
 	public ClientResponse build() {
 
@@ -136,7 +151,7 @@ public ClientResponse build() {
 		// When building ClientResponse manually, the ClientRequest.logPrefix() has to be passed,
 		// e.g. via ClientResponse.Builder, but this (builder) is not used currently.
 
-		return new DefaultClientResponse(httpResponse, this.strategies, "", "");
+		return new DefaultClientResponse(httpResponse, this.strategies, "", "", () -> this.request);
 	}
 
 

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

@@ -18,14 +18,12 @@
 
 import java.net.URI;
 import java.nio.charset.Charset;
-import java.nio.charset.StandardCharsets;
 import java.time.ZonedDateTime;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.function.BiFunction;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Predicate;
@@ -36,7 +34,6 @@
 import reactor.core.publisher.Mono;
 
 import org.springframework.core.ParameterizedTypeReference;
-import org.springframework.core.io.buffer.DataBufferUtils;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpRequest;
@@ -47,9 +44,7 @@
 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.BodyExtractors;
 import org.springframework.web.reactive.function.BodyInserter;
 import org.springframework.web.reactive.function.BodyInserters;
 import org.springframework.web.util.DefaultUriBuilderFactory;
@@ -421,7 +416,7 @@ public HttpHeaders getHeaders() {
 	private static class DefaultResponseSpec implements ResponseSpec {
 
 		private static final StatusHandler DEFAULT_STATUS_HANDLER =
-				new StatusHandler(HttpStatus::isError, DefaultResponseSpec::createResponseException);
+				new StatusHandler(HttpStatus::isError, ClientResponse::createException);
 
 		private final Mono<ClientResponse> responseMono;
 
@@ -442,8 +437,7 @@ public ResponseSpec onStatus(Predicate<HttpStatus> statusPredicate,
 			if (this.statusHandlers.size() == 1 && this.statusHandlers.get(0) == DEFAULT_STATUS_HANDLER) {
 				this.statusHandlers.clear();
 			}
-			this.statusHandlers.add(new StatusHandler(statusPredicate,
-					(clientResponse, request) -> exceptionFunction.apply(clientResponse)));
+			this.statusHandlers.add(new StatusHandler(statusPredicate, exceptionFunction));
 
 			return this;
 		}
@@ -478,24 +472,24 @@ private <T extends Publisher<?>> T handleBody(ClientResponse response,
 			if (HttpStatus.resolve(response.rawStatusCode()) != null) {
 				for (StatusHandler handler : this.statusHandlers) {
 					if (handler.test(response.statusCode())) {
-						HttpRequest request = this.requestSupplier.get();
 						Mono<? extends Throwable> exMono;
 						try {
-							exMono = handler.apply(response, request);
+							exMono = handler.apply(response);
 							exMono = exMono.flatMap(ex -> drainBody(response, ex));
 							exMono = exMono.onErrorResume(ex -> drainBody(response, ex));
 						}
 						catch (Throwable ex2) {
 							exMono = drainBody(response, ex2);
 						}
 						T result = errorFunction.apply(exMono);
+						HttpRequest request = this.requestSupplier.get();
 						return insertCheckpoint(result, response.statusCode(), request);
 					}
 				}
 				return bodyPublisher;
 			}
 			else {
-				return errorFunction.apply(createResponseException(response, this.requestSupplier.get()));
+				return errorFunction.apply(response.createException());
 			}
 		}
 
@@ -523,50 +517,15 @@ else if (result instanceof Flux) {
 			}
 		}
 
-		private static Mono<WebClientResponseException> createResponseException(
-				ClientResponse response, HttpRequest request) {
-
-			return DataBufferUtils.join(response.body(BodyExtractors.toDataBuffers()))
-					.map(dataBuffer -> {
-						byte[] bytes = new byte[dataBuffer.readableByteCount()];
-						dataBuffer.read(bytes);
-						DataBufferUtils.release(dataBuffer);
-						return bytes;
-					})
-					.defaultIfEmpty(new byte[0])
-					.map(bodyBytes -> {
-						Charset charset = response.headers().contentType()
-								.map(MimeType::getCharset)
-								.orElse(StandardCharsets.ISO_8859_1);
-						if (HttpStatus.resolve(response.rawStatusCode()) != null) {
-							return WebClientResponseException.create(
-									response.statusCode().value(),
-									response.statusCode().getReasonPhrase(),
-									response.headers().asHttpHeaders(),
-									bodyBytes,
-									charset,
-									request);
-						}
-						else {
-							return new UnknownHttpStatusCodeException(
-									response.rawStatusCode(),
-									response.headers().asHttpHeaders(),
-									bodyBytes,
-									charset,
-									request);
-						}
-					});
-		}
-
 
 		private static class StatusHandler {
 
 			private final Predicate<HttpStatus> predicate;
 
-			private final BiFunction<ClientResponse, HttpRequest, Mono<? extends Throwable>> exceptionFunction;
+			private final Function<ClientResponse, Mono<? extends Throwable>> exceptionFunction;
 
 			public StatusHandler(Predicate<HttpStatus> predicate,
-					BiFunction<ClientResponse, HttpRequest, Mono<? extends Throwable>> exceptionFunction) {
+					Function<ClientResponse, Mono<? extends Throwable>> exceptionFunction) {
 
 				Assert.notNull(predicate, "Predicate must not be null");
 				Assert.notNull(exceptionFunction, "Function must not be null");
@@ -578,8 +537,8 @@ public boolean test(HttpStatus status) {
 				return this.predicate.test(status);
 			}
 
-			public Mono<? extends Throwable> apply(ClientResponse response, HttpRequest request) {
-				return this.exceptionFunction.apply(response, request);
+			public Mono<? extends Throwable> apply(ClientResponse response) {
+				return this.exceptionFunction.apply(response);
 			}
 		}
 	}

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

@@ -25,6 +25,7 @@
 import org.springframework.core.log.LogFormatUtils;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
+import org.springframework.http.HttpRequest;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.client.reactive.ClientHttpConnector;
 import org.springframework.http.client.reactive.ClientHttpResponse;
@@ -106,7 +107,8 @@ public Mono<ClientResponse> exchange(ClientRequest clientRequest) {
 					.map(httpResponse -> {
 						logResponse(httpResponse, logPrefix);
 						return new DefaultClientResponse(
-								httpResponse, this.strategies, logPrefix, httpMethod.name() + " " + url);
+								httpResponse, this.strategies, logPrefix, httpMethod.name() + " " + url,
+								() -> createRequest(clientRequest));
 					});
 		}
 
@@ -129,6 +131,31 @@ private void logResponse(ClientHttpResponse response, String logPrefix) {
 		private String formatHeaders(HttpHeaders headers) {
 			return this.enableLoggingRequestDetails ? headers.toString() : headers.isEmpty() ? "{}" : "{masked}";
 		}
+
+		private HttpRequest createRequest(ClientRequest request) {
+			return new HttpRequest() {
+
+				@Override
+				public HttpMethod getMethod() {
+					return request.method();
+				}
+
+				@Override
+				public String getMethodValue() {
+					return request.method().name();
+				}
+
+				@Override
+				public URI getURI() {
+					return request.url();
+				}
+
+				@Override
+				public HttpHeaders getHeaders() {
+					return request.headers();
+				}
+			};
+		}
 	}
 
 }