Add ClientResponse::releaseBody

See gh-23498

Author: poutsma

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

@@ -41,14 +41,26 @@
 
 /**
  * Represents an HTTP response, as returned by {@link WebClient} and also
- * {@link ExchangeFunction}. Provides access to the response status and headers,
- * and also methods to consume the response body.
+ * {@link ExchangeFunction}. Provides access to the response status and
+ * headers, and also methods to consume the response body.
  *
- * <p><strong>NOTE:</strong> When given access to a {@link ClientResponse},
+ * <p><strong>NOTE:</strong> When using a {@link ClientResponse}
  * through the {@code WebClient}
  * {@link WebClient.RequestHeadersSpec#exchange() exchange()} method,
- * you must always use one of the body or toEntity methods to ensure resources
- * are released and avoid potential issues with HTTP connection pooling.
+ * you have to make sure that the body is consumed or released by using
+ * one of the following methods:
+ * <ul>
+ * <li>{@link #body(BodyExtractor)}</li>
+ * <li>{@link #bodyToMono(Class)} or
+ *     {@link #bodyToMono(ParameterizedTypeReference)}</li>
+ * <li>{@link #bodyToFlux(Class)} or
+ *     {@link #bodyToFlux(ParameterizedTypeReference)}</li>
+ * <li>{@link #toEntity(Class)} or
+ *     {@link #toEntity(ParameterizedTypeReference)}</li>
+ * <li>{@link #toEntityList(Class)} or
+ *     {@link #toEntityList(ParameterizedTypeReference)}</li>
+ * <li>{@link #releaseBody()}</li>
+ * </ul>
  * You can use {@code bodyToMono(Void.class)} if no response content is
  * expected. However keep in mind that if the response does have content, the
  * connection will be closed and will not be placed back in the pool.
@@ -132,6 +144,14 @@ public interface ClientResponse {
 	 */
 	<T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> elementTypeRef);
 
+	/**
+	 * Releases the body of this response.
+	 * @return a completion signal
+	 * @since 5.2
+	 * @see org.springframework.core.io.buffer.DataBufferUtils#release(DataBuffer)
+	 */
+	Mono<Void> releaseBody();
+
 	/**
 	 * Return this response as a delayed {@code ResponseEntity}.
 	 * @param bodyClass the expected response body type

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

@@ -155,6 +155,13 @@ public <T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> elementTypeRef) {
 		return body(BodyExtractors.toFlux(elementTypeRef));
 	}
 
+	@Override
+	public Mono<Void> releaseBody() {
+		return body(BodyExtractors.toDataBuffers())
+				.map(DataBufferUtils::release)
+				.then();
+	}
+
 	@Override
 	public <T> Mono<ResponseEntity<T>> toEntity(Class<T> bodyType) {
 		return WebClientUtils.toEntity(this, bodyToMono(bodyType));

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

@@ -118,6 +118,11 @@ public <T> Flux<T> bodyToFlux(ParameterizedTypeReference<T> elementTypeRef) {
 		return this.delegate.bodyToFlux(elementTypeRef);
 	}
 
+	@Override
+	public Mono<Void> releaseBody() {
+		return this.delegate.releaseBody();
+	}
+
 	@Override
 	public <T> Mono<ResponseEntity<T>> toEntity(Class<T> bodyType) {
 		return this.delegate.toEntity(bodyType);

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

@@ -166,6 +166,25 @@ public void onStatusWithImmediateErrorAndBodyNotConsumed(String displayName, Dat
 		});
 	}
 
+	@ParameterizedDataBufferAllocatingTest
+	public void releaseBody(String displayName, DataBufferFactory bufferFactory) {
+		super.bufferFactory = bufferFactory;
+
+		this.server.enqueue(new MockResponse()
+				.setResponseCode(200)
+				.setHeader("Content-Type", "text/plain")
+				.setBody("foo bar"));
+
+		Mono<Void> result  = this.webClient.get()
+				.exchange()
+				.flatMap(ClientResponse::releaseBody);
+
+
+		StepVerifier.create(result)
+				.expectComplete()
+				.verify(Duration.ofSeconds(3));
+	}
+
 
 	private void testOnStatus(Throwable expected,
 			Function<ClientResponse, Mono<? extends Throwable>> exceptionFunction) {