Add client-side exception hierarchy

The `WebClient` has a new exception hierarchy:

* `WebClientException` is the root of all exceptions thrown by the
`WebClient`
* `WebClientResponseException` are all exceptions associated with
specific HTTP response status codes
* `WebClientErrorException` and `WebServerErrorException` are
respectively for 4xx and 5xx HTTP status codes

`ResponseExtractor` implementations are adapted to optionally throw
exceptions if it's impossible to extract the relevant parts of the
response (e.g. extracting the response body if the response is a 404).

This commit also introduces `ResponseErrorHandler`s that take care of
the whole exception mapping infrastructure. Since
`WebClientResponseException`s provide the status, headers and response
body, we also need a dedicated mechanism to extract information from the
response body at that level.

The `BodyExtractors` are responsible for extracting that information
from the exception, given they are provided with all the information
they need; in that case, message decoders are required.

To convey all this new information downstream, the `WebClient` now wraps
the message converters and response error handler instances into a
dedicated `WebClientConfig` object.

spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpConnector.java

@@ -22,11 +22,14 @@ import java.util.function.Function;
 import org.springframework.http.HttpMethod;
 
 import reactor.core.publisher.Mono;
+import reactor.io.netty.http.HttpException;
+import reactor.io.netty.http.HttpInbound;
 
 /**
  * Reactor-Netty implementation of {@link ClientHttpConnector}
  *
  * @author Brian Clozel
+ * @see reactor.io.netty.http.HttpClient
  * @since 5.0
  */
 public class ReactorClientHttpConnector implements ClientHttpConnector {
@@ -38,7 +41,10 @@ public class ReactorClientHttpConnector implements ClientHttpConnector {
 		return reactor.io.netty.http.HttpClient.create(uri.getHost(), uri.getPort())
 				.request(io.netty.handler.codec.http.HttpMethod.valueOf(method.name()),
 						uri.toString(),
-						httpOutbound -> requestCallback.apply(new ReactorClientHttpRequest(method, uri, httpOutbound)))
+						httpClientRequest -> requestCallback
+								.apply(new ReactorClientHttpRequest(method, uri, httpClientRequest)))
+				.cast(HttpInbound.class)
+				.otherwise(HttpException.class, exc -> Mono.just(exc.getChannel()))
 				.map(httpInbound -> new ReactorClientHttpResponse(httpInbound));
 	}
 }

spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpResponse.java

@@ -19,7 +19,7 @@ package org.springframework.http.client.reactive;
 import java.util.Collection;
 
 import reactor.core.publisher.Flux;
-import reactor.io.netty.http.HttpClientResponse;
+import reactor.io.netty.http.HttpInbound;
 
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.NettyDataBufferFactory;
@@ -34,16 +34,16 @@ import org.springframework.util.MultiValueMap;
  * {@link ClientHttpResponse} implementation for the Reactor-Netty HTTP client
  *
  * @author Brian Clozel
- * @since 5.0
  * @see reactor.io.netty.http.HttpClient
