Ensure onComplete/onError events will be delivered

violetagg · rstoyanchev · commit 41b13a4e8a7d · 2017-11-21T14:42:05.000-05:00
Issue: SPR-16207
diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerReadPublisher.java b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerReadPublisher.java
@@ -312,7 +312,9 @@ <T> void request(AbstractListenerReadPublisher<T> publisher, long n) {
 		}
 
 		<T> void cancel(AbstractListenerReadPublisher<T> publisher) {
-			publisher.changeState(this, COMPLETED);
+			if (!publisher.changeState(this, COMPLETED)) {
+				publisher.state.get().cancel(publisher);
+			}
 		}
 
 		<T> void onDataAvailable(AbstractListenerReadPublisher<T> publisher) {
@@ -325,6 +327,9 @@ <T> void onAllDataRead(AbstractListenerReadPublisher<T> publisher) {
 					publisher.subscriber.onComplete();
 				}
 			}
+			else {
+				publisher.state.get().onAllDataRead(publisher);
+			}
 		}
 
 		<T> void onError(AbstractListenerReadPublisher<T> publisher, Throwable t) {
@@ -333,6 +338,9 @@ <T> void onError(AbstractListenerReadPublisher<T> publisher, Throwable t) {
 					publisher.subscriber.onError(t);
 				}
 			}
+			else {
+				publisher.state.get().onError(publisher, t);
+			}
 		}
 	}
 
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
@@ -190,6 +190,9 @@ public <T> void onComplete(AbstractListenerWriteFlushProcessor<T> processor) {
 				if (processor.changeState(this, COMPLETED)) {
 					processor.resultPublisher.publishComplete();
 				}
+				else {
+					processor.state.get().onComplete(processor);
+				}
 			}
 		},
 
@@ -212,6 +215,9 @@ public <T> void writeComplete(AbstractListenerWriteFlushProcessor<T> processor)
 					else if (processor.changeState(this, COMPLETED)) {
 						processor.resultPublisher.publishComplete();
 					}
