Cancel Subscription when onError is invoked internally

AbstractResponseBodyProcessor.onError and
AbstractResponseBodyFlushProcessor.onError will be invoked when:
- The Publisher wants to signal with onError that there are failures.
Once onError is invoked the Subscription should be considered canceled.
- The internal implementation wants to signal with onError that there
are failures. In this use case the implementation should invoke
Subscription.cancel()

Author: violetagg

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

@@ -106,6 +106,10 @@ public final void subscribe(Subscriber<? super Void> subscriber) {
 	 */
 	protected abstract void flush() throws IOException;
 
+	private void cancel() {
+		this.subscription.cancel();
+	}
+
 	private void writeComplete() {
 		if (logger.isTraceEnabled()) {
 			logger.trace(this.state + " writeComplete");
@@ -157,11 +161,12 @@ public void writeComplete(AbstractResponseBodyFlushProcessor processor) {
 				else {
 					try {
 						processor.flush();
+						processor.subscription.request(1);
 					}
 					catch (IOException ex) {
+						processor.cancel();
 						processor.onError(ex);
 					}
-					processor.subscription.request(1);
 				}
 			}
 		}, COMPLETED {
@@ -231,6 +236,7 @@ public void onNext(Void aVoid) {
 
 			@Override
 			public void onError(Throwable t) {
+				processor.cancel();
 				processor.onError(t);
 			}
 

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

@@ -159,6 +159,10 @@ protected boolean isWritePossible() {
 	 */
 	protected abstract boolean write(DataBuffer dataBuffer) throws IOException;
 
+	protected void cancel() {
+		this.subscription.cancel();
+	}
+
 	private boolean changeState(State oldState, State newState) {
 		return this.state.compareAndSet(oldState, newState);
 	}
@@ -220,7 +224,6 @@ void onNext(AbstractResponseBodyProcessor processor, DataBuffer dataBuffer) {
 			@Override
 			void onComplete(AbstractResponseBodyProcessor processor) {
 				if (processor.changeState(this, COMPLETED)) {
-					processor.subscriberCompleted = true;
 					processor.publisherDelegate.publishComplete();
 				}
 			}
@@ -258,6 +261,7 @@ void onWritePossible(AbstractResponseBodyProcessor processor) {
 						}
 					}
 					catch (IOException ex) {
+						processor.cancel();
 						processor.onError(ex);
 					}
 				}

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

@@ -224,6 +224,7 @@ public void onWritePossible() throws IOException {
 		@Override
 		public void onError(Throwable ex) {
 			if (bodyProcessor != null) {
+				bodyProcessor.cancel();
 				bodyProcessor.onError(ex);
 			}
 		}