Simple refactorings in AbstractListenerWebSocketSessionSupport

Dropped "Support" from the name since it not only provides support
methods but actually implements WebSocketSession.

Renamed inner classes:
WebSocketMessagePublisher -> WebSocketReceivePublisher
WebSocketMessageProcessor -> WebSocketSendProcessor

Add protected getter for sendProcessor.

Reduce scoping:
WebSocketReceivePublisher -> private
WebSocketSendProcessor -> protected
WebSocketSendProcessor#setReady -> public (class is still protected)

A few more method name alignments and Javadoc updates.

Issue: SPR-14527

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

@@ -18,20 +18,19 @@ package org.springframework.web.reactive.socket.adapter;
 
 import java.io.IOException;
 import java.net.URI;
-
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.reactivestreams.Publisher;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
 import org.springframework.http.server.reactive.AbstractListenerReadPublisher;
 import org.springframework.http.server.reactive.AbstractListenerWriteProcessor;
 import org.springframework.util.Assert;
 import org.springframework.web.reactive.socket.CloseStatus;
 import org.springframework.web.reactive.socket.WebSocketMessage;
-import org.springframework.web.reactive.socket.WebSocketSession;
 import org.springframework.web.reactive.socket.WebSocketMessage.Type;
-
+import org.springframework.web.reactive.socket.WebSocketSession;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
 
 /**
  * Base class for Listener-based {@link WebSocketSession} adapters.
@@ -39,7 +38,7 @@ import reactor.core.publisher.Mono;
  * @author Violeta Georgieva
  * @since 5.0
  */
