Add ParameterizedTypeReference variants to bodyTo[Mono|Flux], toEntity[List]

This commit introduces overloaded variants of `bodytoMono`,
`bodyToFlux`, `toEntity`, and `toEntityList` that take a
`ParameterizedTypeReference`. It also adds similar methods to
`WebClient.ResponseSpec`.

Issue: SPR-15725

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

@@ -23,6 +23,7 @@ import java.util.OptionalLong;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
@@ -75,6 +76,14 @@ public interface ClientResponse {
 	 */
 	<T> Mono<T> bodyToMono(Class<? extends T> elementClass);
 
+	/**
+	 * Extract the body to a {@code Mono}.
+	 * @param typeReference a type reference describing the expected response body type
+	 * @param <T> the element type
+	 * @return a mono containing the body of the given type {@code T}
+	 */
+	<T> Mono<T> bodyToMono(ParameterizedTypeReference<T> typeReference);
+
 	/**
 	 * Extract the body to a {@code Flux}.
 	 * @param elementClass the class of element in the {@code Flux}
@@ -83,6 +92,14 @@ public interface ClientResponse {
 	 */
 	<T> Flux<T> bodyToFlux(Class<? extends T> elementClass);
 
+	/**
+	 * Extract the body to a {@code Flux}.
+	 * @param typeReference a type reference describing the expected response body type
+	 * @param <T> the element type
+	 * @return a flux containing the body of the given type {@code T}
+	 */
+	<T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> typeReference);
+
 	/**
 	 * Return this response as a delayed {@code ResponseEntity}.
 	 * @param bodyType the expected response body type
@@ -91,6 +108,14 @@ public interface ClientResponse {
 	 */
 	<T> Mono<ResponseEntity<T>> toEntity(Class<T> bodyType);
 
+	/**
+	 * Return this response as a delayed {@code ResponseEntity}.
+	 * @param typeReference a type reference describing the expected response body type
+	 * @param <T> response body type
+	 * @return {@code Mono} with the {@code ResponseEntity}
+	 */
+	<T> Mono<ResponseEntity<T>> toEntity(ParameterizedTypeReference<T> typeReference);
+
 	/**
 	 * Return this response as a delayed list of {@code ResponseEntity}s.
 	 * @param elementType the expected response body list element type
@@ -99,6 +124,14 @@ public interface ClientResponse {
 	 */
 	<T> Mono<ResponseEntity<List<T>>> toEntityList(Class<T> elementType);
 
