Simplify access to response body in WebClient

This commit makes a change to WebClient in oder to facilitate getting
the response body as a `Mono<Object>` or `Flux<Object>` without having
to deal with `ClientResponse`.

Specifically, this commit:

 - Adds `RequestHeaderSpec.retrieve` methods, next to `exchange`, that
 return the response body (and not a `ClientResponse`). Two convenience
 methods return the response body as `Mono` or `Flux`.
 - Adds ClientResponse.toRequestEntity to convert the ClientResponse
 into a RequestEntity.

Issue: SPR-15294

Author: poutsma

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

@@ -27,6 +27,7 @@
 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.util.MultiValueMap;
 import org.springframework.web.reactive.function.BodyExtractor;
@@ -88,6 +89,14 @@ public interface ClientResponse {
 	 */
 	<T> Flux<T> bodyToFlux(Class<? extends T> elementClass);
 
+	/**
+	 * Converts this {@code ClientResponse} into a {@code ResponseEntity}.
+	 * @param responseClass the type of response contained in the {@code ResponseEntity}
+	 * @param <T> the response type
+	 * @return a mono containing the response entity
+	 */
+	<T> Mono<ResponseEntity<T>> toResponseEntity(Class<T> responseClass);
+
 
 	/**
 	 * Represents the headers of the HTTP response.

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

@@ -33,6 +33,7 @@
 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.util.MultiValueMap;
@@ -100,6 +101,11 @@ public <T> Flux<T> bodyToFlux(Class<? extends T> elementClass) {
 		return bodyToPublisher(BodyExtractors.toFlux(elementClass), Flux::error);
 	}
 
+	@Override
+	public <T> Mono<ResponseEntity<T>> toResponseEntity(Class<T> responseClass) {
+		return bodyToMono(responseClass)
+				.map(t -> new ResponseEntity<>(t, headers().asHttpHeaders(), statusCode()));
+	}
 
 	private <T extends Publisher<?>> T bodyToPublisher(
 			BodyExtractor<T, ? super ClientHttpResponse> extractor,

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

@@ -26,15 +26,18 @@
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.MediaType;
 import org.springframework.http.client.reactive.ClientHttpRequest;
+import org.springframework.http.client.reactive.ClientHttpResponse;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.LinkedMultiValueMap;
 import org.springframework.util.MultiValueMap;
+import org.springframework.web.reactive.function.BodyExtractor;
 import org.springframework.web.reactive.function.BodyInserter;
 import org.springframework.web.reactive.function.BodyInserters;
 import org.springframework.web.util.DefaultUriBuilderFactory;
@@ -325,6 +328,21 @@ else if (CollectionUtils.isEmpty(this.cookies)) {
 				return result;
 			}
 		}
+
+		@Override
+		public <T> Mono<T> retrieve(BodyExtractor<T, ? super ClientHttpResponse> extractor) {
+			return exchange().map(clientResponse -> clientResponse.body(extractor));
+		}
+
+		@Override
+		public <T> Mono<T> retrieveMono(Class<T> responseType) {
+			return exchange().then(clientResponse -> clientResponse.bodyToMono(responseType));
+		}
+
+		@Override
+		public <T> Flux<T> retrieveFlux(Class<T> responseType) {
+			return exchange().flatMap(clientResponse -> clientResponse.bodyToFlux(responseType));
+		}
 	}
 
 }

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

@@ -23,14 +23,17 @@
 import java.util.function.Function;
 
 import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.MediaType;
 import org.springframework.http.client.reactive.ClientHttpConnector;
 import org.springframework.http.client.reactive.ClientHttpRequest;
+import org.springframework.http.client.reactive.ClientHttpResponse;
 import org.springframework.util.MultiValueMap;
+import org.springframework.web.reactive.function.BodyExtractor;
 import org.springframework.web.reactive.function.BodyInserter;
 import org.springframework.web.util.UriBuilder;
 import org.springframework.web.util.UriBuilderFactory;
@@ -374,6 +377,38 @@ interface RequestHeadersSpec<S extends RequestHeadersSpec<S>> {
 		 */
 		Mono<ClientResponse> exchange();
 
+		/**
+		 * Execute the built request, and use the given extractor to return the response body as a
+		 * delayed {@code T}.
+		 * @param extractor the extractor for the response body
+		 * @param <T> the response type
+		 * @return the body of the response, extracted with {@code extractor}
+		 */
+		<T> Mono<T> retrieve(BodyExtractor<T, ? super ClientHttpResponse> extractor);
+
+		/**
+		 * Execute the built request, and return the response body as a delayed {@code T}.
+		 * <p>This method is a convenient shortcut for {@link #retrieve(BodyExtractor)} with a
+		 * {@linkplain org.springframework.web.reactive.function.BodyExtractors#toMono(Class)
+		 * Mono body extractor}.
+		 * @param responseType the class of the response
+		 * @param <T> the response type
+		 * @return the body of the response
+		 */
+		<T> Mono<T> retrieveMono(Class<T> responseType);
+
+		/**
+		 * Execute the built request, and return the response body as a delayed sequence of
+		 * {@code T}'s.
+		 * <p>This method is a convenient shortcut for {@link #retrieve(BodyExtractor)} with a
+		 * {@linkplain org.springframework.web.reactive.function.BodyExtractors#toFlux(Class)}
+		 * Flux body extractor}.
+		 * @param responseType the class of the response
+		 * @param <T> the response type
+		 * @return the body of the response
+		 */
+		<T> Flux<T> retrieveFlux(Class<T> responseType);
+
 	}
 
 	interface RequestBodySpec extends RequestHeadersSpec<RequestBodySpec> {

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

@@ -133,6 +133,50 @@ public void jsonString() throws Exception {
 		Assert.assertEquals("application/json", recordedRequest.getHeader(HttpHeaders.ACCEPT));
 	}
 
+	@Test
+	public void jsonStringRetrieveMono() throws Exception {
+		String content = "{\"bar\":\"barbar\",\"foo\":\"foofoo\"}";
+		this.server.enqueue(new MockResponse().setHeader("Content-Type", "application/json")
+				.setBody(content));
+
+		Mono<String> result = this.webClient.get()
+				.uri("/json")
+				.accept(MediaType.APPLICATION_JSON)
+				.retrieveMono(String.class);
+
+		StepVerifier.create(result)
+				.expectNext(content)
+				.expectComplete()
+				.verify(Duration.ofSeconds(3));
+
+		RecordedRequest recordedRequest = server.takeRequest();
+		Assert.assertEquals(1, server.getRequestCount());
+		Assert.assertEquals("/json", recordedRequest.getPath());
+		Assert.assertEquals("application/json", recordedRequest.getHeader(HttpHeaders.ACCEPT));
+	}
+
+	@Test
+	public void jsonStringRetrieveFlux() throws Exception {
+		String content = "{\"bar\":\"barbar\",\"foo\":\"foofoo\"}";
+		this.server.enqueue(new MockResponse().setHeader("Content-Type", "application/json")
+				.setBody(content));
+
+		Flux<String> result = this.webClient.get()
+				.uri("/json")
+				.accept(MediaType.APPLICATION_JSON)
+				.retrieveFlux(String.class);
+
+		StepVerifier.create(result)
+				.expectNext(content)
+				.expectComplete()
+				.verify(Duration.ofSeconds(3));
+
+		RecordedRequest recordedRequest = server.takeRequest();
+		Assert.assertEquals(1, server.getRequestCount());
+		Assert.assertEquals("/json", recordedRequest.getPath());
+		Assert.assertEquals("application/json", recordedRequest.getHeader(HttpHeaders.ACCEPT));
+	}
+
 	@Test
 	public void jsonPojoMono() throws Exception {
 		this.server.enqueue(new MockResponse().setHeader("Content-Type", "application/json")