Revisit meaning of response.close() in HTTP client

Prior to this issue, SPR-15920 added this new `close()` method which was
supposed to be called to clean resources after response processing.

This commit changes the meaning of that method: calling `close()` will
close the underlying HTTP connection. This has to be called if the
response body is not consumed by the application, since at that point
the underlying connection might be in an inconsistent state if shared in
a connection pool.

Issue: SPR-15993

Author: bclozel

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

@@ -43,12 +43,13 @@ public interface ClientHttpResponse extends ReactiveHttpInputMessage, Closeable
 	MultiValueMap<String, ResponseCookie> getCookies();
 
 	/**
-	 * Close this response, freeing any resources created.
-	 * <p>This non-blocking method has to be called once the response has been
-	 * processed and the resources are no longer needed; not doing so might
-	 * create resource leaks or connection issues.
-	 * <p>Depending on the client configuration and HTTP version,
-	 * this can lead to closing the connection or returning it to a connection pool.
+	 * Close this response and the underlying HTTP connection.
+	 * <p>This non-blocking method has to be called if its body isn't going
+	 * to be consumed. Not doing so might result in HTTP connection pool
+	 * inconsistencies or memory leaks.
+	 * <p>This shouldn't be called if the response body is read,
+	 * because it would prevent connections to be reused and cancel
+	 * the benefits of using a connection pooling.
 	 */
 	@Override
 	void close();

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

@@ -134,14 +134,13 @@ public interface ClientResponse extends Closeable {
 	<T> Mono<ResponseEntity<List<T>>> toEntityList(ParameterizedTypeReference<T> typeReference);
 
 	/**
-	 * Close this response, freeing any resources created.
-	 * <p>This non-blocking method has to be called once the response has been processed
-	 * and the resources are no longer needed.
-	 * <p>{@code ClientResponse.bodyTo*}, {@code ClientResponse.toEntity*}
-	 * and all methods under {@code WebClient.retrieve()} will close the response
-	 * automatically.
-	 * <p>It is required to call close() manually otherwise; not doing so might
-	 * create resource leaks or connection issues.
+	 * Close this response and the underlying HTTP connection.
+	 * <p>This non-blocking method has to be called if its body isn't going
+	 * to be consumed. Not doing so might result in HTTP connection pool
+	 * inconsistencies or memory leaks.
+	 * <p>This shouldn't be called if the response body is read,
+	 * because it would prevent connections to be reused and cancel
+	 * the benefits of using a connection pooling.
 	 */
 	@Override
 	void close();

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

@@ -98,24 +98,22 @@ public Map<String, Object> hints() {
 
 	@Override
 	public <T> Mono<T> bodyToMono(Class<? extends T> elementClass) {
-		Mono<T> body = body(BodyExtractors.toMono(elementClass));
-		return body.doOnTerminate(this.response::close);
+		return  body(BodyExtractors.toMono(elementClass));
 	}
 
 	@Override
 	public <T> Mono<T> bodyToMono(ParameterizedTypeReference<T> typeReference) {
-		return body(BodyExtractors.toMono(typeReference)).doOnTerminate(this.response::close);
+		return body(BodyExtractors.toMono(typeReference));
 	}
 
 	@Override
 	public <T> Flux<T> bodyToFlux(Class<? extends T> elementClass) {
-		Flux<T> body = body(BodyExtractors.toFlux(elementClass));
-		return body.doOnTerminate(this.response::close);
+		return body(BodyExtractors.toFlux(elementClass));
 	}
 
 	@Override
 	public <T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> typeReference) {
-		return body(BodyExtractors.toFlux(typeReference)).doOnTerminate(this.response::close);
+		return body(BodyExtractors.toFlux(typeReference));
 	}
 
 	@Override
@@ -134,8 +132,7 @@ private <T> Mono<ResponseEntity<T>> toEntityInternal(Mono<T> bodyMono) {
 		return bodyMono
 				.map(body -> new ResponseEntity<>(body, headers, statusCode))
 				.switchIfEmpty(Mono.defer(
-						() -> Mono.just(new ResponseEntity<>(headers, statusCode))))
-				.doOnTerminate(this.response::close);
+						() -> Mono.just(new ResponseEntity<>(headers, statusCode))));
 	}
 
 	@Override
@@ -154,8 +151,7 @@ private <T> Mono<ResponseEntity<List<T>>> toEntityListInternal(Flux<T> bodyFlux)
 		HttpStatus statusCode = statusCode();
 		return bodyFlux
 				.collectList()
-				.map(body -> new ResponseEntity<>(body, headers, statusCode))
-				.doOnTerminate(this.response::close);
+				.map(body -> new ResponseEntity<>(body, headers, statusCode));
 	}
 
 	@Override

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

