From 10139d42fc57131386f57d5b160412cfb0c9fc86 Mon Sep 17 00:00:00 2001 From: Brian Clozel Date: Wed, 27 Sep 2017 13:58:45 +0200 Subject: [PATCH] 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 --- .../client/reactive/ClientHttpResponse.java | 13 ++-- .../function/client/ClientResponse.java | 15 ++--- .../client/DefaultClientResponse.java | 16 ++--- .../function/client/DefaultWebClient.java | 34 +++------- .../reactive/function/client/WebClient.java | 2 + .../function/client/WebClientMockTests.java | 65 +------------------ 6 files changed, 33 insertions(+), 112 deletions(-) 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()); - } - }