+					else {
+						processor.state.get().onComplete(processor);
+					}
 				}
 				else {
 					if (processor.changeState(this, REQUESTED)) {
@@ -238,6 +244,9 @@ public <T> void onFlushPossible(AbstractListenerWriteFlushProcessor<T> processor
 				if (processor.changeState(this, COMPLETED)) {
 					processor.resultPublisher.publishComplete();
 				}
+				else {
+					processor.state.get().onComplete(processor);
+				}
 			}
 			public <T> void onNext(AbstractListenerWriteFlushProcessor<T> processor, Publisher<? extends T> publisher) {
 				// ignore
@@ -275,6 +284,9 @@ public <T> void onError(AbstractListenerWriteFlushProcessor<T> processor, Throwa
 			if (processor.changeState(this, COMPLETED)) {
 				processor.resultPublisher.publishError(ex);
 			}
+			else {
+				processor.state.get().onError(processor, ex);
+			}
 		}
 
 		public <T> void onComplete(AbstractListenerWriteFlushProcessor<T> processor) {
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
@@ -251,6 +251,9 @@ public <T> void onComplete(AbstractListenerWriteProcessor<T> processor) {
 					processor.writingComplete();
 					processor.resultPublisher.publishComplete();
 				}
+				else {
+					processor.state.get().onComplete(processor);
+				}
 			}
 		},
 
@@ -274,15 +277,29 @@ public <T> void onWritePossible(AbstractListenerWriteProcessor<T> processor) {
 						if (writeCompleted) {
 							processor.releaseData();
 							if (!processor.subscriberCompleted) {
-								processor.changeState(WRITING, REQUESTED);
-								processor.suspendWriting();
-								Assert.state(processor.subscription != null, "No subscription");
-								processor.subscription.request(1);
+								if (processor.changeState(WRITING, REQUESTED)) {
+									if (processor.subscriberCompleted) {
+										if (processor.changeState(REQUESTED, COMPLETED)) {
+											processor.writingComplete();
+											processor.resultPublisher.publishComplete();
+										} else {
+											processor.state.get().onComplete(processor);
+										}
+									}
+									else {
+										processor.suspendWriting();
+										Assert.state(processor.subscription != null, "No subscription");
+										processor.subscription.request(1);
+									}
+								}
 							}
 							else {
-								processor.changeState(WRITING, COMPLETED);
-								processor.writingComplete();
-								processor.resultPublisher.publishComplete();
+								if (processor.changeState(WRITING, COMPLETED)) {
+									processor.writingComplete();
+									processor.resultPublisher.publishComplete();
+								} else {
+									processor.state.get().onComplete(processor);
+								}
 							}
 						}
 						else {
@@ -343,6 +360,9 @@ public <T> void onError(AbstractListenerWriteProcessor<T> processor, Throwable e
 				processor.writingComplete();
 				processor.resultPublisher.publishError(ex);
 			}
+			else {
+				processor.state.get().onError(processor, ex);
+			}
 		}
 
 		public <T> void onComplete(AbstractListenerWriteProcessor<T> processor) {
diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/WriteResultPublisher.java b/spring-web/src/main/java/org/springframework/http/server/reactive/WriteResultPublisher.java
@@ -158,13 +158,19 @@ void publishComplete(WriteResultPublisher publisher) {
 					Assert.state(publisher.subscriber != null, "No subscriber");
 					publisher.subscriber.onComplete();
 				}
+				else {
+					publisher.state.get().publishComplete(publisher);
+				}
 			}
 			@Override
 			void publishError(WriteResultPublisher publisher, Throwable t) {
 				if (publisher.changeState(this, COMPLETED)) {
 					Assert.state(publisher.subscriber != null, "No subscriber");
 					publisher.subscriber.onError(t);
 				}
+				else {
+					publisher.state.get().publishError(publisher, t);
+				}
 			}
 		},
 
@@ -196,7 +202,9 @@ void request(WriteResultPublisher publisher, long n) {
 		}
 
 		void cancel(WriteResultPublisher publisher) {
-			publisher.changeState(this, COMPLETED);
+			if (!publisher.changeState(this, COMPLETED)) {
+				publisher.state.get().cancel(publisher);
+			}
 		}
 
 		void publishComplete(WriteResultPublisher publisher) {

Original file line number	Diff line number	Diff line change
`@@ -312,7 +312,9 @@ <T> void request(AbstractListenerReadPublisher<T> publisher, long n) {`
`312`	`312`	`}`
`313`	`313`
`314`	`314`	`<T> void cancel(AbstractListenerReadPublisher<T> publisher) {`
`315`		`- publisher.changeState(this, COMPLETED);`
	`315`	`+ if (!publisher.changeState(this, COMPLETED)) {`
	`316`	`+ publisher.state.get().cancel(publisher);`
	`317`	`+ }`
`316`	`318`	`}`
`317`	`319`
`318`	`320`	`<T> void onDataAvailable(AbstractListenerReadPublisher<T> publisher) {`
`@@ -325,6 +327,9 @@ <T> void onAllDataRead(AbstractListenerReadPublisher<T> publisher) {`
`325`	`327`	`publisher.subscriber.onComplete();`
`326`	`328`	`}`
`327`	`329`	`}`
	`330`	`+ else {`
	`331`	`+ publisher.state.get().onAllDataRead(publisher);`
	`332`	`+ }`
`328`	`333`	`}`
`329`	`334`
`330`	`335`	`<T> void onError(AbstractListenerReadPublisher<T> publisher, Throwable t) {`
`@@ -333,6 +338,9 @@ <T> void onError(AbstractListenerReadPublisher<T> publisher, Throwable t) {`
`333`	`338`	`publisher.subscriber.onError(t);`
`334`	`339`	`}`
`335`	`340`	`}`
	`341`	`+ else {`
	`342`	`+ publisher.state.get().onError(publisher, t);`
	`343`	`+ }`
`336`	`344`	`}`
`337`	`345`	`}`
`338`	`346`
Original file line number	Diff line number	Diff line change
`@@ -190,6 +190,9 @@ public <T> void onComplete(AbstractListenerWriteFlushProcessor<T> processor) {`
`190`	`190`	`if (processor.changeState(this, COMPLETED)) {`
`191`	`191`	`processor.resultPublisher.publishComplete();`
`192`	`192`	`}`
	`193`	`+ else {`
	`194`	`+ processor.state.get().onComplete(processor);`
	`195`	`+ }`
`193`	`196`	`}`
`194`	`197`	`},`
`195`	`198`
`@@ -212,6 +215,9 @@ public <T> void writeComplete(AbstractListenerWriteFlushProcessor<T> processor)`
`212`	`215`	`else if (processor.changeState(this, COMPLETED)) {`
`213`	`216`	`processor.resultPublisher.publishComplete();`
`214`	`217`	`}`
	`218`	`+ else {`
	`219`	`+ processor.state.get().onComplete(processor);`
	`220`	`+ }`
`215`	`221`	`}`
`216`	`222`	`else {`
`217`	`223`	`if (processor.changeState(this, REQUESTED)) {`
`@@ -238,6 +244,9 @@ public <T> void onFlushPossible(AbstractListenerWriteFlushProcessor<T> processor`
`238`	`244`	`if (processor.changeState(this, COMPLETED)) {`
`239`	`245`	`processor.resultPublisher.publishComplete();`
`240`	`246`	`}`
	`247`	`+ else {`
	`248`	`+ processor.state.get().onComplete(processor);`
	`249`	`+ }`
`241`	`250`	`}`
`242`	`251`	`public <T> void onNext(AbstractListenerWriteFlushProcessor<T> processor, Publisher<? extends T> publisher) {`
`243`	`252`	`// ignore`
`@@ -275,6 +284,9 @@ public <T> void onError(AbstractListenerWriteFlushProcessor<T> processor, Throwa`
`275`	`284`	`if (processor.changeState(this, COMPLETED)) {`
`276`	`285`	`processor.resultPublisher.publishError(ex);`
`277`	`286`	`}`
	`287`	`+ else {`
	`288`	`+ processor.state.get().onError(processor, ex);`
	`289`	`+ }`
`278`	`290`	`}`
`279`	`291`
`280`	`292`	`public <T> void onComplete(AbstractListenerWriteFlushProcessor<T> processor) {`
Original file line number	Diff line number	Diff line change
`@@ -158,13 +158,19 @@ void publishComplete(WriteResultPublisher publisher) {`
`158`	`158`	`Assert.state(publisher.subscriber != null, "No subscriber");`
`159`	`159`	`publisher.subscriber.onComplete();`
`160`	`160`	`}`
	`161`	`+ else {`
	`162`	`+ publisher.state.get().publishComplete(publisher);`
	`163`	`+ }`
`161`	`164`	`}`
`162`	`165`	`@Override`
`163`	`166`	`void publishError(WriteResultPublisher publisher, Throwable t) {`
`164`	`167`	`if (publisher.changeState(this, COMPLETED)) {`
`165`	`168`	`Assert.state(publisher.subscriber != null, "No subscriber");`
`166`	`169`	`publisher.subscriber.onError(t);`
`167`	`170`	`}`
	`171`	`+ else {`
	`172`	`+ publisher.state.get().publishError(publisher, t);`
	`173`	`+ }`
`168`	`174`	`}`
`169`	`175`	`},`
`170`	`176`
`@@ -196,7 +202,9 @@ void request(WriteResultPublisher publisher, long n) {`
`196`	`202`	`}`
`197`	`203`
`198`	`204`	`void cancel(WriteResultPublisher publisher) {`
`199`		`- publisher.changeState(this, COMPLETED);`
	`205`	`+ if (!publisher.changeState(this, COMPLETED)) {`
	`206`	`+ publisher.state.get().cancel(publisher);`
	`207`	`+ }`
`200`	`208`	`}`
`201`	`209`
`202`	`210`	`void publishComplete(WriteResultPublisher publisher) {`