Fix race conditions in AbstractListenerReadPublisher

Transition from DEMAND->NO_DEMAND:
Two concurrent threads enter DEMAND.request and DEMAND.onDataAvailable.
And DEMAND.onDataAvailable finishes before DEMAND.request to be able to
update the demand field then a request for reading will be lost.

Transition from READING->NO_DEMAND:
readAndPublish() returns false because there is no demand but before
switching the states READING.request is invoked again a request for
reading will be lost.

Changing READING->DEMAND/NO_DEMAND is made conditional so that the
operations will be executed only if changing states succeeds.

When in READING state detect completion before each next item in order
to exit sooner, if completed.

Issue: SPR-16207

Author: violetagg

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

@@ -132,7 +132,7 @@ protected void suspendReading() {
 	 */
 	private boolean readAndPublish() throws IOException {
 		long r;
-		while ((r = demand) > 0) {
+		while ((r = demand) > 0 && !publisherCompleted) {
 			T data = read();
 			if (data != null) {
 				if (r != Long.MAX_VALUE) {
@@ -292,27 +292,45 @@ <T> void request(AbstractListenerReadPublisher<T> publisher, long n) {
 			<T> void request(AbstractListenerReadPublisher<T> publisher, long n) {
 				if (Operators.validate(n)) {
 					Operators.addCap(DEMAND_FIELD_UPDATER, publisher, n);
+					if (publisher.changeState(NO_DEMAND, DEMAND)) {
+						publisher.checkOnDataAvailable();
+					}
 				}
 			}
 
 			@Override
 			<T> void onDataAvailable(AbstractListenerReadPublisher<T> publisher) {
+				for (;;) {
+					if (!read(publisher)) {
+						return;
+					}
+					long r = publisher.demand;
+					if (r == 0  || publisher.changeState(NO_DEMAND, this)) {
+						break;
+					}
+				}
+			}
+
+			<T> boolean read(AbstractListenerReadPublisher<T> publisher) {
 				if (publisher.changeState(this, READING)) {
 					try {
 						boolean demandAvailable = publisher.readAndPublish();
 						if (demandAvailable) {
-							publisher.changeState(READING, DEMAND);
-							publisher.checkOnDataAvailable();
+							if (publisher.changeState(READING, DEMAND)) {
+								publisher.checkOnDataAvailable();
+								return false;
+							}
 						}
-						else {
-							publisher.changeState(READING, NO_DEMAND);
+						else if (publisher.changeState(READING, NO_DEMAND)) {
 							publisher.suspendReading();
 						}
 					}
 					catch (IOException ex) {
 						publisher.onError(ex);
 					}
+					return true;
 				}
+				return false;
 			}
 		},
 
@@ -321,6 +339,9 @@ <T> void onDataAvailable(AbstractListenerReadPublisher<T> publisher) {
 			<T> void request(AbstractListenerReadPublisher<T> publisher, long n) {
 				if (Operators.validate(n)) {
 					Operators.addCap(DEMAND_FIELD_UPDATER, publisher, n);
+					if (publisher.changeState(NO_DEMAND, DEMAND)) {
+						publisher.checkOnDataAvailable();
+					}
 				}
 			}
 		},
@@ -356,7 +377,10 @@ <T> void request(AbstractListenerReadPublisher<T> publisher, long n) {
 		}
 
 		<T> void cancel(AbstractListenerReadPublisher<T> publisher) {
-			if (!publisher.changeState(this, COMPLETED)) {
+			if (publisher.changeState(this, COMPLETED)) {
+				publisher.publisherCompleted = true;
+			}
+			else {
 				publisher.state.get().cancel(publisher);
 			}
 		}
@@ -367,6 +391,7 @@ <T> void onDataAvailable(AbstractListenerReadPublisher<T> publisher) {
 
 		<T> void onAllDataRead(AbstractListenerReadPublisher<T> publisher) {
 			if (publisher.changeState(this, COMPLETED)) {
+				publisher.publisherCompleted = true;
 				if (publisher.subscriber != null) {
 					publisher.subscriber.onComplete();
 				}