diff --git a/spring-web/src/main/java/org/springframework/http/ReactiveHttpOutputMessage.java b/spring-web/src/main/java/org/springframework/http/ReactiveHttpOutputMessage.java index e0f61a4231..0520faa635 100644 --- a/spring-web/src/main/java/org/springframework/http/ReactiveHttpOutputMessage.java +++ b/spring-web/src/main/java/org/springframework/http/ReactiveHttpOutputMessage.java @@ -27,8 +27,8 @@ import org.springframework.core.io.buffer.DataBufferFactory; /** * A "reactive" HTTP output message that accepts output as a {@link Publisher}. * - *

Typically implemented by an HTTP request on the client-side or a response - * on the server-side. + *

Typically implemented by an HTTP request on the client-side or an + * HTTP response on the server-side. * * @author Arjen Poutsma * @author Sebastien Deleuze @@ -37,30 +37,35 @@ import org.springframework.core.io.buffer.DataBufferFactory; public interface ReactiveHttpOutputMessage extends HttpMessage { /** - * Return a {@link DataBufferFactory} that can be used for creating the body. + * Return a {@link DataBufferFactory} that can be used to create the body. * @return a buffer factory * @see #writeWith(Publisher) */ DataBufferFactory bufferFactory(); /** - * Register an action to be applied just before the message is committed. - * @param action the action + * Register an action to apply just before the HttpOutputMessage is committed. + * @param action the action to apply */ void beforeCommit(Supplier> action); /** - * Use the given {@link Publisher} to write the body of the message to the underlying - * HTTP layer. + * Whether the HttpOutputMessage is committed. + */ + boolean isCommitted(); + + /** + * Use the given {@link Publisher} to write the body of the message to the + * underlying HTTP layer. * @param body the body content publisher * @return a {@link Mono} that indicates completion or error */ Mono writeWith(Publisher body); /** - * Use the given {@link Publisher} of {@code Publishers} to write the body of the - * message to the underlying HTTP layer, flushing after each - * {@code Publisher}. + * Use the given {@link Publisher} of {@code Publishers} to write the body + * of the HttpOutputMessage to the underlying HTTP layer, flushing after + * each {@code Publisher}. * @param body the body content publisher * @return a {@link Mono} that indicates completion or error */ diff --git a/spring-web/src/main/java/org/springframework/http/client/reactive/AbstractClientHttpRequest.java b/spring-web/src/main/java/org/springframework/http/client/reactive/AbstractClientHttpRequest.java index c76d0bc786..765ad4584d 100644 --- a/spring-web/src/main/java/org/springframework/http/client/reactive/AbstractClientHttpRequest.java +++ b/spring-web/src/main/java/org/springframework/http/client/reactive/AbstractClientHttpRequest.java @@ -86,12 +86,23 @@ public abstract class AbstractClientHttpRequest implements ClientHttpRequest { return this.cookies; } + @Override + public void beforeCommit(Supplier> action) { + Assert.notNull(action, "Action must not be null"); + this.commitActions.add(action); + } + + @Override + public boolean isCommitted() { + return this.state.get() != State.NEW; + } + /** * A variant of {@link #doCommit(Supplier)} for a request without body. * @return a completion publisher */ protected Mono doCommit() { - return (this.state.get() == State.NEW ? doCommit(null) : Mono.empty()); + return doCommit(null); } /** @@ -102,14 +113,14 @@ public abstract class AbstractClientHttpRequest implements ClientHttpRequest { */ protected Mono doCommit(Supplier> writeAction) { - if (!this.state.compareAndSet(AbstractClientHttpRequest.State.NEW, AbstractClientHttpRequest.State.COMMITTING)) { + if (!this.state.compareAndSet(State.NEW, State.COMMITTING)) { return Mono.empty(); } this.commitActions.add(() -> { applyHeaders(); applyCookies(); - this.state.set(AbstractClientHttpRequest.State.COMMITTED); + this.state.set(State.COMMITTED); return Mono.empty(); }); @@ -123,12 +134,6 @@ public abstract class AbstractClientHttpRequest implements ClientHttpRequest { return Flux.concat(actions).next(); } - @Override - public void beforeCommit(Supplier> action) { - Assert.notNull(action, "Action must not be null"); - this.commitActions.add(action); - } - /** * Implement this method to apply header changes from {@link #getHeaders()} * to the underlying response. This method is called once only. diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractServerHttpResponse.java b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractServerHttpResponse.java index 91e6d7dd79..3ede8a5627 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractServerHttpResponse.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractServerHttpResponse.java @@ -54,7 +54,7 @@ public abstract class AbstractServerHttpResponse implements ServerHttpResponse { * response during which time pre-commit actions can still make changes to * the response status and headers. */ - private enum State {NEW, COMMITTING, COMMITTED}; + private enum State {NEW, COMMITTING, COMMITTED} private final Log logger = LogFactory.getLog(getClass()); @@ -137,6 +137,11 @@ public abstract class AbstractServerHttpResponse implements ServerHttpResponse { } } + @Override + public boolean isCommitted() { + return this.state.get() != State.NEW; + } + @Override public final Mono writeWith(Publisher body) { return new ChannelSendOperator<>(body, @@ -151,7 +156,7 @@ public abstract class AbstractServerHttpResponse implements ServerHttpResponse { @Override public Mono setComplete() { - return doCommit(); + return doCommit(null); } /** @@ -159,7 +164,7 @@ public abstract class AbstractServerHttpResponse implements ServerHttpResponse { * @return a completion publisher */ protected Mono doCommit() { - return (this.state.get() == State.NEW ? doCommit(null) : Mono.empty()); + return doCommit(null); } /** diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/ServerHttpResponseDecorator.java b/spring-web/src/main/java/org/springframework/http/server/reactive/ServerHttpResponseDecorator.java index 7e4b785db6..0450ceb3a6 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/ServerHttpResponseDecorator.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/ServerHttpResponseDecorator.java @@ -94,6 +94,11 @@ public class ServerHttpResponseDecorator implements ServerHttpResponse { getDelegate().beforeCommit(action); } + @Override + public boolean isCommitted() { + return getDelegate().isCommitted(); + } + @Override public Mono writeWith(Publisher body) { return getDelegate().writeWith(body);