Avoid resume-suspend race condition

This commit turns suspendReading() into a readingPaused() notification
that is invoked after a succession of reads stops because there is no
more demand. Sub-classes can use this notification to suspend, if that
applies to them.

Most importantly the notification is guaranteed not to overlap with
checkOnDataAvailable() which means that suspend does not need to be
atomic and guarded against resume. The two can and do compete all the
time when reading ends with no demand, and a request for demand arrives
concurrently.

Issue: SPR-16207

spring-webflux/src/main/java/org/springframework/web/reactive/socket/adapter/AbstractListenerWebSocketSession.java

@@ -149,17 +149,6 @@ public abstract class AbstractListenerWebSocketSession<T> extends AbstractWebSoc
 	 */
 	protected abstract void resumeReceiving();
 
-	/**
-	 * Whether receiving new message(s) is suspended.
-	 * <p><strong>Note:</strong> if the underlying WebSocket API does not provide
-	 * flow control for receiving messages, then this method as well as
-	 * {@link #canSuspendReceiving()} should both return {@code false}.
-	 * @return returns {@code true} if receiving new message(s) is suspended,
-	 * or otherwise {@code false}.
-	 * @since 5.0.2
-	 */
-	protected abstract boolean isSuspended();
-
 	/**
 	 * Send the given WebSocket message.
 	 */
@@ -231,16 +220,14 @@ public abstract class AbstractListenerWebSocketSession<T> extends AbstractWebSoc
 
 		@Override
 		protected void checkOnDataAvailable() {
-			if (isSuspended()) {
-				resumeReceiving();
-			}
+			resumeReceiving();
 			if (!this.pendingMessages.isEmpty()) {
 				onDataAvailable();
 			}
 		}
 
 		@Override
-		protected void suspendReading() {
+		protected void readingPaused() {
 			suspendReceiving();
 		}
 
@@ -250,14 +237,6 @@ public abstract class AbstractListenerWebSocketSession<T> extends AbstractWebSoc
 			return (WebSocketMessage) this.pendingMessages.poll();
 		}
 
-		@Override
-		public void onAllDataRead() {
-			if (isSuspended()) {
-				resumeReceiving();
-			}
-			super.onAllDataRead();
-		}
-
 		void handleMessage(WebSocketMessage webSocketMessage) {
 			this.pendingMessages.offer(webSocketMessage);
 			onDataAvailable();

spring-webflux/src/main/java/org/springframework/web/reactive/socket/adapter/JettyWebSocketSession.java

@@ -74,14 +74,10 @@ public class JettyWebSocketSession extends AbstractListenerWebSocketSession<Sess
 	@Override
 	protected void resumeReceiving() {
 		SuspendToken tokenToUse = this.suspendToken;
-		Assert.state(tokenToUse != null, "Not suspended");
-		tokenToUse.resume();
 		this.suspendToken = null;
-	}
-
-	@Override
-	protected boolean isSuspended() {
-		return this.suspendToken != null;
+		if (tokenToUse != null) {
+			tokenToUse.resume();
+		}
 	}
 
 	@Override

spring-webflux/src/main/java/org/springframework/web/reactive/socket/adapter/StandardWebSocketSession.java

@@ -71,11 +71,6 @@ public class StandardWebSocketSession extends AbstractListenerWebSocketSession<S
 		// no-op
 	}
 
-	@Override
-	protected boolean isSuspended() {
-		return false;
-	}
-
 	@Override
 	protected boolean sendMessage(WebSocketMessage message) throws IOException {
 		ByteBuffer buffer = message.getPayload().asByteBuffer();

spring-webflux/src/main/java/org/springframework/web/reactive/socket/adapter/TomcatWebSocketSession.java

@@ -71,9 +71,4 @@ public class TomcatWebSocketSession extends StandardWebSocketSession {
 		}
 	}
 
-	@Override
-	protected boolean isSuspended() {
-		return this.suspended == 1;
-	}
-
 }

spring-webflux/src/main/java/org/springframework/web/reactive/socket/adapter/UndertowWebSocketSession.java

@@ -71,11 +71,6 @@ public class UndertowWebSocketSession extends AbstractListenerWebSocketSession<W
 		getDelegate().resumeReceives();
 	}
 
-	@Override
-	protected boolean isSuspended() {
-		return !getDelegate().isReceivesResumed();
-	}
-
 	@Override
 	protected boolean sendMessage(WebSocketMessage message) throws IOException {
 		ByteBuffer buffer = message.getPayload().asByteBuffer();