Moved body processor to writeWithInternal()

Moved ResponseBodyProcessor creation from constructor to
writeWithInternal(), in preparation of supporting both
Publisher<DataBuffer> as well as Publisher<Publisher<DataBuffer>>.

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

@@ -76,12 +76,10 @@ public class ServletHttpHandlerAdapter extends HttpServlet {
 		ServletServerHttpRequest request =
 				new ServletServerHttpRequest(servletRequest, this.dataBufferFactory,
 						this.bufferSize);
-		request.registerListener();
 
 		ServletServerHttpResponse response =
 				new ServletServerHttpResponse(servletResponse, this.dataBufferFactory,
 						this.bufferSize);
-		response.registerListener();
 
 		HandlerResultSubscriber resultSubscriber =
 				new HandlerResultSubscriber(asyncContext);

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

@@ -47,18 +47,25 @@ import org.springframework.util.StringUtils;
  */
 public class ServletServerHttpRequest extends AbstractServerHttpRequest {
 
+	private final Object bodyPublisherMonitor = new Object();
+
+	private volatile RequestBodyPublisher bodyPublisher;
+
 	private final HttpServletRequest request;
 
-	private final RequestBodyPublisher bodyPublisher;
+	private final DataBufferFactory dataBufferFactory;
+
+	private final int bufferSize;
 
 	public ServletServerHttpRequest(HttpServletRequest request,
-			DataBufferFactory dataBufferFactory, int bufferSize) throws IOException {
+			DataBufferFactory dataBufferFactory, int bufferSize) {
 		Assert.notNull(request, "'request' must not be null.");
 		Assert.notNull(dataBufferFactory, "'dataBufferFactory' must not be null");
+		Assert.isTrue(bufferSize > 0);
+
 		this.request = request;
-		this.bodyPublisher =
-				new RequestBodyPublisher(request.getInputStream(), dataBufferFactory,
-						bufferSize);
+		this.dataBufferFactory = dataBufferFactory;
+		this.bufferSize = bufferSize;
 	}
 
 	public HttpServletRequest getServletRequest() {
@@ -136,11 +143,29 @@ public class ServletServerHttpRequest extends AbstractServerHttpRequest {
 
 	@Override
 	public Flux<DataBuffer> getBody() {
-		return Flux.from(this.bodyPublisher);
+		try {
+			RequestBodyPublisher bodyPublisher = this.bodyPublisher;
+			if (bodyPublisher == null) {
+				synchronized (this.bodyPublisherMonitor) {
+					bodyPublisher = this.bodyPublisher;
+					if (bodyPublisher == null) {
+						this.bodyPublisher = bodyPublisher = createBodyPublisher();
+					}
+				}
+			}
+			return Flux.from(bodyPublisher);
+		}
+		catch (IOException ex) {
+			return Flux.error(ex);
+		}
 	}
 
-	public void registerListener() throws IOException {
-		this.bodyPublisher.registerListener();
+	private RequestBodyPublisher createBodyPublisher() throws IOException {
+		RequestBodyPublisher bodyPublisher =
+				new RequestBodyPublisher(request.getInputStream(), this.dataBufferFactory,
+						this.bufferSize);
+		bodyPublisher.registerListener();
+		return bodyPublisher;
 	}
 
 	private static class RequestBodyPublisher extends AbstractRequestBodyPublisher {

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

@@ -42,17 +42,23 @@ import org.springframework.util.Assert;
  */
 public class ServletServerHttpResponse extends AbstractServerHttpResponse {
 
+	private final Object bodyProcessorMonitor = new Object();
+
+	private volatile ResponseBodyProcessor bodyProcessor;
+
 	private final HttpServletResponse response;
 
-	private ResponseBodyProcessor bodyProcessor;
+	private final int bufferSize;
 
 	public ServletServerHttpResponse(HttpServletResponse response,
 			DataBufferFactory dataBufferFactory, int bufferSize) throws IOException {
 		super(dataBufferFactory);
 		Assert.notNull(response, "'response' must not be null");
+		Assert.notNull(dataBufferFactory, "'dataBufferFactory' must not be null");
+		Assert.isTrue(bufferSize > 0);
+
 		this.response = response;
-		this.bodyProcessor =
-				new ResponseBodyProcessor(response.getOutputStream(), bufferSize);
+		this.bufferSize = bufferSize;
 	}
 
 	public HttpServletResponse getServletResponse() {
@@ -69,10 +75,34 @@ public class ServletServerHttpResponse extends AbstractServerHttpResponse {
 
 	@Override
 	protected Mono<Void> writeWithInternal(Publisher<DataBuffer> publisher) {
-		return Mono.from(subscriber -> {
-			publisher.subscribe(this.bodyProcessor);
-			this.bodyProcessor.subscribe(subscriber);
-		});
+		Assert.state(this.bodyProcessor == null,
+				"Response body publisher is already provided");
+		try {
+			synchronized (this.bodyProcessorMonitor) {
+				if (this.bodyProcessor == null) {
+					this.bodyProcessor = createBodyProcessor();
+				}
+				else {
+					throw new IllegalStateException(
+							"Response body publisher is already provided");
+				}
+			}
+			return Mono.from(subscriber -> {
+				publisher.subscribe(this.bodyProcessor);
+				this.bodyProcessor.subscribe(subscriber);
+			});
+		}
+		catch (IOException ex) {
+			return Mono.error(ex);
+		}
+	}
+
+	private ResponseBodyProcessor createBodyProcessor() throws IOException {
+		ResponseBodyProcessor bodyProcessor =
+				new ResponseBodyProcessor(this.response.getOutputStream(),
+						this.bufferSize);
+		bodyProcessor.registerListener();
+		return bodyProcessor;
 	}
 
 	@Override
@@ -110,10 +140,6 @@ public class ServletServerHttpResponse extends AbstractServerHttpResponse {
 		}
 	}
 
-	public void registerListener() throws IOException {
-		this.bodyProcessor.registerListener();
-	}
-
 	private static class ResponseBodyProcessor extends AbstractResponseBodyProcessor {
 
 		private final ResponseBodyWriteListener writeListener =

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

@@ -42,7 +42,6 @@ import org.springframework.util.Assert;
 
 /**
  * Adapt {@link ServerHttpResponse} to the Undertow {@link HttpServerExchange}.
- *
  * @author Marek Hawrylczak
  * @author Rossen Stoyanchev
  * @author Arjen Poutsma
@@ -50,8 +49,11 @@ import org.springframework.util.Assert;
 public class UndertowServerHttpResponse extends AbstractServerHttpResponse
 		implements ZeroCopyHttpOutputMessage {
 
-	private final HttpServerExchange exchange;
+	private final Object bodyProcessorMonitor = new Object();
 
+	private volatile ResponseBodyProcessor bodyProcessor;
+
+	private final HttpServerExchange exchange;
 
 	public UndertowServerHttpResponse(HttpServerExchange exchange,
 			DataBufferFactory dataBufferFactory) {
@@ -60,12 +62,10 @@ public class UndertowServerHttpResponse extends AbstractServerHttpResponse
 		this.exchange = exchange;
 	}
 
-
 	public HttpServerExchange getUndertowExchange() {
 		return this.exchange;
 	}
 
-
 	@Override
 	protected void writeStatusCode() {
 		HttpStatus statusCode = this.getStatusCode();
@@ -74,20 +74,37 @@ public class UndertowServerHttpResponse extends AbstractServerHttpResponse
 		}
 	}
 
-
 	@Override
 	protected Mono<Void> writeWithInternal(Publisher<DataBuffer> publisher) {
-		// lazily create Subscriber, since calling
-		// {@link HttpServerExchange#getResponseChannel} as done in the
-		// ResponseBodyProcessor constructor commits the response status and headers
-		return Mono.from(subscriber -> {
-			ResponseBodyProcessor processor = new ResponseBodyProcessor(this.exchange);
-			processor.registerListener();
-			publisher.subscribe(processor);
-			processor.subscribe(subscriber);
-		});
+		Assert.state(this.bodyProcessor == null,
+				"Response body publisher is already provided");
+		try {
+			synchronized (this.bodyProcessorMonitor) {
+				if (this.bodyProcessor == null) {
+					this.bodyProcessor = createBodyProcessor();
+				}
+				else {
+					throw new IllegalStateException(
+							"Response body publisher is already provided");
+				}
+			}
+			return Mono.from(subscriber -> {
+				publisher.subscribe(this.bodyProcessor);
+				this.bodyProcessor.subscribe(subscriber);
+			});
+		}
+		catch (IOException ex) {
+			return Mono.error(ex);
+		}
 	}
 
+	private ResponseBodyProcessor createBodyProcessor() throws IOException {
+		ResponseBodyProcessor bodyProcessor = new ResponseBodyProcessor(this.exchange);
+		bodyProcessor.registerListener();
+		return bodyProcessor;
+	}
+
+
 	@Override
 	public Mono<Void> writeWith(File file, long position, long count) {
 		writeHeaders();
@@ -99,8 +116,8 @@ public class UndertowServerHttpResponse extends AbstractServerHttpResponse
 			FileChannel in = new FileInputStream(file).getChannel();
 			long result = responseChannel.transferFrom(in, position, count);
 			if (result < count) {
-				return Mono.error(new IOException("Could only write " + result +
-						" out of " + count + " bytes"));
+				return Mono.error(new IOException(
+						"Could only write " + result + " out of " + count + " bytes"));
 			}
 			else {
 				return Mono.empty();