Add committed flag to ReactiveHttpOutputMessage

Issue: SPR-15135

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}.
  *
- * <p>Typically implemented by an HTTP request on the client-side or a response
- * on the server-side.
+ * <p>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<? extends Mono<Void>> 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<Void> writeWith(Publisher<? extends DataBuffer> 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<DataBuffer>}.
+	 * 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<DataBuffer>}.
 	 * @param body the body content publisher
 	 * @return a {@link Mono} that indicates completion or error
 	 */

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<? extends Mono<Void>> 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<Void> 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<Void> doCommit(Supplier<? extends Mono<Void>> 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<? extends Mono<Void>> 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.

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<Void> writeWith(Publisher<? extends DataBuffer> body) {
 		return new ChannelSendOperator<>(body,
@@ -151,7 +156,7 @@ public abstract class AbstractServerHttpResponse implements ServerHttpResponse {
 
 	@Override
 	public Mono<Void> setComplete() {
-		return doCommit();
+		return doCommit(null);
 	}
 
 	/**
@@ -159,7 +164,7 @@ public abstract class AbstractServerHttpResponse implements ServerHttpResponse {
 	 * @return a completion publisher
 	 */
 	protected Mono<Void> doCommit() {
-		return (this.state.get() == State.NEW ? doCommit(null) : Mono.empty());
+		return doCommit(null);
 	}
 
 	/**

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<Void> writeWith(Publisher<? extends DataBuffer> body) {
 		return getDelegate().writeWith(body);