+ * @since 5.0
  */
 public class ReactorClientHttpResponse implements ClientHttpResponse {
 
 	private final NettyDataBufferFactory dataBufferFactory;
 
-	private final HttpClientResponse response;
+	private final HttpInbound response;
 
-	public ReactorClientHttpResponse(HttpClientResponse response) {
+	public ReactorClientHttpResponse(HttpInbound response) {
 		this.response = response;
 		this.dataBufferFactory = new NettyDataBufferFactory(response.delegate().alloc());
 	}

spring-web/src/main/java/org/springframework/web/client/reactive/BodyExtractor.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2002-2016 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.client.reactive;
+
+import java.util.List;
+
+import org.springframework.http.client.reactive.ClientHttpResponse;
+import org.springframework.http.converter.reactive.HttpMessageConverter;
+
+/**
+ * A {@code BodyExtractor} extracts the content of a raw {@link ClientHttpResponse},
+ * decoding the response body and using a target composition API.
+ *
+ * <p>See static factory methods in {@link ResponseExtractors}
+ * and {@link org.springframework.web.client.reactive.support.RxJava1ResponseExtractors}.
+ *
+ * @author Brian Clozel
+ * @since 5.0
+ */
+public interface BodyExtractor<T> {
+
+	/**
+	 * Extract content from the response body
+	 * @param clientResponse the raw HTTP response
+	 * @param messageConverters the message converters that decode the response body
+	 * @return the relevant content
+	 */
+	T extract(ClientHttpResponse clientResponse, List<HttpMessageConverter<?>> messageConverters);
+}

spring-web/src/main/java/org/springframework/web/client/reactive/DefaultResponseErrorHandler.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2002-2016 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.client.reactive;
+
+import java.util.List;
+
+import org.springframework.http.HttpStatus;
+import org.springframework.http.client.reactive.ClientHttpResponse;
+import org.springframework.http.converter.reactive.HttpMessageConverter;
+
+/**
+ * Default implementation of the {@link ResponseErrorHandler} interface
+ * that throws {@link WebClientErrorException}s for HTTP 4xx responses
+ * and {@link WebServerErrorException}s for HTTP 5xx responses.
+ *
+ * @author Brian Clozel
+ * @since 5.0
+ */
+public class DefaultResponseErrorHandler implements ResponseErrorHandler {
+
+	@Override
+	public void handleError(ClientHttpResponse response, List<HttpMessageConverter<?>> messageConverters) {
+		HttpStatus responseStatus = response.getStatusCode();
+		if (responseStatus.is4xxClientError()) {
+			throw new WebClientErrorException(response, messageConverters);
+		}
+		if (responseStatus.is5xxServerError()) {
+			throw new WebServerErrorException(response, messageConverters);
+		}
+	}
+}

spring-web/src/main/java/org/springframework/web/client/reactive/ResponseErrorHandler.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2002-2016 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.client.reactive;
+
+import java.util.List;
+
+import org.springframework.http.client.reactive.ClientHttpResponse;
+import org.springframework.http.converter.reactive.HttpMessageConverter;
+
+/**
+ * Strategy interface used by the {@link WebClient} to handle
+ * errors in {@link ClientHttpResponse}s if needed.
+ *
+ * @author Brian Clozel
+ * @see DefaultResponseErrorHandler
+ * @since 5.0
+ */
+public interface ResponseErrorHandler {
+
+	/**
+	 * Handle the error in the given response.
+	 * Implementations will typically inspect the {@link ClientHttpResponse#getStatusCode() HttpStatus}
+	 * of the response and throw {@link WebClientException}s in case of errors.
+	 */
+	void handleError(ClientHttpResponse response, List<HttpMessageConverter<?>> messageConverters);
+
+}

spring-web/src/main/java/org/springframework/web/client/reactive/ResponseExtractor.java

@@ -16,24 +16,28 @@
 
 package org.springframework.web.client.reactive;
 
-import java.util.List;
-
 import reactor.core.publisher.Mono;
 
 import org.springframework.http.client.reactive.ClientHttpResponse;
-import org.springframework.http.converter.reactive.HttpMessageConverter;
 
 /**
  * A {@code ResponseExtractor} extracts the relevant part of a
  * raw {@link org.springframework.http.client.reactive.ClientHttpResponse},
  * optionally decoding the response body and using a target composition API.
  *
- * <p>See static factory methods in {@link ResponseExtractors}.
+ * <p>See static factory methods in {@link ResponseExtractors} and
+ * {@link org.springframework.web.client.reactive.support.RxJava1ResponseExtractors}.
  *
  * @author Brian Clozel
  * @since 5.0
  */
 public interface ResponseExtractor<T> {
 
-	T extract(Mono<ClientHttpResponse> clientResponse, List<HttpMessageConverter<?>> messageConverters);
+	/**
+	 * Extract content from the response
+	 * @param clientResponse the raw HTTP response
+	 * @param webClientConfig the {@link WebClient} configuration information
+	 * @return the relevant part of the response
+	 */
+	T extract(Mono<ClientHttpResponse> clientResponse, WebClientConfig webClientConfig);
 }

spring-web/src/main/java/org/springframework/web/client/reactive/ResponseExtractors.java

@@ -17,7 +17,9 @@
 package org.springframework.web.client.reactive;
 
 import java.util.List;
-import java.util.Optional;
+
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
 
 import org.springframework.core.ResolvableType;
 import org.springframework.http.HttpHeaders;
@@ -26,12 +28,9 @@ import org.springframework.http.ResponseEntity;
 import org.springframework.http.client.reactive.ClientHttpResponse;
 import org.springframework.http.converter.reactive.HttpMessageConverter;
 
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
 /**
- * Static factory methods for {@link ResponseExtractor} based on the {@link Flux} and
- * {@link Mono} APIs.
+ * Static factory methods for {@link ResponseExtractor} and {@link BodyExtractor},
+ * based on the {@link Flux} and {@link Mono} APIs.
  *
  * @author Brian Clozel
  * @since 5.0
@@ -46,9 +45,11 @@ public class ResponseExtractors {
 	 */
 	@SuppressWarnings("unchecked")
 	public static <T> ResponseExtractor<Mono<T>> body(ResolvableType bodyType) {
-		return (clientResponse, messageConverters) -> (Mono<T>) clientResponse
-				.flatMap(resp -> decodeResponseBody(resp, bodyType,
-						messageConverters))
+		return (clientResponse, webClientConfig) -> (Mono<T>) clientResponse
+				.doOnNext(response -> webClientConfig.getResponseErrorHandler()
+						.handleError(response, webClientConfig.getMessageConverters()))
+				.flatMap(resp -> decodeResponseBodyAsMono(resp, bodyType,
+						webClientConfig.getMessageConverters()))
 				.next();
 	}
 
@@ -60,13 +61,33 @@ public class ResponseExtractors {
 		return body(bodyType);
 	}
 
+	/**
+	 * Extract the response body and decode it, returning it as a {@code Mono<T>}.
+	 * @see ResolvableType#forClassWithGenerics(Class, Class[])
+	 */
+	@SuppressWarnings("unchecked")
+	public static <T> BodyExtractor<Mono<T>> as(ResolvableType bodyType) {
+		return (clientResponse, messageConverters) ->
+				decodeResponseBodyAsMono(clientResponse, bodyType, messageConverters);
+	}
+
+	/**
+	 * Extract the response body and decode it, returning it as a {@code Mono<T>}
+	 */
+	public static <T> BodyExtractor<Mono<T>> as(Class<T> sourceClass) {
+		ResolvableType bodyType = ResolvableType.forClass(sourceClass);
+		return as(bodyType);
+	}
+
 	/**
 	 * Extract the response body and decode it, returning it as a {@code Flux<T>}.
 	 * @see ResolvableType#forClassWithGenerics(Class, Class[])
 	 */
 	public static <T> ResponseExtractor<Flux<T>> bodyStream(ResolvableType bodyType) {
-		return (clientResponse, messageConverters) -> clientResponse
-				.flatMap(resp -> decodeResponseBody(resp, bodyType, messageConverters));
+		return (clientResponse, webClientConfig) -> clientResponse
+				.doOnNext(response -> webClientConfig.getResponseErrorHandler()
+						.handleError(response, webClientConfig.getMessageConverters()))
+				.flatMap(resp -> decodeResponseBody(resp, bodyType, webClientConfig.getMessageConverters()));
 	}
 
 	/**
@@ -77,22 +98,39 @@ public class ResponseExtractors {
 		return bodyStream(bodyType);
 	}
 
+	/**
+	 * Extract the response body and decode it, returning it as a {@code Flux<T>}
+	 * @see ResolvableType#forClassWithGenerics(Class, Class[])
+	 */
+	@SuppressWarnings("unchecked")
+	public static <T> BodyExtractor<Flux<T>> asStream(ResolvableType bodyType) {
+		return (clientResponse, messageConverters) ->
+				(Flux<T>) decodeResponseBody(clientResponse, bodyType, messageConverters);
+	}
+
+	/**
+	 * Extract the response body and decode it, returning it as a {@code Flux<T>}
+	 */
+	public static <T> BodyExtractor<Flux<T>> asStream(Class<T> sourceClass) {
+		ResolvableType bodyType = ResolvableType.forClass(sourceClass);
+		return asStream(bodyType);
+	}
+
 	/**
 	 * Extract the full response body as a {@code ResponseEntity} with its body decoded as
 	 * a single type {@code T}.
 	 * @see ResolvableType#forClassWithGenerics(Class, Class[])
 	 */
 	@SuppressWarnings("unchecked")
-	public static <T> ResponseExtractor<Mono<ResponseEntity<T>>> response(
-			ResolvableType bodyType) {
-		return (clientResponse, messageConverters) -> clientResponse.then(response -> {
-			return Mono.when(
-					decodeResponseBody(response, bodyType,
-							messageConverters).next().defaultIfEmpty(
-							EMPTY_BODY),
-					Mono.just(response.getHeaders()),
-					Mono.just(response.getStatusCode()));
-		}).map(tuple -> {
+	public static <T> ResponseExtractor<Mono<ResponseEntity<T>>> response(ResolvableType bodyType) {
+
+		return (clientResponse, webClientConfig) -> clientResponse.then(response ->
+				Mono.when(
+						decodeResponseBodyAsMono(response, bodyType,
+								webClientConfig.getMessageConverters()).defaultIfEmpty(EMPTY_BODY),
+						Mono.just(response.getHeaders()),
+						Mono.just(response.getStatusCode()))
+		).map(tuple -> {
 			Object body = (tuple.getT1() != EMPTY_BODY ? tuple.getT1() : null);
 			return new ResponseEntity<>((T) body, tuple.getT2(), tuple.getT3());
 		});
@@ -102,8 +140,7 @@ public class ResponseExtractors {
 	 * Extract the full response body as a {@code ResponseEntity} with its body decoded as
 	 * a single type {@code T}.
 	 */
-	public static <T> ResponseExtractor<Mono<ResponseEntity<T>>> response(
-			Class<T> bodyClass) {
+	public static <T> ResponseExtractor<Mono<ResponseEntity<T>>> response(Class<T> bodyClass) {
 		ResolvableType bodyType = ResolvableType.forClass(bodyClass);
 		return response(bodyType);
 	}
@@ -113,11 +150,10 @@ public class ResponseExtractors {
 	 * a {@code Flux<T>}.
 	 * @see ResolvableType#forClassWithGenerics(Class, Class[])
 	 */
-	public static <T> ResponseExtractor<Mono<ResponseEntity<Flux<T>>>> responseStream(
-			ResolvableType type) {
-		return (clientResponse, messageConverters) -> clientResponse
+	public static <T> ResponseExtractor<Mono<ResponseEntity<Flux<T>>>> responseStream(ResolvableType type) {
+		return (clientResponse, webClientConfig) -> clientResponse
 				.map(response -> new ResponseEntity<>(
-						ResponseExtractors.<T> decodeResponseBody(response, type, messageConverters),
+						decodeResponseBody(response, type, webClientConfig.getMessageConverters()),
 						response.getHeaders(), response.getStatusCode()));
 	}
 
@@ -125,8 +161,7 @@ public class ResponseExtractors {
 	 * Extract the full response body as a {@code ResponseEntity} with its body decoded as
 	 * a {@code Flux<T>}.
 	 */
-	public static <T> ResponseExtractor<Mono<ResponseEntity<Flux<T>>>> responseStream(
-			Class<T> sourceClass) {
+	public static <T> ResponseExtractor<Mono<ResponseEntity<Flux<T>>>> responseStream(Class<T> sourceClass) {
 		ResolvableType resolvableType = ResolvableType.forClass(sourceClass);
 		return responseStream(resolvableType);
 	}
@@ -135,30 +170,37 @@ public class ResponseExtractors {
 	 * Extract the response headers as an {@code HttpHeaders} instance.
 	 */
 	public static ResponseExtractor<Mono<HttpHeaders>> headers() {
-		return (clientResponse, messageConverters) -> clientResponse.map(resp -> resp.getHeaders());
+		return (clientResponse, webClientConfig) -> clientResponse.map(resp -> resp.getHeaders());
 	}
 
 	@SuppressWarnings("unchecked")
 	protected static <T> Flux<T> decodeResponseBody(ClientHttpResponse response,
-			ResolvableType responseType,
-			List<HttpMessageConverter<?>> messageConverters) {
+			ResolvableType responseType, List<HttpMessageConverter<?>> messageConverters) {
 
 		MediaType contentType = response.getHeaders().getContentType();
-		Optional<HttpMessageConverter<?>> converter = resolveConverter(messageConverters,
-				responseType, contentType);
-		if (!converter.isPresent()) {
-			return Flux.error(new IllegalStateException(
-					"Could not decode response body of type '" + contentType
-							+ "' with target type '" + responseType.toString() + "'"));
-		}
-		return (Flux<T>) converter.get().read(responseType, response);
+		HttpMessageConverter<?> converter = resolveConverter(messageConverters, responseType, contentType);
+		return (Flux<T>) converter.read(responseType, response);
 	}
 
-	protected static Optional<HttpMessageConverter<?>> resolveConverter(
-			List<HttpMessageConverter<?>> messageConverters, ResolvableType type,
-			MediaType mediaType) {
-		return messageConverters.stream().filter(e -> e.canRead(type, mediaType))
-				.findFirst();
+	@SuppressWarnings("unchecked")
+	protected static <T> Mono<T> decodeResponseBodyAsMono(ClientHttpResponse response,
+			ResolvableType responseType, List<HttpMessageConverter<?>> messageConverters) {
+
+		MediaType contentType = response.getHeaders().getContentType();
+		HttpMessageConverter<?> converter = resolveConverter(messageConverters, responseType, contentType);
+		return (Mono<T>) converter.readMono(responseType, response);
+	}
+
+	protected static HttpMessageConverter<?> resolveConverter(
+			List<HttpMessageConverter<?>> messageConverters, ResolvableType responseType, MediaType contentType) {
+
+		return messageConverters.stream()
+				.filter(e -> e.canRead(responseType, contentType))
+				.findFirst()
+				.orElseThrow(() ->
+						new WebClientException(
+								"Could not decode response body of type '" + contentType
+										+ "' with target type '" + responseType.toString() + "'"));
 	}
 
 }

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

@@ -92,7 +92,7 @@ public final class WebClient {
 
 	private ClientHttpConnector clientHttpConnector;
 
-	private List<HttpMessageConverter<?>> messageConverters;
+	private final DefaultWebClientConfig webClientConfig;
 
 	/**
 	 * Create a {@code WebClient} instance, using the {@link ClientHttpConnector}
@@ -111,8 +111,11 @@ public final class WebClient {
 	 */
 	public WebClient(ClientHttpConnector clientHttpConnector) {
 		this.clientHttpConnector = clientHttpConnector;
-		this.messageConverters = new ArrayList<>();
-		addDefaultHttpMessageConverters(this.messageConverters);
+		this.webClientConfig = new DefaultWebClientConfig();
+		List<HttpMessageConverter<?>> converters = new ArrayList<>();
+		addDefaultHttpMessageConverters(converters);
+		this.webClientConfig.setMessageConverters(converters);
+		this.webClientConfig.setResponseErrorHandler(new DefaultResponseErrorHandler());
 	}
 
 	/**
@@ -141,7 +144,14 @@ public final class WebClient {
 	 * messages
 	 */
 	public void setMessageConverters(List<HttpMessageConverter<?>> messageConverters) {
-		this.messageConverters = messageConverters;
+		this.webClientConfig.setMessageConverters(messageConverters);
+	}
+
+	/**
+	 * Set the {@link ResponseErrorHandler} to use for handling HTTP response errors
+	 */
+	public void setResponseErrorHandler(ResponseErrorHandler responseErrorHandler) {
+		this.webClientConfig.setResponseErrorHandler(responseErrorHandler);
 	}
 
 	/**
@@ -167,17 +177,41 @@ public final class WebClient {
 		return new WebResponseActions() {
 			@Override
 			public void doWithStatus(Consumer<HttpStatus> consumer) {
-				clientResponse.doOnNext(clientHttpResponse ->
-						consumer.accept(clientHttpResponse.getStatusCode()));
+				clientResponse.doOnNext(clientHttpResponse -> consumer.accept(clientHttpResponse.getStatusCode()));
 			}
 
 			@Override
 			public <T> T extract(ResponseExtractor<T> extractor) {
-				return extractor.extract(clientResponse, messageConverters);
+				return extractor.extract(clientResponse, webClientConfig);
 			}
 		};
 	}
 
+	protected class DefaultWebClientConfig implements WebClientConfig {
+
+		private List<HttpMessageConverter<?>> messageConverters;
+
+		private ResponseErrorHandler responseErrorHandler;
+
+		@Override
+		public List<HttpMessageConverter<?>> getMessageConverters() {
+			return messageConverters;
+		}
+
+		public void setMessageConverters(List<HttpMessageConverter<?>> messageConverters) {
+			this.messageConverters = messageConverters;
+		}
+
+		@Override
+		public ResponseErrorHandler getResponseErrorHandler() {
+			return responseErrorHandler;
+		}
+
+		public void setResponseErrorHandler(ResponseErrorHandler responseErrorHandler) {
+			this.responseErrorHandler = responseErrorHandler;
+		}
+	}
+
 	protected class DefaultRequestCallback implements Function<ClientHttpRequest, Mono<Void>> {
 
 		private final ClientWebRequest clientWebRequest;
@@ -198,7 +232,8 @@ public final class WebClient {
 					.forEach(cookie -> clientHttpRequest.getCookies().add(cookie.getName(), cookie));
 			if (this.clientWebRequest.getBody() != null) {
 				return writeRequestBody(this.clientWebRequest.getBody(),
-						this.clientWebRequest.getElementType(), clientHttpRequest, messageConverters);
+						this.clientWebRequest.getElementType(),
+						clientHttpRequest, webClientConfig.getMessageConverters());
 			}
 			else {
 				return clientHttpRequest.setComplete();

spring-web/src/main/java/org/springframework/web/client/reactive/WebClientConfig.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2002-2016 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.client.reactive;
+
+import java.util.List;
+
+import org.springframework.http.converter.reactive.HttpMessageConverter;
+
+/**
+ * Interface that makes the {@link WebClient} configuration information
+ * available to downstream infrastructure such as {@link ResponseErrorHandler}s.
+ *
+ * @author Brian Clozel
+ * @since 5.0
+ */
+public interface WebClientConfig {
+
+	/**
+	 * Return the message converters that can help encoding/decoding the HTTP message body
+	 */
+	List<HttpMessageConverter<?>> getMessageConverters();
+
+	/**
+	 * Return the configured {@link ResponseErrorHandler}
+	 */
+	ResponseErrorHandler getResponseErrorHandler();
+}

spring-web/src/main/java/org/springframework/web/client/reactive/WebClientErrorException.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2002-2016 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.client.reactive;
+
+import java.util.List;
+
+import org.springframework.http.HttpStatus;
+import org.springframework.http.client.reactive.ClientHttpResponse;
+import org.springframework.http.converter.reactive.HttpMessageConverter;
+
+/**
+ * Exception thrown when an HTTP 4xx is received.
+ *
+ * @author Brian Clozel
+ * @since 5.0
+ */
+@SuppressWarnings("serial")
+public class WebClientErrorException extends WebClientResponseException {
+
+	/**
+	 * Construct a new instance of {@code HttpClientErrorException} based on a {@link ClientHttpResponse}
+	 * and {@link HttpMessageConverter}s to optionally help decoding the response body
+	 * @param clientResponse the HTTP response
+	 * @param messageConverters the message converters that may decode the HTTP response body
+	 */
+	public WebClientErrorException(ClientHttpResponse clientResponse,
+			List<HttpMessageConverter<?>> messageConverters) {
+		super(clientResponse.getStatusCode().value() + " " + clientResponse.getStatusCode().getReasonPhrase(),
+				clientResponse, messageConverters);
+	}
+}

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

@@ -20,7 +20,7 @@ import org.springframework.core.NestedRuntimeException;
 
 /**
  * Base class for exceptions thrown by {@link WebClient} whenever
- * it encounters client-side errors.
+ * it encounters errors.
  *
  * @author Brian Clozel
  * @since 5.0

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

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2002-2016 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.client.reactive;
+
+import java.util.List;
+
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.client.reactive.ClientHttpResponse;
+import org.springframework.http.converter.reactive.HttpMessageConverter;
+
+/**
+ * Base class for exceptions associated with specific HTTP client response status codes.
+ *
+ * @author Brian Clozel
+ * @since 5.0
+ */
+@SuppressWarnings("serial")
+public class WebClientResponseException extends WebClientException {
+
+	private final ClientHttpResponse clientResponse;
+
+	private final List<HttpMessageConverter<?>> messageConverters;
+
+	/**
+	 * Construct a new instance of {@code WebClientResponseException} with the given response data
+	 * @param message the given error message
+	 * @param clientResponse the HTTP response
+	 * @param messageConverters the message converters that maay decode the HTTP response body
+	 */
+	public WebClientResponseException(String message, ClientHttpResponse clientResponse,
+			List<HttpMessageConverter<?>> messageConverters) {
+		super(message);
+		this.clientResponse = clientResponse;
+		this.messageConverters = messageConverters;
+	}
+
+	/**
+	 * Return the HTTP status
+	 */
+	public HttpStatus getStatus() {
+		return this.clientResponse.getStatusCode();
+	}
+
+	/**
+	 * Return the HTTP response headers
+	 */
+	public HttpHeaders getResponseHeaders() {
+		return this.clientResponse.getHeaders();
+	}
+
+	/**
+	 * Perform an extraction of the response body into a higher level representation.
+	 *
+	 * <pre class="code">
+	 * static imports: ResponseExtractors.*
+	 *
+	 * String responseBody = clientResponse.getResponseBody(as(String.class));
+	 * </pre>
+	 */
+	public <T> T getResponseBody(BodyExtractor<T> extractor) {
+		return extractor.extract(this.clientResponse, this.messageConverters);
+	}
+}

spring-web/src/main/java/org/springframework/web/client/reactive/WebResponseActions.java

@@ -39,11 +39,11 @@ public interface WebResponseActions {
 	 * Perform an extraction of the response body into a higher level representation.
 	 *
 	 * <pre class="code">
-	 * static imports: HttpRequestBuilders.*, HttpResponseExtractors.*
+	 * static imports: ClientWebRequestBuilder.*, ResponseExtractors.*
 	 *
 	 * webClient
-	 *   .perform(get(baseUrl.toString()).accept(MediaType.TEXT_PLAIN))
-	 *   .extract(response(String.class));
+	 *   .perform(get(url).accept(MediaType.TEXT_PLAIN))
+	 *   .extract(body(String.class));
 	 * </pre>
 	 */
 	<T> T extract(ResponseExtractor<T> extractor);

spring-web/src/main/java/org/springframework/web/client/reactive/WebServerErrorException.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2002-2016 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.client.reactive;
+
+import java.util.List;
+
+import org.springframework.http.HttpStatus;
+import org.springframework.http.client.reactive.ClientHttpResponse;
+import org.springframework.http.converter.reactive.HttpMessageConverter;
+
+/**
+ * Exception thrown when an HTTP 5xx is received.
+ *
+ * @author Brian Clozel
+ * @since 5.0
+ */
+@SuppressWarnings("serial")
+public class WebServerErrorException extends WebClientResponseException {
+
+	/**
+	 * Construct a new instance of {@code HttpServerErrorException} based on a {@link ClientHttpResponse}
+	 * and {@link HttpMessageConverter}s to optionally help decoding the response body
+	 * @param clientResponse the HTTP response
+	 * @param messageConverters the message converters that may decode the HTTP response body
+	 */
+	public WebServerErrorException(ClientHttpResponse clientResponse, List<HttpMessageConverter<?>> messageConverters) {
+		super(clientResponse.getStatusCode().value() + " " + clientResponse.getStatusCode().getReasonPhrase(),
+				clientResponse, messageConverters);
+	}
+}

spring-web/src/main/java/org/springframework/web/client/reactive/support/RxJava1ResponseExtractors.java

@@ -17,7 +17,6 @@
 package org.springframework.web.client.reactive.support;
 
 import java.util.List;
-import java.util.Optional;
 
 import reactor.adapter.RxJava1Adapter;
 import reactor.core.publisher.Flux;
@@ -31,10 +30,12 @@ import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.http.client.reactive.ClientHttpResponse;
 import org.springframework.http.converter.reactive.HttpMessageConverter;
+import org.springframework.web.client.reactive.BodyExtractor;
 import org.springframework.web.client.reactive.ResponseExtractor;
+import org.springframework.web.client.reactive.WebClientException;
 
 /**
- * Static factory methods for {@link ResponseExtractor}
+ * Static factory methods for {@link ResponseExtractor} and {@link BodyExtractor},
  * based on the {@link Observable} and {@link Single} APIs.
  *
  * @author Brian Clozel
@@ -42,16 +43,60 @@ import org.springframework.web.client.reactive.ResponseExtractor;
  */
 public class RxJava1ResponseExtractors {
 
+	/**
+	 * Extract the response body and decode it, returning it as a {@code Single<T>}.
+	 * @see ResolvableType#forClassWithGenerics(Class, Class[])
+	 */
+	@SuppressWarnings("unchecked")
+	public static <T> ResponseExtractor<Single<T>> body(ResolvableType bodyType) {
+
+		return (clientResponse, webClientConfig) -> (Single<T>) RxJava1Adapter
+				.publisherToSingle(clientResponse
+						.doOnNext(response -> webClientConfig.getResponseErrorHandler()
+								.handleError(response, webClientConfig.getMessageConverters()))
+						.flatMap(resp -> decodeResponseBodyAsMono(resp, bodyType, webClientConfig.getMessageConverters())));
+	}
+
 	/**
 	 * Extract the response body and decode it, returning it as a {@code Single<T>}.
 	 */
 	@SuppressWarnings("unchecked")
 	public static <T> ResponseExtractor<Single<T>> body(Class<T> sourceClass) {
 
-		ResolvableType resolvableType = ResolvableType.forClass(sourceClass);
-		return (clientResponse, messageConverters) -> (Single<T>) RxJava1Adapter
-				.publisherToSingle(clientResponse
-						.flatMap(resp -> decodeResponseBody(resp, resolvableType, messageConverters)).next());
+		ResolvableType bodyType = ResolvableType.forClass(sourceClass);
+		return body(bodyType);
+	}
+
+	/**
+	 * Extract the response body and decode it, returning it as a {@code Single<T>}.
+	 * @see ResolvableType#forClassWithGenerics(Class, Class[])
+	 */
+	@SuppressWarnings("unchecked")
+	public static <T> BodyExtractor<Single<T>> as(ResolvableType bodyType) {
+		return (clientResponse, messageConverters) ->
+				(Single<T>) RxJava1Adapter.publisherToSingle(
+						decodeResponseBodyAsMono(clientResponse, bodyType, messageConverters));
+	}
+
+	/**
+	 * Extract the response body and decode it, returning it as a {@code Single<T>}
+	 */
+	public static <T> BodyExtractor<Single<T>> as(Class<T> sourceClass) {
+		ResolvableType bodyType = ResolvableType.forClass(sourceClass);
+		return as(bodyType);
+	}
+
+	/**
+	 * Extract the response body and decode it, returning it as an {@code Observable<T>}
+	 * @see ResolvableType#forClassWithGenerics(Class, Class[])
+	 */
+	public static <T> ResponseExtractor<Observable<T>> bodyStream(ResolvableType bodyType) {
+
+		return (clientResponse, webClientConfig) -> RxJava1Adapter
+				.publisherToObservable(clientResponse
+						.doOnNext(response -> webClientConfig.getResponseErrorHandler()
+								.handleError(response, webClientConfig.getMessageConverters()))
+						.flatMap(resp -> decodeResponseBody(resp, bodyType, webClientConfig.getMessageConverters())));
 	}
 
 	/**
@@ -59,29 +104,56 @@ public class RxJava1ResponseExtractors {
 	 */
 	public static <T> ResponseExtractor<Observable<T>> bodyStream(Class<T> sourceClass) {
 
-		ResolvableType resolvableType = ResolvableType.forClass(sourceClass);
-		return (clientResponse, messageConverters) -> RxJava1Adapter
-				.publisherToObservable(clientResponse
-						.flatMap(resp -> decodeResponseBody(resp, resolvableType, messageConverters)));
+		ResolvableType bodyType = ResolvableType.forClass(sourceClass);
+		return bodyStream(bodyType);
+	}
+
+	/**
+	 * Extract the response body and decode it, returning it as a {@code Observable<T>}.
+	 * @see ResolvableType#forClassWithGenerics(Class, Class[])
+	 */
+	@SuppressWarnings("unchecked")
+	public static <T> BodyExtractor<Observable<T>> asStream(ResolvableType bodyType) {
+		return (clientResponse, messageConverters) ->
+				(Observable<T>) RxJava1Adapter
+						.publisherToObservable(decodeResponseBody(clientResponse, bodyType, messageConverters));
+	}
+
+	/**
+	 * Extract the response body and decode it, returning it as a {@code Observable<T>}.
+	 */
+	public static <T> BodyExtractor<Observable<T>> asStream(Class<T> sourceClass) {
+		ResolvableType bodyType = ResolvableType.forClass(sourceClass);
+		return asStream(bodyType);
+	}
+
+	/**
+	 * Extract the full response body as a {@code ResponseEntity}
+	 * with its body decoded as a single type {@code T}.
+	 * @see ResolvableType#forClassWithGenerics(Class, Class[])
+	 */
+	@SuppressWarnings("unchecked")
+	public static <T> ResponseExtractor<Single<ResponseEntity<T>>> response(ResolvableType bodyType) {
+
+		return (clientResponse, webClientConfig) ->
+				RxJava1Adapter.publisherToSingle(clientResponse
+						.then(response ->
+								Mono.when(
+										decodeResponseBody(response, bodyType, webClientConfig.getMessageConverters()).next(),
+										Mono.just(response.getHeaders()),
+										Mono.just(response.getStatusCode())))
+						.map(tuple ->
+							new ResponseEntity<>((T) tuple.getT1(), tuple.getT2(), tuple.getT3())));
 	}
 
 	/**
 	 * Extract the full response body as a {@code ResponseEntity}
 	 * with its body decoded as a single type {@code T}.
 	 */
-	@SuppressWarnings("unchecked")
 	public static <T> ResponseExtractor<Single<ResponseEntity<T>>> response(Class<T> sourceClass) {
 
-		ResolvableType resolvableType = ResolvableType.forClass(sourceClass);
-		return (clientResponse, messageConverters) ->
-				RxJava1Adapter.publisherToSingle(clientResponse
-						.then(response ->
-								Mono.when(
-										decodeResponseBody(response, resolvableType, messageConverters).next(),
-										Mono.just(response.getHeaders()),
-										Mono.just(response.getStatusCode())))
-						.map(tuple ->
-								new ResponseEntity<>((T) tuple.getT1(), tuple.getT2(), tuple.getT3())));
+		ResolvableType bodyType = ResolvableType.forClass(sourceClass);
+		return response(bodyType);
 	}
 
 	/**
@@ -90,10 +162,19 @@ public class RxJava1ResponseExtractors {
 	 */
 	public static <T> ResponseExtractor<Single<ResponseEntity<Observable<T>>>> responseStream(Class<T> sourceClass) {
 		ResolvableType resolvableType = ResolvableType.forClass(sourceClass);
-		return (clientResponse, messageConverters) -> RxJava1Adapter.publisherToSingle(
-				clientResponse.map(response -> new ResponseEntity<>(
-						RxJava1Adapter.publisherToObservable(
-								RxJava1ResponseExtractors.<T> decodeResponseBody(response, resolvableType, messageConverters)),
+		return responseStream(resolvableType);
+	}
+
+	/**
+	 * Extract the full response body as a {@code ResponseEntity}
+	 * with its body decoded as an {@code Observable<T>}
+	 * @see ResolvableType#forClassWithGenerics(Class, Class[])
+	 */
+	public static <T> ResponseExtractor<Single<ResponseEntity<Observable<T>>>> responseStream(ResolvableType bodyType) {
+		return (clientResponse, webClientConfig) -> RxJava1Adapter.publisherToSingle(clientResponse
+				.map(response -> new ResponseEntity<>(
+								RxJava1Adapter
+								.publisherToObservable(decodeResponseBody(response, bodyType, webClientConfig.getMessageConverters())),
 						response.getHeaders(),
 						response.getStatusCode())));
 	}
@@ -107,22 +188,33 @@ public class RxJava1ResponseExtractors {
 	}
 
 	@SuppressWarnings("unchecked")
-	protected static <T> Flux<T> decodeResponseBody(ClientHttpResponse response, ResolvableType responseType,
-			List<HttpMessageConverter<?>> messageConverters) {
+	protected static <T> Flux<T> decodeResponseBody(ClientHttpResponse response,
+			ResolvableType responseType, List<HttpMessageConverter<?>> messageConverters) {
 
 		MediaType contentType = response.getHeaders().getContentType();
-		Optional<HttpMessageConverter<?>> converter = resolveConverter(messageConverters, responseType, contentType);
-		if (!converter.isPresent()) {
-			return Flux.error(new IllegalStateException("Could not decode response body of type '" + contentType +
-					"' with target type '" + responseType.toString() + "'"));
-		}
-		return (Flux<T>) converter.get().read(responseType, response);
+		HttpMessageConverter<?> converter = resolveConverter(messageConverters, responseType, contentType);
+		return (Flux<T>) converter.read(responseType, response);
 	}
 
+	@SuppressWarnings("unchecked")
+	protected static <T> Mono<T> decodeResponseBodyAsMono(ClientHttpResponse response,
+			ResolvableType responseType, List<HttpMessageConverter<?>> messageConverters) {
 
-	protected static Optional<HttpMessageConverter<?>> resolveConverter(List<HttpMessageConverter<?>> messageConverters,
-			ResolvableType type, MediaType mediaType) {
-		return messageConverters.stream().filter(e -> e.canRead(type, mediaType)).findFirst();
+		MediaType contentType = response.getHeaders().getContentType();
+		HttpMessageConverter<?> converter = resolveConverter(messageConverters, responseType, contentType);
+		return (Mono<T>) converter.readMono(responseType, response);
+	}
+
+	protected static HttpMessageConverter<?> resolveConverter(
+			List<HttpMessageConverter<?>> messageConverters, ResolvableType responseType, MediaType contentType) {
+
+		return messageConverters.stream()
+				.filter(e -> e.canRead(responseType, contentType))
+				.findFirst()
+				.orElseThrow(() ->
+						new WebClientException(
+								"Could not decode response body of type '" + contentType
+										+ "' with target type '" + responseType.toString() + "'"));
 	}
 
 }