Minor refactoring + polish reactive WebSocket support

Rename classes not specific to Tomcat:
TomcatWebSocketSession -> StandardWebSocketSession
TomcatWebSocketHandlerAdapter -> StandardWebSocketHandlerAdapter

WebSocketSessionSupport is renamed to AbstractWebSocketSession since it
actually is a WebSocketSession and pre-implements a number of methods.

ServerEndpointRegistration is now package private (mainly for use in
upgrade strategies) and renamed to DefaultServerEndpointConfig.

Author: rstoyanchev

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

@@ -23,11 +23,10 @@
  * Representation of WebSocket "close" status codes and reasons. Status codes
  * in the 1xxx range are pre-defined by the protocol.
  *
- * <p>See <a href="https://tools.ietf.org/html/rfc6455#section-7.4.1">
- *     RFC 6455, Section 7.4.1 "Defined Status Codes"</a>.
- *
  * @author Rossen Stoyanchev
  * @since 5.0
+ * @see <a href="https://tools.ietf.org/html/rfc6455#section-7.4.1">
+ *     RFC 6455, Section 7.4.1 "Defined Status Codes"</a>
  */
 public final class CloseStatus {
 

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

@@ -24,42 +24,54 @@
 import org.springframework.util.Assert;
 
 /**
- * Simple container of information from a WebSocket handshake request.
+ * Simple container of information related to the handshake request that started
+ * the {@link WebSocketSession} session.
  *
  * @author Rossen Stoyanchev
  * @since 5.0
+ * @see WebSocketSession#getHandshakeInfo()
  */
 public class HandshakeInfo {
 
 	private final URI uri;
 
 	private final HttpHeaders headers;
 
-	private final Mono<Principal> principal;
+	private final Mono<Principal> principalMono;
 
 
-	public HandshakeInfo(URI uri, HttpHeaders headers, Mono<Principal> principal) {
+	public HandshakeInfo(URI uri, HttpHeaders headers, Mono<Principal> principalMono) {
 		Assert.notNull(uri, "URI is required.");
 		Assert.notNull(headers, "HttpHeaders are required.");
-		Assert.notNull(principal, "Prinicpal is required.");
+		Assert.notNull(principalMono, "Principal is required.");
 		this.uri = uri;
 		this.headers = headers;
-		this.principal = principal;
+		this.principalMono = principalMono;
 	}
 
 
+	/**
+	 * Return the URL for the WebSocket endpoint.
+	 */
 	public URI getUri() {
 		return this.uri;
 	}
 
+	/**
+	 * Return the headers from the handshake HTTP request.
+	 */
 	public HttpHeaders getHeaders() {
 		return this.headers;
 	}
 
+	/**
+	 * Return the principal associated with the handshake HTTP request, if any.
+	 */
 	public Mono<Principal> getPrincipal() {
-		return this.principal;
+		return this.principalMono;
 	}
 
+
 	@Override
 	public String toString() {
 		return "HandshakeInfo[uri=" + this.uri + ", headers=" + this.headers + "]";

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

@@ -21,7 +21,7 @@
 import reactor.core.publisher.Mono;
 
 /**
- * Handler for a WebSocket-style session interaction.
+ * Handler for a WebSocket session.
  *
  * @author Rossen Stoyanchev
  * @since 5.0
@@ -37,9 +37,10 @@ default List<String> getSubProtocols() {
 	}
 
 	/**
-	 * Handle the given WebSocket session.
-	 * @param session the session
-	 * @return signals completion for session handling
+	 * Handle the WebSocket session.
+	 * @param session the session to handle
+	 * @return completion {@code Mono<Void>} to indicate the outcome of the
+	 * WebSocket session handling.
 	 */
 	Mono<Void> handle(WebSocketSession session);
 

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

@@ -24,7 +24,9 @@
 
 /**
  * Representation of a WebSocket message.
- * Use one of the static factory methods in this class to create a message.
+ * <p>See static factory methods in {@link WebSocketSession} for creating messages
+ * with the {@link org.springframework.core.io.buffer.DataBufferFactory
+ * DataBufferFactory} for the session.
  *
  * @author Rossen Stoyanchev
  * @since 5.0
@@ -37,15 +39,10 @@ public class WebSocketMessage {
 
 
 	/**
-	 * Constructor for a WebSocketMessage. To create, see factory methods:
-	 * <ul>
-	 * <li>{@link WebSocketSession#textMessage}
-	 * <li>{@link WebSocketSession#binaryMessage}
-	 * <li>{@link WebSocketSession#pingMessage}
-	 * <li>{@link WebSocketSession#pongMessage}
-	 * </ul>
-	 * <p>Alternatively use {@link WebSocketSession#bufferFactory()} to create
-	 * the payload and then invoke this constructor.
+	 * Constructor for a WebSocketMessage.
+	 * <p>See static factory methods in {@link WebSocketSession} or alternatively
+	 * use {@link WebSocketSession#bufferFactory()} to create the payload and
+	 * then invoke this constructor.
 	 */
 	public WebSocketMessage(Type type, DataBuffer payload) {
 		Assert.notNull(type, "'type' must not be null");
@@ -81,7 +78,7 @@ public String getPayloadAsText() {
 
 	/**
 	 * Retain the data buffer for the message payload, which is useful on
-	 * runtimes with pooled buffers, e.g. Netty. A shortcut for:
+	 * runtimes (e.g. Netty) with pooled buffers. A shortcut for:
 	 * <pre>
 	 * DataBuffer payload = message.getPayload();
 	 * DataBufferUtils.retain(payload);
@@ -94,8 +91,8 @@ public WebSocketMessage retain() {
 	}
 
 	/**
-	 * Release the payload {@code DataBuffer} which is useful on runtimes with
-	 * pooled buffers such as Netty. Effectively a shortcut for:
+	 * Release the payload {@code DataBuffer} which is useful on runtimes
+	 * (e.g. Netty) with pooled buffers such as Netty. A shortcut for:
 	 * <pre>
 	 * DataBuffer payload = message.getPayload();
 	 * DataBufferUtils.release(payload);

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

@@ -25,7 +25,15 @@
 import org.springframework.core.io.buffer.DataBufferFactory;
 
 /**
- * Representation for a WebSocket session.
+ * Represents a WebSocket session with Reactive Streams input and output.
+ *
+ * <p>On the server side a WebSocket session can be handled by mapping
+ * requests to a {@link WebSocketHandler} and ensuring there is a
+ * {@link org.springframework.web.reactive.socket.server.support.WebSocketHandlerAdapter
+ * WebSocketHandlerAdapter} strategy registered in Spring configuration.
+ * On the client side a {@link WebSocketHandler} can be provided to a
+ * {@link org.springframework.web.reactive.socket.client.WebSocketClient
+ * WebSocketClient}.
  *
  * @author Rossen Stoyanchev
  * @since 5.0

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

@@ -33,13 +33,15 @@
 import org.springframework.web.reactive.socket.WebSocketSession;
 
 /**
- * Base class for Listener-based {@link WebSocketSession} adapters.
+ * Base class for {@link WebSocketSession} implementations that bridge between
+ * event-listener WebSocket APIs (e.g. Java WebSocket API JSR-356, Jetty,
+ * Undertow) and Reactive Streams.
  *
  * @author Violeta Georgieva
  * @author Rossen Stoyanchev
  * @since 5.0
  */
-public abstract class AbstractListenerWebSocketSession<T> extends WebSocketSessionSupport<T> {
+public abstract class AbstractListenerWebSocketSession<T> extends AbstractWebSocketSession<T> {
 
 	/**
 	 * The "back-pressure" buffer size to use if the underlying WebSocket API

spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/adapter/WebSocketSessionSupport.java renamed to spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/adapter/AbstractWebSocketSession.java

@@ -28,20 +28,19 @@
 import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.util.Assert;
-import org.springframework.web.reactive.socket.CloseStatus;
 import org.springframework.web.reactive.socket.HandshakeInfo;
 import org.springframework.web.reactive.socket.WebSocketMessage;
 import org.springframework.web.reactive.socket.WebSocketSession;
 
 /**
- * Base class for {@link WebSocketSession} implementations wrapping and
- * delegating to the native WebSocket session (or connection) of the underlying
- * WebSocket runtime.
+ * Convenient base class for {@link WebSocketSession} implementations that
+ * holds common fields and exposes accessors. Also implements the
+ * {@code WebSocketMessage} factory methods.
  *
  * @author Rossen Stoyanchev
  * @since 5.0
  */
-public abstract class WebSocketSessionSupport<T> implements WebSocketSession {
+public abstract class AbstractWebSocketSession<T> implements WebSocketSession {
 
 	protected final Log logger = LogFactory.getLog(getClass());
 
@@ -58,7 +57,7 @@ public abstract class WebSocketSessionSupport<T> implements WebSocketSession {
 	/**
 	 * Create a new instance and associate the given attributes with it.
 	 */
-	protected WebSocketSessionSupport(T delegate, String id, HandshakeInfo handshakeInfo,
+	protected AbstractWebSocketSession(T delegate, String id, HandshakeInfo handshakeInfo,
 			DataBufferFactory bufferFactory) {
 
 		Assert.notNull(delegate, "Native session is required.");
@@ -73,10 +72,7 @@ protected WebSocketSessionSupport(T delegate, String id, HandshakeInfo handshake
 	}
 
 
-	/**
-	 * Return the native session of the underlying runtime.
-	 */
-	public T getDelegate() {
+	protected T getDelegate() {
 		return this.delegate;
 	}
 
@@ -105,6 +101,9 @@ public DataBufferFactory bufferFactory() {
 		return this.bufferFactory;
 	}
 
+
+	// WebSocketMessage factory methods
+
 	@Override
 	public WebSocketMessage textMessage(String payload) {
 		byte[] bytes = payload.getBytes(StandardCharsets.UTF_8);
@@ -130,16 +129,6 @@ public WebSocketMessage pongMessage(Function<DataBufferFactory, DataBuffer> payl
 		return new WebSocketMessage(WebSocketMessage.Type.PONG, payload);
 	}
 
-	@Override
-	public final Mono<Void> close(CloseStatus status) {
-		if (logger.isDebugEnabled()) {
-			logger.debug("Closing " + this);
-		}
-		return closeInternal(status);
-	}
-
-	protected abstract Mono<Void> closeInternal(CloseStatus status);
-
 
 	@Override
 	public String toString() {

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

@@ -40,74 +40,75 @@
 import org.springframework.web.reactive.socket.WebSocketMessage.Type;
 
 /**
- * Jetty {@code WebSocketHandler} implementation adapting and
- * delegating to a Spring {@link WebSocketHandler}.
+ * Jetty {@link WebSocket @WebSocket} handler that delegates events to a
+ * reactive {@link WebSocketHandler} and its session.
  * 
  * @author Violeta Georgieva
+ * @author Rossen Stoyanchev
  * @since 5.0
  */
 @WebSocket
 public class JettyWebSocketHandlerAdapter extends WebSocketHandlerAdapterSupport {
 
 	private static final ByteBuffer EMPTY_PAYLOAD = ByteBuffer.wrap(new byte[0]);
 
-	private JettyWebSocketSession session;
+	private JettyWebSocketSession delegateSession;
 
 
-	public JettyWebSocketHandlerAdapter(HandshakeInfo handshakeInfo, DataBufferFactory bufferFactory,
-			WebSocketHandler delegate) {
+	public JettyWebSocketHandlerAdapter(WebSocketHandler delegate, HandshakeInfo info,
+			DataBufferFactory bufferFactory) {
 
-		super(handshakeInfo, bufferFactory, delegate);
+		super(delegate, info, bufferFactory);
 	}
 
 
 	@OnWebSocketConnect
 	public void onWebSocketConnect(Session session) {
-		this.session = new JettyWebSocketSession(session, getHandshakeInfo(), getBufferFactory());
+		this.delegateSession = new JettyWebSocketSession(session, getHandshakeInfo(), bufferFactory());
 		HandlerResultSubscriber subscriber = new HandlerResultSubscriber();
-		getDelegate().handle(this.session).subscribe(subscriber);
+		getDelegate().handle(this.delegateSession).subscribe(subscriber);
 	}
 
 	@OnWebSocketMessage
 	public void onWebSocketText(String message) {
-		if (this.session != null) {
+		if (this.delegateSession != null) {
 			WebSocketMessage webSocketMessage = toMessage(Type.TEXT, message);
-			this.session.handleMessage(webSocketMessage.getType(), webSocketMessage);
+			this.delegateSession.handleMessage(webSocketMessage.getType(), webSocketMessage);
 		}
 	}
 
 	@OnWebSocketMessage
 	public void onWebSocketBinary(byte[] message, int offset, int length) {
-		if (this.session != null) {
+		if (this.delegateSession != null) {
 			ByteBuffer buffer = ByteBuffer.wrap(message, offset, length);
 			WebSocketMessage webSocketMessage = toMessage(Type.BINARY, buffer);
-			session.handleMessage(webSocketMessage.getType(), webSocketMessage);
+			delegateSession.handleMessage(webSocketMessage.getType(), webSocketMessage);
 		}
 	}
 
 	@OnWebSocketFrame
 	public void onWebSocketFrame(Frame frame) {
-		if (this.session != null) {
+		if (this.delegateSession != null) {
 			if (OpCode.PONG == frame.getOpCode()) {
 				ByteBuffer buffer = (frame.getPayload() != null ? frame.getPayload() : EMPTY_PAYLOAD);
 				WebSocketMessage webSocketMessage = toMessage(Type.PONG, buffer);
-				session.handleMessage(webSocketMessage.getType(), webSocketMessage);
+				delegateSession.handleMessage(webSocketMessage.getType(), webSocketMessage);
 			}
 		}
 	}
 
 	private <T> WebSocketMessage toMessage(Type type, T message) {
 		if (Type.TEXT.equals(type)) {
 			byte[] bytes = ((String) message).getBytes(StandardCharsets.UTF_8);
-			DataBuffer buffer = getBufferFactory().wrap(bytes);
+			DataBuffer buffer = bufferFactory().wrap(bytes);
 			return new WebSocketMessage(Type.TEXT, buffer);
 		}
 		else if (Type.BINARY.equals(type)) {
-			DataBuffer buffer = getBufferFactory().wrap((ByteBuffer) message);
+			DataBuffer buffer = bufferFactory().wrap((ByteBuffer) message);
 			return new WebSocketMessage(Type.BINARY, buffer);
 		}
 		else if (Type.PONG.equals(type)) {
-			DataBuffer buffer = getBufferFactory().wrap((ByteBuffer) message);
+			DataBuffer buffer = bufferFactory().wrap((ByteBuffer) message);
 			return new WebSocketMessage(Type.PONG, buffer);
 		}
 		else {
@@ -117,15 +118,15 @@ else if (Type.PONG.equals(type)) {
 
 	@OnWebSocketClose
 	public void onWebSocketClose(int statusCode, String reason) {
-		if (this.session != null) {
-			this.session.handleClose(new CloseStatus(statusCode, reason));
+		if (this.delegateSession != null) {
+			this.delegateSession.handleClose(new CloseStatus(statusCode, reason));
 		}
 	}
 
 	@OnWebSocketError
 	public void onWebSocketError(Throwable cause) {
-		if (this.session != null) {
-			this.session.handleError(cause);
+		if (this.delegateSession != null) {
+			this.delegateSession.handleError(cause);
 		}
 	}
 
@@ -144,15 +145,16 @@ public void onNext(Void aVoid) {
 
 		@Override
 		public void onError(Throwable ex) {
-			if (session != null) {
-				session.close(new CloseStatus(CloseStatus.SERVER_ERROR.getCode(), ex.getMessage()));
+			if (delegateSession != null) {
+				int code = CloseStatus.SERVER_ERROR.getCode();
+				delegateSession.close(new CloseStatus(code, ex.getMessage()));
 			}
 		}
 
 		@Override
 		public void onComplete() {
-			if (session != null) {
-				session.close();
+			if (delegateSession != null) {
+				delegateSession.close();
 			}
 		}
 	}