Enforce use of unpooled data buffers in 5.0.x

Issue: SPR-17501

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;
 
 

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

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);
 				});
 	}
 

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<HttpServerRequest, HttpServerResponse, Mono<Void>> {
 
+	// 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<HttpServerRequest,
 
 	@Override
 	public Mono<Void> 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()) {

spring-web/src/main/java/org/springframework/http/server/reactive/ReactorServerHttpRequest.java

@@ -167,7 +167,12 @@ class ReactorServerHttpRequest extends AbstractServerHttpRequest {
 
 	@Override
 	public Flux<DataBuffer> 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")