diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerServerHttpResponse.java b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerServerHttpResponse.java index d9126fa0aa..b14f95b30e 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerServerHttpResponse.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerServerHttpResponse.java @@ -26,7 +26,9 @@ import org.springframework.core.io.buffer.DataBuffer; import org.springframework.core.io.buffer.DataBufferFactory; /** - * Abstract base class for listener-based server responses, i.e. Servlet 3.1 and Undertow. + * Abstract base class for listener-based server responses, e.g. Servlet 3.1 + * and Undertow. + * * @author Arjen Poutsma * @since 5.0 */ @@ -34,10 +36,12 @@ public abstract class AbstractListenerServerHttpResponse extends AbstractServerH private final AtomicBoolean writeCalled = new AtomicBoolean(); + public AbstractListenerServerHttpResponse(DataBufferFactory dataBufferFactory) { super(dataBufferFactory); } + @Override protected final Mono writeWithInternal(Publisher body) { return writeAndFlushWithInternal(Mono.just(body)); @@ -46,13 +50,13 @@ public abstract class AbstractListenerServerHttpResponse extends AbstractServerH @Override protected final Mono writeAndFlushWithInternal(Publisher> body) { if (this.writeCalled.compareAndSet(false, true)) { - Processor, Void> bodyProcessor = - createBodyFlushProcessor(); + Processor, Void> bodyProcessor = createBodyFlushProcessor(); return Mono.from(subscriber -> { body.subscribe(bodyProcessor); bodyProcessor.subscribe(subscriber); }); - } else { + } + else { return Mono.error(new IllegalStateException( "writeWith() or writeAndFlushWith() has already been called")); } @@ -64,4 +68,5 @@ public abstract class AbstractListenerServerHttpResponse extends AbstractServerH * {@link #writeAndFlushWithInternal(Publisher)}. */ protected abstract Processor, Void> createBodyFlushProcessor(); + } diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractResponseBodyFlushProcessor.java b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractResponseBodyFlushProcessor.java index 207d9b1c2b..ef63b70377 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractResponseBodyFlushProcessor.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractResponseBodyFlushProcessor.java @@ -35,26 +35,25 @@ import org.springframework.core.io.buffer.DataBuffer; * Servlet 3.1 and Undertow support. * * @author Arjen Poutsma + * @author Violeta Georgieva * @since 5.0 * @see ServletServerHttpRequest * @see UndertowHttpHandlerAdapter * @see ServerHttpResponse#writeAndFlushWith(Publisher) */ -abstract class AbstractResponseBodyFlushProcessor - implements Processor, Void> { +abstract class AbstractResponseBodyFlushProcessor implements Processor, Void> { protected final Log logger = LogFactory.getLog(getClass()); - private final ResponseBodyWriteResultPublisher publisherDelegate = - new ResponseBodyWriteResultPublisher(); + private final ResponseBodyWriteResultPublisher resultPublisher = new ResponseBodyWriteResultPublisher(); - private final AtomicReference state = - new AtomicReference<>(State.UNSUBSCRIBED); + private final AtomicReference state = new AtomicReference<>(State.UNSUBSCRIBED); private volatile boolean subscriberCompleted; private Subscription subscription; + // Subscriber @Override @@ -89,13 +88,15 @@ abstract class AbstractResponseBodyFlushProcessor this.state.get().onComplete(this); } + // Publisher @Override public final void subscribe(Subscriber subscriber) { - this.publisherDelegate.subscribe(subscriber); + this.resultPublisher.subscribe(subscriber); } + /** * Creates a new processor for subscribing to a body chunk. */ @@ -106,8 +107,9 @@ abstract class AbstractResponseBodyFlushProcessor */ protected abstract void flush() throws IOException; - private void cancel() { - this.subscription.cancel(); + + private boolean changeState(State oldState, State newState) { + return this.state.compareAndSet(oldState, newState); } private void writeComplete() { @@ -118,15 +120,17 @@ abstract class AbstractResponseBodyFlushProcessor } - private boolean changeState(State oldState, State newState) { - return this.state.compareAndSet(oldState, newState); + private void cancel() { + this.subscription.cancel(); } + private enum State { + UNSUBSCRIBED { + @Override - public void onSubscribe(AbstractResponseBodyFlushProcessor processor, - Subscription subscription) { + public void onSubscribe(AbstractResponseBodyFlushProcessor processor, Subscription subscription) { Objects.requireNonNull(subscription, "Subscription cannot be null"); if (processor.changeState(this, REQUESTED)) { processor.subscription = subscription; @@ -138,25 +142,25 @@ abstract class AbstractResponseBodyFlushProcessor } }, REQUESTED { + @Override - public void onNext(AbstractResponseBodyFlushProcessor processor, - Publisher chunk) { + public void onNext(AbstractResponseBodyFlushProcessor processor, Publisher chunk) { if (processor.changeState(this, RECEIVED)) { - Processor chunkProcessor = - processor.createBodyProcessor(); + Processor chunkProcessor = processor.createBodyProcessor(); chunk.subscribe(chunkProcessor); chunkProcessor.subscribe(new WriteSubscriber(processor)); } } @Override - void onComplete(AbstractResponseBodyFlushProcessor processor) { + public void onComplete(AbstractResponseBodyFlushProcessor processor) { if (processor.changeState(this, COMPLETED)) { - processor.publisherDelegate.publishComplete(); + processor.resultPublisher.publishComplete(); } } }, RECEIVED { + @Override public void writeComplete(AbstractResponseBodyFlushProcessor processor) { try { @@ -169,7 +173,7 @@ abstract class AbstractResponseBodyFlushProcessor if (processor.subscriberCompleted) { if (processor.changeState(this, COMPLETED)) { - processor.publisherDelegate.publishComplete(); + processor.resultPublisher.publishComplete(); } } else { @@ -180,11 +184,12 @@ abstract class AbstractResponseBodyFlushProcessor } @Override - void onComplete(AbstractResponseBodyFlushProcessor processor) { + public void onComplete(AbstractResponseBodyFlushProcessor processor) { processor.subscriberCompleted = true; } }, COMPLETED { + @Override public void onNext(AbstractResponseBodyFlushProcessor processor, Publisher publisher) { @@ -193,33 +198,31 @@ abstract class AbstractResponseBodyFlushProcessor } @Override - void onError(AbstractResponseBodyFlushProcessor processor, Throwable t) { + public void onError(AbstractResponseBodyFlushProcessor processor, Throwable t) { // ignore } @Override - void onComplete(AbstractResponseBodyFlushProcessor processor) { + public void onComplete(AbstractResponseBodyFlushProcessor processor) { // ignore } }; - public void onSubscribe(AbstractResponseBodyFlushProcessor processor, - Subscription subscription) { + public void onSubscribe(AbstractResponseBodyFlushProcessor processor, Subscription subscription) { subscription.cancel(); } - public void onNext(AbstractResponseBodyFlushProcessor processor, - Publisher publisher) { + public void onNext(AbstractResponseBodyFlushProcessor processor, Publisher publisher) { throw new IllegalStateException(toString()); } - void onError(AbstractResponseBodyFlushProcessor processor, Throwable t) { + public void onError(AbstractResponseBodyFlushProcessor processor, Throwable ex) { if (processor.changeState(this, COMPLETED)) { - processor.publisherDelegate.publishError(t); + processor.resultPublisher.publishError(ex); } } - void onComplete(AbstractResponseBodyFlushProcessor processor) { + public void onComplete(AbstractResponseBodyFlushProcessor processor) { throw new IllegalStateException(toString()); } @@ -227,6 +230,7 @@ abstract class AbstractResponseBodyFlushProcessor // ignore } + private static class WriteSubscriber implements Subscriber { private final AbstractResponseBodyFlushProcessor processor; @@ -236,8 +240,8 @@ abstract class AbstractResponseBodyFlushProcessor } @Override - public void onSubscribe(Subscription s) { - s.request(Long.MAX_VALUE); + public void onSubscribe(Subscription subscription) { + subscription.request(Long.MAX_VALUE); } @Override @@ -245,14 +249,14 @@ abstract class AbstractResponseBodyFlushProcessor } @Override - public void onError(Throwable t) { - processor.cancel(); - processor.onError(t); + public void onError(Throwable ex) { + this.processor.cancel(); + this.processor.onError(ex); } @Override public void onComplete() { - processor.writeComplete(); + this.processor.writeComplete(); } } } diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractResponseBodyProcessor.java b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractResponseBodyProcessor.java index 2463fb9414..3e62264dd1 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractResponseBodyProcessor.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractResponseBodyProcessor.java @@ -48,11 +48,9 @@ abstract class AbstractResponseBodyProcessor implements Processor state = - new AtomicReference<>(State.UNSUBSCRIBED); + private final AtomicReference state = new AtomicReference<>(State.UNSUBSCRIBED); private volatile DataBuffer currentBuffer; @@ -60,6 +58,7 @@ abstract class AbstractResponseBodyProcessor implements Processor subscriber) { - this.publisherDelegate.subscribe(subscriber); + this.resultPublisher.subscribe(subscriber); } + // listener methods /** @@ -167,6 +168,7 @@ abstract class AbstractResponseBodyProcessor implements Processor, Void> createBodyFlushProcessor() { + Processor, Void> processor = new ResponseBodyFlushProcessor(); + registerListener(); + return processor; + } + private void registerListener() { try { outputStream().setWriteListener(writeListener); @@ -121,6 +131,10 @@ public class ServletServerHttpResponse extends AbstractListenerServerHttpRespons } } + private ServletOutputStream outputStream() throws IOException { + return this.response.getOutputStream(); + } + private void flush() throws IOException { ServletOutputStream outputStream = outputStream(); if (outputStream.isReady()) { @@ -138,16 +152,6 @@ public class ServletServerHttpResponse extends AbstractListenerServerHttpRespons } } - private ServletOutputStream outputStream() throws IOException { - return this.response.getOutputStream(); - } - - @Override - protected Processor, Void> createBodyFlushProcessor() { - Processor, Void> processor = new ResponseBodyFlushProcessor(); - registerListener(); - return processor; - } private class ResponseBodyProcessor extends AbstractResponseBodyProcessor { @@ -155,11 +159,13 @@ public class ServletServerHttpResponse extends AbstractListenerServerHttpRespons private final int bufferSize; + public ResponseBodyProcessor(ServletOutputStream outputStream, int bufferSize) { this.outputStream = outputStream; this.bufferSize = bufferSize; } + @Override protected boolean isWritePossible() { return this.outputStream.isReady(); @@ -201,8 +207,7 @@ public class ServletServerHttpResponse extends AbstractListenerServerHttpRespons byte[] buffer = new byte[this.bufferSize]; int bytesRead = -1; - while (this.outputStream.isReady() && - (bytesRead = input.read(buffer)) != -1) { + while (this.outputStream.isReady() && (bytesRead = input.read(buffer)) != -1) { this.outputStream.write(buffer, 0, bytesRead); bytesWritten += bytesRead; } diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpResponse.java b/spring-web/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpResponse.java index b8df141cb9..30102b6e2d 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpResponse.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/UndertowServerHttpResponse.java @@ -55,13 +55,14 @@ public class UndertowServerHttpResponse extends AbstractListenerServerHttpRespon private StreamSinkChannel responseChannel; - public UndertowServerHttpResponse(HttpServerExchange exchange, - DataBufferFactory dataBufferFactory) { - super(dataBufferFactory); + + public UndertowServerHttpResponse(HttpServerExchange exchange, DataBufferFactory bufferFactory) { + super(bufferFactory); Assert.notNull(exchange, "'exchange' is required."); this.exchange = exchange; } + public HttpServerExchange getUndertowExchange() { return this.exchange; } @@ -78,10 +79,8 @@ public class UndertowServerHttpResponse extends AbstractListenerServerHttpRespon public Mono writeWith(File file, long position, long count) { writeHeaders(); writeCookies(); - try { - StreamSinkChannel responseChannel = - getUndertowExchange().getResponseChannel(); + StreamSinkChannel responseChannel = getUndertowExchange().getResponseChannel(); @SuppressWarnings("resource") FileChannel in = new FileInputStream(file).getChannel(); long result = responseChannel.transferFrom(in, position, count); @@ -123,21 +122,21 @@ public class UndertowServerHttpResponse extends AbstractListenerServerHttpRespon } } - private ResponseBodyProcessor createBodyProcessor() { - if (this.responseChannel == null) { - this.responseChannel = this.exchange.getResponseChannel(); - } - ResponseBodyProcessor bodyProcessor = - new ResponseBodyProcessor( this.responseChannel); - bodyProcessor.registerListener(); - return bodyProcessor; - } - @Override protected AbstractResponseBodyFlushProcessor createBodyFlushProcessor() { return new ResponseBodyFlushProcessor(); } + private ResponseBodyProcessor createBodyProcessor() { + if (this.responseChannel == null) { + this.responseChannel = this.exchange.getResponseChannel(); + } + ResponseBodyProcessor bodyProcessor = new ResponseBodyProcessor( this.responseChannel); + bodyProcessor.registerListener(); + return bodyProcessor; + } + + private static class ResponseBodyProcessor extends AbstractResponseBodyProcessor { private final ChannelListener listener = new WriteListener(); @@ -146,11 +145,13 @@ public class UndertowServerHttpResponse extends AbstractListenerServerHttpRespon private volatile ByteBuffer byteBuffer; + public ResponseBodyProcessor(StreamSinkChannel responseChannel) { Assert.notNull(responseChannel, "'responseChannel' must not be null"); this.responseChannel = responseChannel; } + public void registerListener() { this.responseChannel.getWriteSetter().set(this.listener); this.responseChannel.resumeWrites(); @@ -202,9 +203,7 @@ public class UndertowServerHttpResponse extends AbstractListenerServerHttpRespon public void handleEvent(StreamSinkChannel channel) { onWritePossible(); } - } - } private class ResponseBodyFlushProcessor extends AbstractResponseBodyFlushProcessor { @@ -223,6 +222,6 @@ public class UndertowServerHttpResponse extends AbstractListenerServerHttpRespon UndertowServerHttpResponse.this.responseChannel.flush(); } } - } + }