+	/**
+	 * Return this response as a delayed list of {@code ResponseEntity}s.
+	 * @param typeReference a type reference describing the expected response body type
+	 * @param <T> the type of elements in the list
+	 * @return {@code Mono} with the list of {@code ResponseEntity}s
+	 */
+	<T> Mono<ResponseEntity<List<T>>> toEntityList(ParameterizedTypeReference<T> typeReference);
+
 
 	/**
 	 * Represents the headers of the HTTP response.

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

@@ -25,6 +25,7 @@ import java.util.OptionalLong;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
@@ -99,16 +100,35 @@ class DefaultClientResponse implements ClientResponse {
 		return body(BodyExtractors.toMono(elementClass));
 	}
 
+	@Override
+	public <T> Mono<T> bodyToMono(ParameterizedTypeReference<T> typeReference) {
+		return body(BodyExtractors.toMono(typeReference));
+	}
+
 	@Override
 	public <T> Flux<T> bodyToFlux(Class<? extends T> elementClass) {
 		return body(BodyExtractors.toFlux(elementClass));
 	}
 
+	@Override
+	public <T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> typeReference) {
+		return body(BodyExtractors.toFlux(typeReference));
+	}
+
 	@Override
 	public <T> Mono<ResponseEntity<T>> toEntity(Class<T> bodyType) {
+		return toEntityInternal(bodyToMono(bodyType));
+	}
+
+	@Override
+	public <T> Mono<ResponseEntity<T>> toEntity(ParameterizedTypeReference<T> typeReference) {
+		return toEntityInternal(bodyToMono(typeReference));
+	}
+
+	private <T> Mono<ResponseEntity<T>> toEntityInternal(Mono<T> bodyMono) {
 		HttpHeaders headers = headers().asHttpHeaders();
 		HttpStatus statusCode = statusCode();
-		return bodyToMono(bodyType)
+		return bodyMono
 				.map(body -> new ResponseEntity<>(body, headers, statusCode))
 				.switchIfEmpty(Mono.defer(
 						() -> Mono.just(new ResponseEntity<>(headers, statusCode))));
@@ -116,9 +136,19 @@ class DefaultClientResponse implements ClientResponse {
 
 	@Override
 	public <T> Mono<ResponseEntity<List<T>>> toEntityList(Class<T> responseType) {
+		return toEntityListInternal(bodyToFlux(responseType));
+	}
+
+	@Override
+	public <T> Mono<ResponseEntity<List<T>>> toEntityList(
+			ParameterizedTypeReference<T> typeReference) {
+		return toEntityListInternal(bodyToFlux(typeReference));
+	}
+
+	private <T> Mono<ResponseEntity<List<T>>> toEntityListInternal(Flux<T> bodyFlux) {
 		HttpHeaders headers = headers().asHttpHeaders();
 		HttpStatus statusCode = statusCode();
-		return bodyToFlux(responseType)
+		return bodyFlux
 				.collectList()
 				.map(body -> new ResponseEntity<>(body, headers, statusCode));
 	}

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

@@ -35,6 +35,7 @@ import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
@@ -422,6 +423,13 @@ class DefaultWebClient implements WebClient {
 							Mono::error));
 		}
 
+		@Override
+		public <T> Mono<T> bodyToMono(ParameterizedTypeReference<T> typeReference) {
+			return this.responseMono.flatMap(
+					response -> bodyToPublisher(response, BodyExtractors.toMono(typeReference),
+							Mono::error));
+		}
+
 		@Override
 		public <T> Flux<T> bodyToFlux(Class<T> elementType) {
 			return this.responseMono.flatMapMany(
@@ -429,6 +437,13 @@ class DefaultWebClient implements WebClient {
 							Flux::error));
 		}
 
+		@Override
+		public <T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> typeReference) {
+			return this.responseMono.flatMapMany(
+					response -> bodyToPublisher(response, BodyExtractors.toFlux(typeReference),
+							Flux::error));
+		}
+
 		private <T extends Publisher<?>> T bodyToPublisher(ClientResponse response,
 				BodyExtractor<T, ? super ClientHttpResponse> extractor,
 				Function<Throwable, T> errorFunction) {

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

@@ -29,6 +29,7 @@ import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpMethod;
 import org.springframework.http.HttpStatus;
@@ -559,8 +560,9 @@ public interface WebClient {
 				Function<ClientResponse, ? extends Throwable> exceptionFunction);
 
 		/**
-		 * Extract the body to a {@code Mono}. If the response has status code 4xx or 5xx, the
-		 * {@code Mono} will contain a {@link WebClientException}.
+		 * Extract the body to a {@code Mono}. By default, if the response has status code 4xx or
+		 * 5xx, the {@code Mono} will contain a {@link WebClientException}. This can be overridden
+		 * 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
@@ -569,8 +571,20 @@ public interface WebClient {
 		<T> Mono<T> bodyToMono(Class<T> bodyType);
 
 		/**
-		 * Extract the body to a {@code Flux}. If the response has status code 4xx or 5xx, the
-		 * {@code Flux} will contain a {@link WebClientException}.
+		 * Extract the body to a {@code Mono}. By default, if the response has status code 4xx or
+		 * 5xx, the {@code Mono} will contain a {@link WebClientException}. This can be overridden
+		 * 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
+		 * 4xx or 5xx
+		 */
+		<T> Mono<T> bodyToMono(ParameterizedTypeReference<T> typeReference);
+
+		/**
+		 * Extract the body to a {@code Flux}. By default, if the response has status code 4xx or
+		 * 5xx, the {@code Flux} will contain a {@link WebClientException}. This can be overridden
+         * 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
@@ -578,6 +592,17 @@ public interface WebClient {
 		 */
 		<T> Flux<T> bodyToFlux(Class<T> elementType);
 
+		/**
+		 * Extract the body to a {@code Flux}. By default, if the response has status code 4xx or
+		 * 5xx, the {@code Flux} will contain a {@link WebClientException}. This can be overridden
+         * 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
+		 * 4xx or 5xx
+		 */
+		<T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> typeReference);
+
 	}
 
 

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

@@ -29,6 +29,7 @@ import org.junit.Test;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.core.codec.StringDecoder;
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DefaultDataBuffer;
@@ -38,6 +39,7 @@ import org.springframework.http.HttpRange;
 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.DecoderHttpMessageReader;
 import org.springframework.http.codec.HttpMessageReader;
@@ -149,6 +151,29 @@ public class DefaultClientResponseTests {
 		assertEquals("foo", resultMono.block());
 	}
 
