UndertowServerHttpRequest: suspend reading when no demand

- When there is no demand for reading, the implementation
should suspend reading otherwise useless events will be
send by Undertow to the registered read listener.
- There is not need to wait for an event for reading/writing after calling
resumeReads/resumeWrites

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

@@ -113,6 +113,12 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 	@Nullable
 	protected abstract T read() throws IOException;
 
+	/**
+	 * Suspend reading. Defaults to no-op.
+	 */
+	protected void suspendReading() {
+	}
+
 
 	/**
 	 * Read and publish data from the input. Continue till there is no more
@@ -256,6 +262,7 @@ public abstract class AbstractListenerReadPublisher<T> implements Publisher<T> {
 						}
 						else {
 							publisher.changeState(READING, NO_DEMAND);
+							publisher.suspendReading();
 						}
 					}
 					catch (IOException ex) {

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

@@ -152,7 +152,28 @@ class UndertowServerHttpRequest extends AbstractServerHttpRequest {
 		protected void checkOnDataAvailable() {
 			// TODO: The onDataAvailable() call below can cause a StackOverflowError
 			// since this method is being called from onDataAvailable() itself.
-			onDataAvailable();
+			if (isReadPossible()) {
+				onDataAvailable();
+			}
+		}
+
+		private boolean isReadPossible() {
+			if (!this.channel.isReadResumed()) {
+				this.channel.resumeReads();
+			}
+			return this.channel.isReadResumed();
+		}
+
+		@Override
+		protected void suspendReading() {
+			this.channel.suspendReads();
+		}
+
+		@Override
+		public void onAllDataRead() {
+			this.channel.getReadSetter().set(null);
+			this.channel.resumeReads();
+			super.onAllDataRead();
 		}
 
 		@Override

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

@@ -152,12 +152,10 @@ class UndertowServerHttpResponse extends AbstractListenerServerHttpResponse impl
 		if (this.responseChannel == null) {
 			this.responseChannel = this.exchange.getResponseChannel();
 		}
-		if (this.responseChannel.isWriteResumed()) {
-			return true;
-		} else {
+		if (!this.responseChannel.isWriteResumed()) {
 			this.responseChannel.resumeWrites();
-			return false;
 		}
+		return this.responseChannel.isWriteResumed();
 	}