Improved RxNetty support.

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

@@ -34,18 +34,16 @@ public class RxNettyHttpHandlerAdapter implements RequestHandler<ByteBuf, ByteBu
 
 	private final HttpHandler httpHandler;
 
-	private final NettyDataBufferAllocator allocator;
-
-	public RxNettyHttpHandlerAdapter(HttpHandler httpHandler,
-			NettyDataBufferAllocator allocator) {
+	public RxNettyHttpHandlerAdapter(HttpHandler httpHandler) {
 		Assert.notNull(httpHandler, "'httpHandler' is required");
-		Assert.notNull(allocator, "'allocator' must not be null");
 		this.httpHandler = httpHandler;
-		this.allocator = allocator;
 	}
 
 	@Override
 	public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) {
+		NettyDataBufferAllocator allocator =
+				new NettyDataBufferAllocator(response.unsafeNettyChannel().alloc());
+
 		RxNettyServerHttpRequest adaptedRequest =
 				new RxNettyServerHttpRequest(request, allocator);
 		RxNettyServerHttpResponse adaptedResponse = new RxNettyServerHttpResponse(response);

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

@@ -16,6 +16,8 @@
 
 package org.springframework.http.server.reactive;
 
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
 import io.netty.handler.codec.http.HttpResponseStatus;
 import io.netty.handler.codec.http.cookie.Cookie;
 import io.netty.handler.codec.http.cookie.DefaultCookie;
@@ -26,6 +28,7 @@ import reactor.core.publisher.Mono;
 import rx.Observable;
 
 import org.springframework.core.io.buffer.DataBuffer;
+import org.springframework.core.io.buffer.NettyDataBuffer;
 import org.springframework.http.HttpCookie;
 import org.springframework.http.HttpStatus;
 import org.springframework.util.Assert;
@@ -38,10 +41,9 @@ import org.springframework.util.Assert;
  */
 public class RxNettyServerHttpResponse extends AbstractServerHttpResponse {
 
-	private final HttpServerResponse<?> response;
+	private final HttpServerResponse<ByteBuf> response;
 
-
-	public RxNettyServerHttpResponse(HttpServerResponse<?> response) {
+	public RxNettyServerHttpResponse(HttpServerResponse<ByteBuf> response) {
 		Assert.notNull("'response', response must not be null.");
 		this.response = response;
 	}
@@ -58,15 +60,19 @@ public class RxNettyServerHttpResponse extends AbstractServerHttpResponse {
 
 	@Override
 	protected Mono<Void> setBodyInternal(Publisher<DataBuffer> publisher) {
-		Observable<byte[]> content = RxJava1ObservableConverter.from(publisher).map(this::toBytes);
-		Observable<Void> completion = this.response.writeBytes(content);
+		Observable<ByteBuf> content =
+				RxJava1ObservableConverter.from(publisher).map(this::toByteBuf);
+		Observable<Void> completion = this.response.write(content);
 		return RxJava1ObservableConverter.from(completion).after();
 	}
 
-	private byte[] toBytes(DataBuffer buffer) {
-		byte[] bytes = new byte[buffer.readableByteCount()];
-		buffer.read(bytes);
-		return bytes;
+	private ByteBuf toByteBuf(DataBuffer buffer) {
+		if (buffer instanceof NettyDataBuffer) {
+			return ((NettyDataBuffer) buffer).getNativeBuffer();
+		}
+		else {
+			return Unpooled.wrappedBuffer(buffer.asByteBuffer());
+		}
 	}
 
 	@Override

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

@@ -17,10 +17,7 @@
 package org.springframework.http.server.reactive.boot;
 
 import io.netty.buffer.ByteBuf;
-import io.netty.buffer.ByteBufAllocator;
-import io.netty.buffer.UnpooledByteBufAllocator;
 
-import org.springframework.core.io.buffer.NettyDataBufferAllocator;
 import org.springframework.http.server.reactive.RxNettyHttpHandlerAdapter;
 import org.springframework.util.Assert;
 
@@ -34,22 +31,12 @@ public class RxNettyHttpServer extends HttpServerSupport implements HttpServer {
 
 	private io.reactivex.netty.protocol.http.server.HttpServer<ByteBuf, ByteBuf> rxNettyServer;
 
-	private NettyDataBufferAllocator allocator;
-
 	private boolean running;
 
-	public void setAllocator(ByteBufAllocator allocator) {
-		Assert.notNull(allocator, "'allocator' must not be null");
-		this.allocator = new NettyDataBufferAllocator(allocator);
-	}
-
 	@Override
 	public void afterPropertiesSet() throws Exception {
 		Assert.notNull(getHttpHandler());
-		if (allocator == null) {
-			allocator = new NettyDataBufferAllocator(UnpooledByteBufAllocator.DEFAULT);
-		}
-		this.rxNettyHandler = new RxNettyHttpHandlerAdapter(getHttpHandler(), allocator);
+		this.rxNettyHandler = new RxNettyHttpHandlerAdapter(getHttpHandler());
 
 		this.rxNettyServer = (getPort() != -1 ?
 				io.reactivex.netty.protocol.http.server.HttpServer.newServer(getPort()) :