+	@Test
+	public void bodyToMonoTypeReference() throws Exception {
+		DefaultDataBufferFactory factory = new DefaultDataBufferFactory();
+		DefaultDataBuffer dataBuffer =
+				factory.wrap(ByteBuffer.wrap("foo".getBytes(StandardCharsets.UTF_8)));
+		Flux<DataBuffer> body = Flux.just(dataBuffer);
+
+		HttpHeaders httpHeaders = new HttpHeaders();
+		httpHeaders.setContentType(MediaType.TEXT_PLAIN);
+		when(mockResponse.getHeaders()).thenReturn(httpHeaders);
+		when(mockResponse.getStatusCode()).thenReturn(HttpStatus.OK);
+		when(mockResponse.getBody()).thenReturn(body);
+
+		List<HttpMessageReader<?>> messageReaders = Collections
+				.singletonList(new DecoderHttpMessageReader<>(StringDecoder.allMimeTypes(true)));
+		when(mockExchangeStrategies.messageReaders()).thenReturn(messageReaders);
+
+		Mono<String> resultMono =
+				defaultClientResponse.bodyToMono(new ParameterizedTypeReference<String>() {
+				});
+		assertEquals("foo", resultMono.block());
+	}
+
 	@Test
 	public void bodyToFlux() throws Exception {
 		DefaultDataBufferFactory factory = new DefaultDataBufferFactory();
@@ -171,4 +196,124 @@ public class DefaultClientResponseTests {
 		assertEquals(Collections.singletonList("foo"), result.block());
 	}
 
+	@Test
+	public void bodyToFluxTypeReference() throws Exception {
+		DefaultDataBufferFactory factory = new DefaultDataBufferFactory();
+		DefaultDataBuffer dataBuffer =
+				factory.wrap(ByteBuffer.wrap("foo".getBytes(StandardCharsets.UTF_8)));
+		Flux<DataBuffer> body = Flux.just(dataBuffer);
+
+		HttpHeaders httpHeaders = new HttpHeaders();
+		httpHeaders.setContentType(MediaType.TEXT_PLAIN);
+		when(mockResponse.getHeaders()).thenReturn(httpHeaders);
+		when(mockResponse.getStatusCode()).thenReturn(HttpStatus.OK);
+		when(mockResponse.getBody()).thenReturn(body);
+
+		List<HttpMessageReader<?>> messageReaders = Collections
+				.singletonList(new DecoderHttpMessageReader<>(StringDecoder.allMimeTypes(true)));
+		when(mockExchangeStrategies.messageReaders()).thenReturn(messageReaders);
+
+		Flux<String> resultFlux =
+				defaultClientResponse.bodyToFlux(new ParameterizedTypeReference<String>() {});
+		Mono<List<String>> result = resultFlux.collectList();
+		assertEquals(Collections.singletonList("foo"), result.block());
+	}
+
+	@Test
+	public void toEntity() throws Exception {
+		DefaultDataBufferFactory factory = new DefaultDataBufferFactory();
+		DefaultDataBuffer dataBuffer =
+				factory.wrap(ByteBuffer.wrap("foo".getBytes(StandardCharsets.UTF_8)));
+		Flux<DataBuffer> body = Flux.just(dataBuffer);
+
+		HttpHeaders httpHeaders = new HttpHeaders();
+		httpHeaders.setContentType(MediaType.TEXT_PLAIN);
+		when(mockResponse.getHeaders()).thenReturn(httpHeaders);
+		when(mockResponse.getStatusCode()).thenReturn(HttpStatus.OK);
+		when(mockResponse.getBody()).thenReturn(body);
+
+		List<HttpMessageReader<?>> messageReaders = Collections
+				.singletonList(new DecoderHttpMessageReader<>(StringDecoder.allMimeTypes(true)));
+		when(mockExchangeStrategies.messageReaders()).thenReturn(messageReaders);
+
+		ResponseEntity<String> result = defaultClientResponse.toEntity(String.class).block();
+		assertEquals("foo", result.getBody());
+		assertEquals(HttpStatus.OK, result.getStatusCode());
+		assertEquals(MediaType.TEXT_PLAIN, result.getHeaders().getContentType());
+	}
+
+	@Test
+	public void toEntityTypeReference() throws Exception {
+		DefaultDataBufferFactory factory = new DefaultDataBufferFactory();
+		DefaultDataBuffer dataBuffer =
+				factory.wrap(ByteBuffer.wrap("foo".getBytes(StandardCharsets.UTF_8)));
+		Flux<DataBuffer> body = Flux.just(dataBuffer);
+
+		HttpHeaders httpHeaders = new HttpHeaders();
+		httpHeaders.setContentType(MediaType.TEXT_PLAIN);
+		when(mockResponse.getHeaders()).thenReturn(httpHeaders);
+		when(mockResponse.getStatusCode()).thenReturn(HttpStatus.OK);
+		when(mockResponse.getBody()).thenReturn(body);
+
+		List<HttpMessageReader<?>> messageReaders = Collections
+				.singletonList(new DecoderHttpMessageReader<>(StringDecoder.allMimeTypes(true)));
+		when(mockExchangeStrategies.messageReaders()).thenReturn(messageReaders);
+
+		ResponseEntity<String> result = defaultClientResponse.toEntity(
+				new ParameterizedTypeReference<String>() {}).block();
+		assertEquals("foo", result.getBody());
+		assertEquals(HttpStatus.OK, result.getStatusCode());
+		assertEquals(MediaType.TEXT_PLAIN, result.getHeaders().getContentType());
+	}
+
+	@Test
+	public void toEntityList() throws Exception {
+		DefaultDataBufferFactory factory = new DefaultDataBufferFactory();
+		DefaultDataBuffer dataBuffer =
+				factory.wrap(ByteBuffer.wrap("foo".getBytes(StandardCharsets.UTF_8)));
+		Flux<DataBuffer> body = Flux.just(dataBuffer);
+
+		HttpHeaders httpHeaders = new HttpHeaders();
+		httpHeaders.setContentType(MediaType.TEXT_PLAIN);
+		when(mockResponse.getHeaders()).thenReturn(httpHeaders);
+		when(mockResponse.getStatusCode()).thenReturn(HttpStatus.OK);
+		when(mockResponse.getBody()).thenReturn(body);
+
+		List<HttpMessageReader<?>> messageReaders = Collections
+				.singletonList(new DecoderHttpMessageReader<>(StringDecoder.allMimeTypes(true)));
+		when(mockExchangeStrategies.messageReaders()).thenReturn(messageReaders);
+
+		ResponseEntity<List<String>> result = defaultClientResponse.toEntityList(String.class).block();
+		assertEquals(Collections.singletonList("foo"), result.getBody());
+		assertEquals(HttpStatus.OK, result.getStatusCode());
+		assertEquals(MediaType.TEXT_PLAIN, result.getHeaders().getContentType());
+	}
+
+	@Test
+	public void toEntityListTypeReference() throws Exception {
+		DefaultDataBufferFactory factory = new DefaultDataBufferFactory();
+		DefaultDataBuffer dataBuffer =
+				factory.wrap(ByteBuffer.wrap("foo".getBytes(StandardCharsets.UTF_8)));
+		Flux<DataBuffer> body = Flux.just(dataBuffer);
+
+		HttpHeaders httpHeaders = new HttpHeaders();
+		httpHeaders.setContentType(MediaType.TEXT_PLAIN);
+		when(mockResponse.getHeaders()).thenReturn(httpHeaders);
+		when(mockResponse.getStatusCode()).thenReturn(HttpStatus.OK);
+		when(mockResponse.getBody()).thenReturn(body);
+
+		List<HttpMessageReader<?>> messageReaders = Collections
+				.singletonList(new DecoderHttpMessageReader<>(StringDecoder.allMimeTypes(true)));
+		when(mockExchangeStrategies.messageReaders()).thenReturn(messageReaders);
+
+		ResponseEntity<List<String>> result = defaultClientResponse.toEntityList(
+				new ParameterizedTypeReference<String>() {}).block();
+		assertEquals(Collections.singletonList("foo"), result.getBody());
+		assertEquals(HttpStatus.OK, result.getStatusCode());
+		assertEquals(MediaType.TEXT_PLAIN, result.getHeaders().getContentType());
+	}
+
+
+
+
 }

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

@@ -32,6 +32,7 @@ import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
+import org.springframework.core.ParameterizedTypeReference;
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
@@ -157,6 +158,29 @@ public class WebClientIntegrationTests {
 		Assert.assertEquals("application/json", recordedRequest.getHeader(HttpHeaders.ACCEPT));
 	}
 
+	@Test
+	public void jsonStringRetrieveMonoTypeReference() 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)
+				.retrieve()
+				.bodyToMono(new ParameterizedTypeReference<String>() {});
+
+		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 jsonStringExchangeEntity() throws Exception {
 		String content = "{\"bar\":\"barbar\",\"foo\":\"foofoo\"}";
@@ -237,6 +261,29 @@ public class WebClientIntegrationTests {
 		Assert.assertEquals("application/json", recordedRequest.getHeader(HttpHeaders.ACCEPT));
 	}
 
+	@Test
+	public void jsonStringRetrieveFluxTypeReference() 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)
+				.retrieve()
+				.bodyToFlux(new ParameterizedTypeReference<String>() {});
+
+		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")