From 18da718bf5be81e0888d38a9b5b627f1c0f774b1 Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Fri, 16 Nov 2018 11:26:54 -0500 Subject: [PATCH] Enforce use of unpooled data buffers in 5.0.x Issue: SPR-17501 --- .../client/reactive/ReactorClientHttpConnector.java | 7 +++++++ .../client/reactive/ReactorClientHttpRequest.java | 5 +---- .../client/reactive/ReactorClientHttpResponse.java | 12 +++++------- .../server/reactive/ReactorHttpHandlerAdapter.java | 10 +++++++--- .../server/reactive/ReactorServerHttpRequest.java | 7 ++++++- 5 files changed, 26 insertions(+), 15 deletions(-) diff --git a/spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpConnector.java b/spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpConnector.java index 56a59bbf96..615c2b7973 100644 --- a/spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpConnector.java +++ b/spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpConnector.java @@ -20,6 +20,7 @@ import java.net.URI; import java.util.function.Consumer; import java.util.function.Function; +import io.netty.buffer.UnpooledByteBufAllocator; import reactor.core.publisher.Mono; import reactor.ipc.netty.http.client.HttpClient; import reactor.ipc.netty.http.client.HttpClientOptions; @@ -27,6 +28,7 @@ import reactor.ipc.netty.http.client.HttpClientRequest; import reactor.ipc.netty.http.client.HttpClientResponse; import reactor.ipc.netty.options.ClientOptions; +import org.springframework.core.io.buffer.NettyDataBufferFactory; import org.springframework.http.HttpMethod; /** @@ -38,6 +40,11 @@ import org.springframework.http.HttpMethod; */ public class ReactorClientHttpConnector implements ClientHttpConnector { + // 5.0.x only: no buffer pooling + static final NettyDataBufferFactory BUFFER_FACTORY = + new NettyDataBufferFactory(new UnpooledByteBufAllocator(false)); + + private final HttpClient httpClient; diff --git a/spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpRequest.java b/spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpRequest.java index 3302bc79ab..99a2b5426a 100644 --- a/spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpRequest.java +++ b/spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpRequest.java @@ -48,21 +48,18 @@ class ReactorClientHttpRequest extends AbstractClientHttpRequest implements Zero private final HttpClientRequest httpRequest; - private final NettyDataBufferFactory bufferFactory; - public ReactorClientHttpRequest(HttpMethod httpMethod, URI uri, HttpClientRequest httpRequest) { this.httpMethod = httpMethod; this.uri = uri; this.httpRequest = httpRequest.failOnClientError(false).failOnServerError(false); - this.bufferFactory = new NettyDataBufferFactory(httpRequest.alloc()); } @Override public DataBufferFactory bufferFactory() { - return this.bufferFactory; + return ReactorClientHttpConnector.BUFFER_FACTORY; } @Override diff --git a/spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpResponse.java b/spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpResponse.java index c050e2cdda..d06c74e886 100644 --- a/spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpResponse.java +++ b/spring-web/src/main/java/org/springframework/http/client/reactive/ReactorClientHttpResponse.java @@ -23,7 +23,6 @@ import reactor.core.publisher.Flux; import reactor.ipc.netty.http.client.HttpClientResponse; import org.springframework.core.io.buffer.DataBuffer; -import org.springframework.core.io.buffer.NettyDataBufferFactory; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseCookie; @@ -41,8 +40,6 @@ import org.springframework.util.MultiValueMap; */ class ReactorClientHttpResponse implements ClientHttpResponse { - private final NettyDataBufferFactory dataBufferFactory; - private final HttpClientResponse response; private final AtomicBoolean bodyConsumed = new AtomicBoolean(); @@ -50,7 +47,6 @@ class ReactorClientHttpResponse implements ClientHttpResponse { public ReactorClientHttpResponse(HttpClientResponse response) { this.response = response; - this.dataBufferFactory = new NettyDataBufferFactory(response.channel().alloc()); } @@ -64,9 +60,11 @@ class ReactorClientHttpResponse implements ClientHttpResponse { // isn't consistent in doing so and may hang without completion. Assert.state(this.bodyConsumed.compareAndSet(false, true), "The client response body can only be consumed once.")) - .map(buf -> { - buf.retain(); - return dataBufferFactory.wrap(buf); + .map(byteBuf -> { + // 5.0.x only: do not retain, make a copy.. + byte[] data = new byte[byteBuf.readableBytes()]; + byteBuf.readBytes(data); + return ReactorClientHttpConnector.BUFFER_FACTORY.wrap(data); }); } diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/ReactorHttpHandlerAdapter.java b/spring-web/src/main/java/org/springframework/http/server/reactive/ReactorHttpHandlerAdapter.java index a53deaf2ff..827038fa9e 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/ReactorHttpHandlerAdapter.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/ReactorHttpHandlerAdapter.java @@ -19,6 +19,7 @@ package org.springframework.http.server.reactive; import java.net.URISyntaxException; import java.util.function.BiFunction; +import io.netty.buffer.UnpooledByteBufAllocator; import io.netty.handler.codec.http.HttpResponseStatus; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -39,6 +40,10 @@ import org.springframework.util.Assert; */ public class ReactorHttpHandlerAdapter implements BiFunction> { + // 5.0.x only: no buffer pooling + private static final NettyDataBufferFactory BUFFER_FACTORY = + new NettyDataBufferFactory(new UnpooledByteBufAllocator(false)); + private static final Log logger = LogFactory.getLog(ReactorHttpHandlerAdapter.class); @@ -53,12 +58,11 @@ public class ReactorHttpHandlerAdapter implements BiFunction apply(HttpServerRequest request, HttpServerResponse response) { - NettyDataBufferFactory bufferFactory = new NettyDataBufferFactory(response.alloc()); ServerHttpRequest adaptedRequest; ServerHttpResponse adaptedResponse; try { - adaptedRequest = new ReactorServerHttpRequest(request, bufferFactory); - adaptedResponse = new ReactorServerHttpResponse(response, bufferFactory); + adaptedRequest = new ReactorServerHttpRequest(request, BUFFER_FACTORY); + adaptedResponse = new ReactorServerHttpResponse(response, BUFFER_FACTORY); } catch (URISyntaxException ex) { if (logger.isWarnEnabled()) { diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/ReactorServerHttpRequest.java b/spring-web/src/main/java/org/springframework/http/server/reactive/ReactorServerHttpRequest.java index be5c6572fc..89d3a25214 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/ReactorServerHttpRequest.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/ReactorServerHttpRequest.java @@ -167,7 +167,12 @@ class ReactorServerHttpRequest extends AbstractServerHttpRequest { @Override public Flux getBody() { - return this.request.receive().retain().map(this.bufferFactory::wrap); + // 5.0.x only: do not retain, make a copy.. + return this.request.receive().map(byteBuf -> { + byte[] data = new byte[byteBuf.readableBytes()]; + byteBuf.readBytes(data); + return bufferFactory.wrap(data); + }); } @SuppressWarnings("unchecked")