-public abstract class AbstractListenerWebSocketSessionSupport<T> extends WebSocketSessionSupport<T> {
+public abstract class AbstractListenerWebSocketSession<T> extends WebSocketSessionSupport<T> {
 
 	private final AtomicBoolean sendCalled = new AtomicBoolean();
 
@@ -47,12 +46,12 @@ public abstract class AbstractListenerWebSocketSessionSupport<T> extends WebSock
 
 	private final URI uri;
 
-	protected final WebSocketMessagePublisher webSocketMessagePublisher =
+	private final WebSocketReceivePublisher receivePublisher = new WebSocketReceivePublisher();
-			new WebSocketMessagePublisher();
 
-	protected volatile WebSocketMessageProcessor webSocketMessageProcessor;
+	private volatile WebSocketSendProcessor sendProcessor;
 
-	public AbstractListenerWebSocketSessionSupport(T delegate, String id, URI uri) {
+
+	public AbstractListenerWebSocketSession(T delegate, String id, URI uri) {
 		super(delegate);
 		Assert.notNull(id, "'id' is required.");
 		Assert.notNull(uri, "'uri' is required.");
@@ -60,6 +59,7 @@ public abstract class AbstractListenerWebSocketSessionSupport<T> extends WebSock
 		this.uri = uri;
 	}
 
+
 	@Override
 	public String getId() {
 		return this.id;
@@ -70,18 +70,22 @@ public abstract class AbstractListenerWebSocketSessionSupport<T> extends WebSock
 		return this.uri;
 	}
 
+	protected WebSocketSendProcessor getSendProcessor() {
+		return this.sendProcessor;
+	}
+
 	@Override
 	public Flux<WebSocketMessage> receive() {
-		return Flux.from(this.webSocketMessagePublisher);
+		return Flux.from(this.receivePublisher);
 	}
 
 	@Override
 	public Mono<Void> send(Publisher<WebSocketMessage> messages) {
 		if (this.sendCalled.compareAndSet(false, true)) {
-			this.webSocketMessageProcessor = new WebSocketMessageProcessor();
+			this.sendProcessor = new WebSocketSendProcessor();
 			return Mono.from(subscriber -> {
-					messages.subscribe(this.webSocketMessageProcessor);
+					messages.subscribe(this.sendProcessor);
-					this.webSocketMessageProcessor.subscribe(subscriber);
+					this.sendProcessor.subscribe(subscriber);
 			});
 		}
 		else {
@@ -97,32 +101,38 @@ public abstract class AbstractListenerWebSocketSessionSupport<T> extends WebSock
 		// no-op
 	}
 
-	protected abstract boolean writeInternal(WebSocketMessage message) throws IOException;
+	protected boolean isReadyToReceive() {
+		return this.receivePublisher.isReadyToReceive();
+	}
 
-	/** Handle a message callback from the Servlet container */
+	protected abstract boolean sendMessage(WebSocketMessage message) throws IOException;
+
+	/** Handle a message callback from the WebSocketHandler adapter */
 	void handleMessage(Type type, WebSocketMessage message) {
-		this.webSocketMessagePublisher.processWebSocketMessage(message);
+		this.receivePublisher.handleMessage(message);
 	}
 
-	/** Handle a error callback from the Servlet container */
+	/** Handle an error callback from the WebSocketHandler adapter */
 	void handleError(Throwable ex) {
-		this.webSocketMessagePublisher.onError(ex);
+		this.receivePublisher.onError(ex);
-		if (this.webSocketMessageProcessor != null) {
+		if (this.sendProcessor != null) {
-			this.webSocketMessageProcessor.cancel();
+			this.sendProcessor.cancel();
-			this.webSocketMessageProcessor.onError(ex);
+			this.sendProcessor.onError(ex);
 		}
 	}
 
-	/** Handle a complete callback from the Servlet container */
+	/** Handle a close callback from the WebSocketHandler adapter */
 	void handleClose(CloseStatus reason) {
-		this.webSocketMessagePublisher.onAllDataRead();
+		this.receivePublisher.onAllDataRead();
-		if (this.webSocketMessageProcessor != null) {
+		if (this.sendProcessor != null) {
-			this.webSocketMessageProcessor.cancel();
+			this.sendProcessor.cancel();
-			this.webSocketMessageProcessor.onComplete();
+			this.sendProcessor.onComplete();
 		}
 	}
 
-	final class WebSocketMessagePublisher extends AbstractListenerReadPublisher<WebSocketMessage> {
+
+	private final class WebSocketReceivePublisher extends AbstractListenerReadPublisher<WebSocketMessage> {
+
 		private volatile WebSocketMessage webSocketMessage;
 
 		@Override
@@ -144,52 +154,47 @@ public abstract class AbstractListenerWebSocketSessionSupport<T> extends WebSock
 			return null;
 		}
 
-		void processWebSocketMessage(WebSocketMessage webSocketMessage) {
+		void handleMessage(WebSocketMessage webSocketMessage) {
 			this.webSocketMessage = webSocketMessage;
 			suspendReceives();
 			onDataAvailable();
 		}
 
-		boolean canAccept() {
+		boolean isReadyToReceive() {
 			return this.webSocketMessage == null;
 		}
 	}
 
-	final class WebSocketMessageProcessor extends AbstractListenerWriteProcessor<WebSocketMessage> {
+	protected final class WebSocketSendProcessor extends AbstractListenerWriteProcessor<WebSocketMessage> {
+
 		private volatile boolean isReady = true;
 
 		@Override
 		protected boolean write(WebSocketMessage message) throws IOException {
-			return writeInternal(message);
+			return sendMessage(message);
 		}
 
 		@Override
 		protected void releaseData() {
 			if (logger.isTraceEnabled()) {
-				logger.trace("releaseBuffer: " + this.currentData);
+				logger.trace("releaseData: " + this.currentData);
 			}
 			this.currentData = null;
 		}
 
 		@Override
-		protected boolean isDataEmpty(WebSocketMessage data) {
+		protected boolean isDataEmpty(WebSocketMessage message) {
-			return data.getPayload().readableByteCount() == 0;
+			return message.getPayload().readableByteCount() == 0;
 		}
 
 		@Override
 		protected boolean isWritePossible() {
-			if (this.isReady && this.currentData != null) {
+			return this.isReady && this.currentData != null;
-				return true;
-			}
-			else {
-				return false;
-			}
 		}
 
-		void setReady(boolean ready) {
+		public void setReady(boolean ready) {
 			this.isReady = ready;
 		}
-
 	}
 
-}
+}

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

@@ -35,7 +35,7 @@ import reactor.core.publisher.Mono;
  * @author Violeta Georgieva
  * @since 5.0
  */
-public class JettyWebSocketSession extends AbstractListenerWebSocketSessionSupport<Session> {
+public class JettyWebSocketSession extends AbstractListenerWebSocketSession<Session> {
 
 	public JettyWebSocketSession(Session session) {
 		super(session, ObjectUtils.getIdentityHexString(session),
@@ -49,15 +49,15 @@ public class JettyWebSocketSession extends AbstractListenerWebSocketSessionSuppo
 	}
 
 	@Override
-	protected boolean writeInternal(WebSocketMessage message) throws IOException {
+	protected boolean sendMessage(WebSocketMessage message) throws IOException {
 		if (WebSocketMessage.Type.TEXT.equals(message.getType())) {
-			this.webSocketMessageProcessor.setReady(false);
+			getSendProcessor().setReady(false);
 			getDelegate().getRemote().sendString(
 					new String(message.getPayload().asByteBuffer().array(), StandardCharsets.UTF_8),
 					new WebSocketMessageWriteCallback());
 		}
 		else if (WebSocketMessage.Type.BINARY.equals(message.getType())) {
-			this.webSocketMessageProcessor.setReady(false);
+			getSendProcessor().setReady(false);
 			getDelegate().getRemote().sendBytes(message.getPayload().asByteBuffer(),
 					new WebSocketMessageWriteCallback());
 		}
@@ -77,14 +77,14 @@ public class JettyWebSocketSession extends AbstractListenerWebSocketSessionSuppo
 
 		@Override
 		public void writeFailed(Throwable x) {
-			webSocketMessageProcessor.cancel();
+			getSendProcessor().cancel();
-			webSocketMessageProcessor.onError(x);
+			getSendProcessor().onError(x);
 		}
 
 		@Override
 		public void writeSuccess() {
-			webSocketMessageProcessor.setReady(true);
+			getSendProcessor().setReady(true);
-			webSocketMessageProcessor.onWritePossible();
+			getSendProcessor().onWritePossible();
 		}
 
 	}

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

@@ -65,7 +65,7 @@ public class TomcatWebSocketHandlerAdapter extends Endpoint {
 			@Override
 			public void onMessage(String message) {
 				while (true) {
-					if (wsSession.canWebSocketMessagePublisherAccept()) {
+					if (wsSession.isReadyToReceive()) {
 						WebSocketMessage wsMessage = toMessage(message);
 						wsSession.handleMessage(wsMessage.getType(), wsMessage);
 						break;
@@ -79,7 +79,7 @@ public class TomcatWebSocketHandlerAdapter extends Endpoint {
 			@Override
 			public void onMessage(ByteBuffer message) {
 				while (true) {
-					if (wsSession.canWebSocketMessagePublisherAccept()) {
+					if (wsSession.isReadyToReceive()) {
 						WebSocketMessage wsMessage = toMessage(message);
 						wsSession.handleMessage(wsMessage.getType(), wsMessage);
 						break;
@@ -93,7 +93,7 @@ public class TomcatWebSocketHandlerAdapter extends Endpoint {
 			@Override
 			public void onMessage(PongMessage message) {
 				while (true) {
-					if (wsSession.canWebSocketMessagePublisherAccept()) {
+					if (wsSession.isReadyToReceive()) {
 						WebSocketMessage wsMessage = toMessage(message);
 						wsSession.handleMessage(wsMessage.getType(), wsMessage);
 						break;

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

@@ -38,7 +38,7 @@ import reactor.core.publisher.Mono;
  * @author Violeta Georgieva
  * @since 5.0
  */
-public class TomcatWebSocketSession extends AbstractListenerWebSocketSessionSupport<Session> {
+public class TomcatWebSocketSession extends AbstractListenerWebSocketSession<Session> {
 
 	public TomcatWebSocketSession(Session session) {
 		super(session, session.getId(), session.getRequestURI());
@@ -56,20 +56,16 @@ public class TomcatWebSocketSession extends AbstractListenerWebSocketSessionSupp
 		return Mono.empty();
 	}
 
-	boolean canWebSocketMessagePublisherAccept() {
-		return this.webSocketMessagePublisher.canAccept();
-	}
-
 	@Override
-	protected boolean writeInternal(WebSocketMessage message) throws IOException {
+	protected boolean sendMessage(WebSocketMessage message) throws IOException {
 		if (WebSocketMessage.Type.TEXT.equals(message.getType())) {
-			this.webSocketMessageProcessor.setReady(false);
+			getSendProcessor().setReady(false);
 			getDelegate().getAsyncRemote().sendText(
 					new String(message.getPayload().asByteBuffer().array(), StandardCharsets.UTF_8),
 					new WebSocketMessageSendHandler());
 		}
 		else if (WebSocketMessage.Type.BINARY.equals(message.getType())) {
-			this.webSocketMessageProcessor.setReady(false);
+			getSendProcessor().setReady(false);
 			getDelegate().getAsyncRemote().sendBinary(message.getPayload().asByteBuffer(),
 					new WebSocketMessageSendHandler());
 		}
@@ -90,12 +86,12 @@ public class TomcatWebSocketSession extends AbstractListenerWebSocketSessionSupp
 		@Override
 		public void onResult(SendResult result) {
 			if (result.isOK()) {
-				webSocketMessageProcessor.setReady(true);
+				getSendProcessor().setReady(true);
-				webSocketMessageProcessor.onWritePossible();
+				getSendProcessor().onWritePossible();
 			}
 			else {
-				webSocketMessageProcessor.cancel();
+				getSendProcessor().cancel();
-				webSocketMessageProcessor.onError(result.getException());
+				getSendProcessor().onError(result.getException());
 			}
 		}
 

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

@@ -39,7 +39,7 @@ import reactor.core.publisher.Mono;
  * @author Violeta Georgieva
  * @since 5.0
  */
-public class UndertowWebSocketSession extends AbstractListenerWebSocketSessionSupport<WebSocketChannel> {
+public class UndertowWebSocketSession extends AbstractListenerWebSocketSession<WebSocketChannel> {
 
 	public UndertowWebSocketSession(WebSocketChannel channel) throws URISyntaxException {
 		super(channel, ObjectUtils.getIdentityHexString(channel), new URI(channel.getUrl()));
@@ -63,25 +63,25 @@ public class UndertowWebSocketSession extends AbstractListenerWebSocketSessionSu
 	}
 
 	@Override
-	protected boolean writeInternal(WebSocketMessage message) throws IOException {
+	protected boolean sendMessage(WebSocketMessage message) throws IOException {
 		if (WebSocketMessage.Type.TEXT.equals(message.getType())) {
-			this.webSocketMessageProcessor.setReady(false);
+			getSendProcessor().setReady(false);
 			WebSockets.sendText(
 					new String(message.getPayload().asByteBuffer().array(), StandardCharsets.UTF_8),
 					getDelegate(), new WebSocketMessageSendHandler());
 		}
 		else if (WebSocketMessage.Type.BINARY.equals(message.getType())) {
-			this.webSocketMessageProcessor.setReady(false);
+			getSendProcessor().setReady(false);
 			WebSockets.sendBinary(message.getPayload().asByteBuffer(),
 					getDelegate(), new WebSocketMessageSendHandler());
 		}
 		else if (WebSocketMessage.Type.PING.equals(message.getType())) {
-			this.webSocketMessageProcessor.setReady(false);
+			getSendProcessor().setReady(false);
 			WebSockets.sendPing(message.getPayload().asByteBuffer(),
 					getDelegate(), new WebSocketMessageSendHandler());
 		}
 		else if (WebSocketMessage.Type.PONG.equals(message.getType())) {
-			this.webSocketMessageProcessor.setReady(false);
+			getSendProcessor().setReady(false);
 			WebSockets.sendPong(message.getPayload().asByteBuffer(),
 					getDelegate(), new WebSocketMessageSendHandler());
 		}
@@ -95,15 +95,15 @@ public class UndertowWebSocketSession extends AbstractListenerWebSocketSessionSu
 
 		@Override
 		public void complete(WebSocketChannel channel, Void context) {
-			webSocketMessageProcessor.setReady(true);
+			getSendProcessor().setReady(true);
-			webSocketMessageProcessor.onWritePossible();
+			getSendProcessor().onWritePossible();
 		}
 
 		@Override
 		public void onError(WebSocketChannel channel, Void context,
 				Throwable throwable) {
-			webSocketMessageProcessor.cancel();
+			getSendProcessor().cancel();
-			webSocketMessageProcessor.onError(throwable);
+			getSendProcessor().onError(throwable);
 		}
 
 	}