diff --git a/spring-web/src/main/java/org/springframework/http/client/reactive/ClientHttpResponse.java b/spring-web/src/main/java/org/springframework/http/client/reactive/ClientHttpResponse.java index 3a3955a629..39bf1b3e77 100644 --- a/spring-web/src/main/java/org/springframework/http/client/reactive/ClientHttpResponse.java +++ b/spring-web/src/main/java/org/springframework/http/client/reactive/ClientHttpResponse.java @@ -43,12 +43,13 @@ public interface ClientHttpResponse extends ReactiveHttpInputMessage, Closeable MultiValueMap getCookies(); /** - * Close this response, freeing any resources created. - *

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. - *

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. + *

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. + *

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(); diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/ClientResponse.java b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/ClientResponse.java index 6e4654d456..2247a6c5bc 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/ClientResponse.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/ClientResponse.java @@ -134,14 +134,13 @@ public interface ClientResponse extends Closeable { Mono>> toEntityList(ParameterizedTypeReference typeReference); /** - * Close this response, freeing any resources created. - *

This non-blocking method has to be called once the response has been processed - * and the resources are no longer needed. - *

{@code ClientResponse.bodyTo*}, {@code ClientResponse.toEntity*} - * and all methods under {@code WebClient.retrieve()} will close the response - * automatically. - *

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. + *

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. + *

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(); diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultClientResponse.java b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultClientResponse.java index 31d8820d25..c58e3ba455 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultClientResponse.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultClientResponse.java @@ -98,24 +98,22 @@ class DefaultClientResponse implements ClientResponse { @Override public Mono bodyToMono(Class elementClass) { - Mono body = body(BodyExtractors.toMono(elementClass)); - return body.doOnTerminate(this.response::close); + return body(BodyExtractors.toMono(elementClass)); } @Override public Mono bodyToMono(ParameterizedTypeReference typeReference) { - return body(BodyExtractors.toMono(typeReference)).doOnTerminate(this.response::close); + return body(BodyExtractors.toMono(typeReference)); } @Override public Flux bodyToFlux(Class elementClass) { - Flux body = body(BodyExtractors.toFlux(elementClass)); - return body.doOnTerminate(this.response::close); + return body(BodyExtractors.toFlux(elementClass)); } @Override public Flux bodyToFlux(ParameterizedTypeReference typeReference) { - return body(BodyExtractors.toFlux(typeReference)).doOnTerminate(this.response::close); + return body(BodyExtractors.toFlux(typeReference)); } @Override @@ -134,8 +132,7 @@ class DefaultClientResponse implements ClientResponse { 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 @@ class DefaultClientResponse implements ClientResponse { 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 diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java index f74a832cc7..29e20a40e8 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/DefaultWebClient.java @@ -409,7 +409,7 @@ class DefaultWebClient implements WebClient { @SuppressWarnings("unchecked") public Mono bodyToMono(Class bodyType) { return this.responseMono.flatMap( - response -> bodyToMono(response, BodyExtractors.toMono(bodyType), + response -> bodyToPublisher(response, BodyExtractors.toMono(bodyType), this::monoThrowableToMono)); } @@ -417,38 +417,25 @@ class DefaultWebClient implements WebClient { @SuppressWarnings("unchecked") public Mono bodyToMono(ParameterizedTypeReference typeReference) { return this.responseMono.flatMap( - response -> bodyToMono(response, BodyExtractors.toMono(typeReference), - this::monoThrowableToMono)); + response -> bodyToPublisher(response, BodyExtractors.toMono(typeReference), + mono -> (Mono)mono)); } private Mono monoThrowableToMono(Mono mono) { return mono.flatMap(Mono::error); } - private Mono bodyToMono(ClientResponse response, - BodyExtractor, ? super ClientHttpResponse> extractor, - Function, Mono> 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 Flux bodyToFlux(Class elementType) { return this.responseMono.flatMapMany( - response -> bodyToFlux(response, BodyExtractors.toFlux(elementType), + response -> bodyToPublisher(response, BodyExtractors.toFlux(elementType), this::monoThrowableToFlux)); } @Override public Flux bodyToFlux(ParameterizedTypeReference typeReference) { return this.responseMono.flatMapMany( - response -> bodyToFlux(response, BodyExtractors.toFlux(typeReference), + response -> bodyToPublisher(response, BodyExtractors.toFlux(typeReference), this::monoThrowableToFlux)); } @@ -456,17 +443,16 @@ class DefaultWebClient implements WebClient { return mono.flatMapMany(Flux::error); } - private Flux bodyToFlux(ClientResponse response, - BodyExtractor, ? super ClientHttpResponse> extractor, - Function, Flux> errorFunction) { + private > T bodyToPublisher(ClientResponse response, + BodyExtractor extractor, + Function, 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 createResponseException(ClientResponse response) { diff --git a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/WebClient.java b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/WebClient.java index eec6489270..247e0804a6 100644 --- a/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/WebClient.java +++ b/spring-webflux/src/main/java/org/springframework/web/reactive/function/client/WebClient.java @@ -491,6 +491,8 @@ public interface WebClient { * .retrieve() * .bodyToMono(Pojo.class); * + *

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(); diff --git a/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientMockTests.java b/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientMockTests.java index fabe7abce7..0fe7994fff 100644 --- a/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientMockTests.java +++ b/spring-webflux/src/test/java/org/springframework/web/reactive/function/client/WebClientMockTests.java @@ -2,19 +2,15 @@ package org.springframework.web.reactive.function.client; 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 class WebClientMockTests { @Test public void shouldDisposeResponseManually() { - Mono headers= this.webClient + Mono headers = this.webClient .get().uri("/test") .exchange() .map(response -> response.headers().asHttpHeaders()); @@ -54,63 +50,4 @@ public class WebClientMockTests { assertFalse(this.response.isClosed()); } - @Test - public void shouldDisposeResponseExchangeMono() { - Mono 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 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 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 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 body = this.webClient - .get().uri("/test") - .retrieve() - .bodyToFlux(String.class); - StepVerifier.create(body) - .expectNext("example") - .verifyComplete(); - assertTrue(this.response.isClosed()); - } - }