diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteFlushProcessor.java b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteFlushProcessor.java index 9009026bbc..f88416283c 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteFlushProcessor.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteFlushProcessor.java @@ -132,11 +132,15 @@ public abstract class AbstractListenerWriteFlushProcessor implements Processo /** * Flush the output if ready, or otherwise {@link #isFlushPending()} should * return true after. + *

This is primarily for the Servlet non-blocking I/O API where flush + * cannot be called without a readyToWrite check. */ protected abstract void flush() throws IOException; /** * Whether flushing is pending. + *

This is primarily for the Servlet non-blocking I/O API where flush + * cannot be called without a readyToWrite check. */ protected abstract boolean isFlushPending(); diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteProcessor.java b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteProcessor.java index 7aed653eb2..bc55d0052b 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteProcessor.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteProcessor.java @@ -193,6 +193,12 @@ public abstract class AbstractListenerWriteProcessor implements Processor implements Processor implements Processor { @@ -151,16 +141,24 @@ class UndertowServerHttpResponse extends AbstractListenerServerHttpResponse impl @Nullable private volatile ByteBuffer byteBuffer; + /** Keep track of write listener calls, for {@link #writePossible}. */ + private volatile boolean writePossible; + + public ResponseBodyProcessor(StreamSinkChannel channel) { Assert.notNull(channel, "StreamSinkChannel must not be null"); this.channel = channel; - this.channel.getWriteSetter().set(c -> onWritePossible()); + this.channel.getWriteSetter().set(c -> { + this.writePossible = true; + onWritePossible(); + }); this.channel.suspendWrites(); } @Override protected boolean isWritePossible() { - return UndertowServerHttpResponse.this.isWritePossible(); + this.channel.resumeWrites(); + return this.writePossible; } @Override @@ -172,6 +170,10 @@ class UndertowServerHttpResponse extends AbstractListenerServerHttpResponse impl if (logger.isTraceEnabled()) { logger.trace("write: " + dataBuffer); } + + // Track write listener calls from here on.. + this.writePossible = false; + int total = buffer.remaining(); int written = writeByteBuffer(buffer); @@ -181,6 +183,10 @@ class UndertowServerHttpResponse extends AbstractListenerServerHttpResponse impl if (written != total) { return false; } + + // We wrote all, so can still write more.. + this.writePossible = true; + if (logger.isTraceEnabled()) { logger.trace("releaseData: " + dataBuffer); } @@ -239,11 +245,12 @@ class UndertowServerHttpResponse extends AbstractListenerServerHttpResponse impl @Override protected void flush() throws IOException { - if (UndertowServerHttpResponse.this.responseChannel != null) { + StreamSinkChannel channel = UndertowServerHttpResponse.this.responseChannel; + if (channel != null) { if (logger.isTraceEnabled()) { logger.trace("flush"); } - UndertowServerHttpResponse.this.responseChannel.flush(); + channel.flush(); } } @@ -255,7 +262,13 @@ class UndertowServerHttpResponse extends AbstractListenerServerHttpResponse impl @Override protected boolean isWritePossible() { - return UndertowServerHttpResponse.this.isWritePossible(); + StreamSinkChannel channel = UndertowServerHttpResponse.this.responseChannel; + if (channel != null) { + // We can always call flush, just ensure writes are on.. + channel.resumeWrites(); + return true; + } + return false; } @Override