@@ -409,64 +409,50 @@ public ResponseSpec onStatus(Predicate<HttpStatus> statusPredicate,
 		@SuppressWarnings("unchecked")
 		public <T> Mono<T> bodyToMono(Class<T> bodyType) {
 			return this.responseMono.flatMap(
-					response -> bodyToMono(response, BodyExtractors.toMono(bodyType),
+					response -> bodyToPublisher(response, BodyExtractors.toMono(bodyType),
 							this::monoThrowableToMono));
 		}
 
 		@Override
 		@SuppressWarnings("unchecked")
 		public <T> Mono<T> bodyToMono(ParameterizedTypeReference<T> typeReference) {
 			return this.responseMono.flatMap(
-					response -> bodyToMono(response, BodyExtractors.toMono(typeReference),
-							this::monoThrowableToMono));
+					response -> bodyToPublisher(response, BodyExtractors.toMono(typeReference),
+							mono -> (Mono<T>)mono));
 		}
 
 		private <T> Mono<T> monoThrowableToMono(Mono<? extends Throwable> mono) {
 			return mono.flatMap(Mono::error);
 		}
 
-		private <T> Mono<T> bodyToMono(ClientResponse response,
-				BodyExtractor<Mono<T>, ? super ClientHttpResponse> extractor,
-				Function<Mono<? extends Throwable>, Mono<T>> errorFunction) {
-
-			return this.statusHandlers.stream()
-					.filter(statusHandler -> statusHandler.test(response.statusCode()))
-					.findFirst()
-					.map(statusHandler -> statusHandler.apply(response))
-					.map(errorFunction::apply)
-					.orElse(response.body(extractor))
-					.doAfterTerminate(response::close);
-		}
-
- 		@Override
+		@Override
 		public <T> Flux<T> bodyToFlux(Class<T> elementType) {
 			return this.responseMono.flatMapMany(
-					response -> bodyToFlux(response, BodyExtractors.toFlux(elementType),
+					response -> bodyToPublisher(response, BodyExtractors.toFlux(elementType),
 							this::monoThrowableToFlux));
 		}
 
 		@Override
 		public <T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> typeReference) {
 			return this.responseMono.flatMapMany(
-					response -> bodyToFlux(response, BodyExtractors.toFlux(typeReference),
+					response -> bodyToPublisher(response, BodyExtractors.toFlux(typeReference),
 							this::monoThrowableToFlux));
 		}
 
 		private <T> Flux<T> monoThrowableToFlux(Mono<? extends Throwable> mono) {
 			return mono.flatMapMany(Flux::error);
 		}
 
-		private <T> Flux<T> bodyToFlux(ClientResponse response,
-				BodyExtractor<Flux<T>, ? super ClientHttpResponse> extractor,
-				Function<Mono<? extends Throwable>, Flux<T>> errorFunction) {
+		private <T extends Publisher<?>> T bodyToPublisher(ClientResponse response,
+				BodyExtractor<T, ? super ClientHttpResponse> extractor,
+				Function<Mono<? extends Throwable>, T> errorFunction) {
 
 			return this.statusHandlers.stream()
 					.filter(statusHandler -> statusHandler.test(response.statusCode()))
 					.findFirst()
 					.map(statusHandler -> statusHandler.apply(response))
 					.map(errorFunction::apply)
-					.orElse(response.body(extractor))
-					.doAfterTerminate(response::close);
+					.orElse(response.body(extractor));
 		}
 
 		private static Mono<WebClientResponseException> createResponseException(ClientResponse response) {

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

@@ -491,6 +491,8 @@ interface RequestHeadersSpec<S extends RequestHeadersSpec<S>> {
 		 *     .retrieve()
 		 *     .bodyToMono(Pojo.class);
 		 * </pre>
+		 * <p>Since this method reads the response body,
+		 * {@link ClientResponse#close()} should not be called.
 		 * @return spec with options for extracting the response body
 		 */
 		ResponseSpec retrieve();

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

@@ -2,19 +2,15 @@
 
 import org.junit.Before;
 import org.junit.Test;
-import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.test.StepVerifier;
 
 import org.springframework.http.HttpHeaders;
 import org.springframework.http.HttpStatus;
-import org.springframework.http.ResponseEntity;
 import org.springframework.http.client.reactive.ClientHttpConnector;
 import org.springframework.mock.http.client.reactive.test.MockClientHttpResponse;
 
-import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.Mockito.mock;
@@ -44,7 +40,7 @@ public void setUp() throws Exception {
 
 	@Test
 	public void shouldDisposeResponseManually() {
-		Mono<HttpHeaders> headers= this.webClient
+		Mono<HttpHeaders> headers = this.webClient
 				.get().uri("/test")
 				.exchange()
 				.map(response -> response.headers().asHttpHeaders());
@@ -54,63 +50,4 @@ public void shouldDisposeResponseManually() {
 		assertFalse(this.response.isClosed());
 	}
 
-	@Test
-	public void shouldDisposeResponseExchangeMono() {
-		Mono<String> body = this.webClient
-				.get().uri("/test")
-				.exchange()
-				.flatMap(response -> response.bodyToMono(String.class));
-		StepVerifier.create(body)
-				.expectNext("example")
-				.verifyComplete();
-		assertTrue(this.response.isClosed());
-	}
-
-	@Test
-	public void shouldDisposeResponseExchangeFlux() {
-		Flux<String> body = this.webClient
-				.get().uri("/test")
-				.exchange()
-				.flatMapMany(response -> response.bodyToFlux(String.class));
-		StepVerifier.create(body)
-				.expectNext("example")
-				.verifyComplete();
-		assertTrue(this.response.isClosed());
-	}
-
-	@Test
-	public void shouldDisposeResponseExchangeEntity() {
-		ResponseEntity<String> entity = this.webClient
-				.get().uri("/test")
-				.exchange()
-				.flatMap(response -> response.toEntity(String.class))
-				.block();
-		assertEquals("example", entity.getBody());
-		assertTrue(this.response.isClosed());
-	}
-
-	@Test
-	public void shouldDisposeResponseRetrieveMono() {
-		Mono<String> body = this.webClient
-				.get().uri("/test")
-				.retrieve()
-				.bodyToMono(String.class);
-		StepVerifier.create(body)
-				.expectNext("example")
-				.verifyComplete();
-		assertTrue(this.response.isClosed());
-	}
-
-	@Test
-	public void shouldDisposeResponseRetrieveFlux() {
-		Flux<String> body = this.webClient
-				.get().uri("/test")
-				.retrieve()
-				.bodyToFlux(String.class);
-		StepVerifier.create(body)
-				.expectNext("example")
-				.verifyComplete();
-		assertTrue(this.response.isClosed());
-	}
-
 }