Polish

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

@@ -213,7 +213,12 @@ public abstract class AbstractListenerWebSocketSession<T> extends WebSocketSessi
 			return this.isReady && this.currentData != null;
 		}
 
-		public void setReady(boolean ready) {
+		/**
+		 * Sub-classes can invoke this before sending a message (false) and
+		 * after receiving the async send callback (true) effective translating
+		 * async completion callback into simple flow control.
+		 */
+		public void setReadyToSend(boolean ready) {
 			this.isReady = ready;
 		}
 	}

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

@@ -30,6 +30,8 @@ import org.eclipse.jetty.websocket.api.extensions.Frame;
 import org.eclipse.jetty.websocket.common.OpCode;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
+
+import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.util.Assert;
@@ -50,84 +52,90 @@ public class JettyWebSocketHandlerAdapter {
 
 	private static final ByteBuffer EMPTY_PAYLOAD = ByteBuffer.wrap(new byte[0]);
 
+
 	private final DataBufferFactory bufferFactory = new DefaultDataBufferFactory(false);
 
-	private final WebSocketHandler handler;
+	private final WebSocketHandler delegate;
+
+	private JettyWebSocketSession session;
 
-	private JettyWebSocketSession wsSession;
 
-	public JettyWebSocketHandlerAdapter(WebSocketHandler handler) {
-		Assert.notNull("'handler' is required");
-		this.handler = handler;
+	public JettyWebSocketHandlerAdapter(WebSocketHandler delegate) {
+		Assert.notNull("WebSocketHandler is required");
+		this.delegate = delegate;
 	}
 
+
 	@OnWebSocketConnect
 	public void onWebSocketConnect(Session session) {
-		this.wsSession = new JettyWebSocketSession(session);
+		this.session = new JettyWebSocketSession(session);
 
-		HandlerResultSubscriber resultSubscriber = new HandlerResultSubscriber();
-		this.handler.handle(this.wsSession).subscribe(resultSubscriber);
+		HandlerResultSubscriber subscriber = new HandlerResultSubscriber();
+		this.delegate.handle(this.session).subscribe(subscriber);
 	}
 
 	@OnWebSocketMessage
 	public void onWebSocketText(String message) {
-		if (this.wsSession != null) {
-			WebSocketMessage wsMessage = toMessage(Type.TEXT, message);
-			this.wsSession.handleMessage(wsMessage.getType(), wsMessage);
+		if (this.session != null) {
+			WebSocketMessage webSocketMessage = toMessage(Type.TEXT, message);
+			this.session.handleMessage(webSocketMessage.getType(), webSocketMessage);
 		}
 	}
 
 	@OnWebSocketMessage
 	public void onWebSocketBinary(byte[] message, int offset, int length) {
-		if (this.wsSession != null) {
-			WebSocketMessage wsMessage = toMessage(Type.BINARY, ByteBuffer.wrap(message, offset, length));
-			wsSession.handleMessage(wsMessage.getType(), wsMessage);
+		if (this.session != null) {
+			ByteBuffer buffer = ByteBuffer.wrap(message, offset, length);
+			WebSocketMessage webSocketMessage = toMessage(Type.BINARY, buffer);
+			session.handleMessage(webSocketMessage.getType(), webSocketMessage);
 		}
 	}
 
 	@OnWebSocketFrame
 	public void onWebSocketFrame(Frame frame) {
-		if (this.wsSession != null) {
+		if (this.session != null) {
 			if (OpCode.PONG == frame.getOpCode()) {
-				ByteBuffer message = frame.getPayload() != null ? frame.getPayload() : EMPTY_PAYLOAD;
-				WebSocketMessage wsMessage = toMessage(Type.PONG, message);
-				wsSession.handleMessage(wsMessage.getType(), wsMessage);
+				ByteBuffer buffer = (frame.getPayload() != null ? frame.getPayload() : EMPTY_PAYLOAD);
+				WebSocketMessage webSocketMessage = toMessage(Type.PONG, buffer);
+				session.handleMessage(webSocketMessage.getType(), webSocketMessage);
 			}
 		}
 	}
 
 	@OnWebSocketClose
 	public void onWebSocketClose(int statusCode, String reason) {
-		if (this.wsSession != null) {
-			this.wsSession.handleClose(new CloseStatus(statusCode, reason));
+		if (this.session != null) {
+			this.session.handleClose(new CloseStatus(statusCode, reason));
 		}
 	}
 
 	@OnWebSocketError
 	public void onWebSocketError(Throwable cause) {
-		if (this.wsSession != null) {
-			this.wsSession.handleError(cause);
+		if (this.session != null) {
+			this.session.handleError(cause);
 		}
 	}
 
 	private <T> WebSocketMessage toMessage(Type type, T message) {
 		if (Type.TEXT.equals(type)) {
-			return WebSocketMessage.create(Type.TEXT,
-					bufferFactory.wrap(((String) message).getBytes(StandardCharsets.UTF_8)));
+			byte[] bytes = ((String) message).getBytes(StandardCharsets.UTF_8);
+			DataBuffer buffer = this.bufferFactory.wrap(bytes);
+			return WebSocketMessage.create(Type.TEXT, buffer);
 		}
 		else if (Type.BINARY.equals(type)) {
-			return WebSocketMessage.create(Type.BINARY,
-					bufferFactory.wrap((ByteBuffer) message));
+			DataBuffer buffer = this.bufferFactory.wrap((ByteBuffer) message);
+			return WebSocketMessage.create(Type.BINARY, buffer);
 		}
 		else if (Type.PONG.equals(type)) {
-			return WebSocketMessage.create(Type.PONG,
-					bufferFactory.wrap((ByteBuffer) message));
+			DataBuffer buffer = this.bufferFactory.wrap((ByteBuffer) message);
+			return WebSocketMessage.create(Type.PONG, buffer);
 		}
 		else {
 			throw new IllegalArgumentException("Unexpected message type: " + message);
 		}
 	}
 
+
 	private final class HandlerResultSubscriber implements Subscriber<Void> {
 
 		@Override
@@ -142,15 +150,15 @@ public class JettyWebSocketHandlerAdapter {
 
 		@Override
 		public void onError(Throwable ex) {
-			if (wsSession != null) {
-				wsSession.close(new CloseStatus(CloseStatus.SERVER_ERROR.getCode(), ex.getMessage()));
+			if (session != null) {
+				session.close(new CloseStatus(CloseStatus.SERVER_ERROR.getCode(), ex.getMessage()));
 			}
 		}
 
 		@Override
 		public void onComplete() {
-			if (wsSession != null) {
-				wsSession.close();
+			if (session != null) {
+				session.close();
 			}
 		}
 	}

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

@@ -17,6 +17,7 @@
 package org.springframework.web.reactive.socket.adapter;
 
 import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 
 import org.eclipse.jetty.websocket.api.Session;
@@ -52,22 +53,21 @@ public class JettyWebSocketSession extends AbstractListenerWebSocketSession<Sess
 
 	@Override
 	protected boolean sendMessage(WebSocketMessage message) throws IOException {
+		ByteBuffer buffer = message.getPayload().asByteBuffer();
 		if (WebSocketMessage.Type.TEXT.equals(message.getType())) {
-			getSendProcessor().setReady(false);
-			getDelegate().getRemote().sendString(
-					new String(message.getPayload().asByteBuffer().array(), StandardCharsets.UTF_8),
-					new WebSocketMessageWriteCallback());
+			getSendProcessor().setReadyToSend(false);
+			String text = new String(buffer.array(), StandardCharsets.UTF_8);
+			getDelegate().getRemote().sendString(text, new SendProcessorCallback());
 		}
 		else if (WebSocketMessage.Type.BINARY.equals(message.getType())) {
-			getSendProcessor().setReady(false);
-			getDelegate().getRemote().sendBytes(message.getPayload().asByteBuffer(),
-					new WebSocketMessageWriteCallback());
+			getSendProcessor().setReadyToSend(false);
+			getDelegate().getRemote().sendBytes(buffer, new SendProcessorCallback());
 		}
 		else if (WebSocketMessage.Type.PING.equals(message.getType())) {
-			getDelegate().getRemote().sendPing(message.getPayload().asByteBuffer());
+			getDelegate().getRemote().sendPing(buffer);
 		}
 		else if (WebSocketMessage.Type.PONG.equals(message.getType())) {
-			getDelegate().getRemote().sendPong(message.getPayload().asByteBuffer());
+			getDelegate().getRemote().sendPong(buffer);
 		}
 		else {
 			throw new IllegalArgumentException("Unexpected message type: " + message.getType());
@@ -91,7 +91,7 @@ public class JettyWebSocketSession extends AbstractListenerWebSocketSession<Sess
 	}
 
 
-	private final class WebSocketMessageWriteCallback implements WriteCallback {
+	private final class SendProcessorCallback implements WriteCallback {
 
 		@Override
 		public void writeFailed(Throwable x) {
@@ -101,7 +101,7 @@ public class JettyWebSocketSession extends AbstractListenerWebSocketSession<Sess
 
 		@Override
 		public void writeSuccess() {
-			getSendProcessor().setReady(true);
+			getSendProcessor().setReadyToSend(true);
 			getSendProcessor().onWritePossible();
 		}
 

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

@@ -18,16 +18,16 @@ package org.springframework.web.reactive.socket.adapter;
 
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
-
 import javax.websocket.CloseReason;
 import javax.websocket.Endpoint;
 import javax.websocket.EndpointConfig;
-import javax.websocket.MessageHandler;
 import javax.websocket.PongMessage;
 import javax.websocket.Session;
 
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
+
+import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.util.Assert;
@@ -47,84 +47,72 @@ public class TomcatWebSocketHandlerAdapter extends Endpoint {
 
 	private final DataBufferFactory bufferFactory = new DefaultDataBufferFactory(false);
 
-	private final WebSocketHandler handler;
+	private final WebSocketHandler delegate;
 
-	private TomcatWebSocketSession wsSession;
+	private TomcatWebSocketSession session;
 
-	public TomcatWebSocketHandlerAdapter(WebSocketHandler handler) {
-		Assert.notNull("'handler' is required");
-		this.handler = handler;
+
+	public TomcatWebSocketHandlerAdapter(WebSocketHandler delegate) {
+		Assert.notNull("WebSocketHandler is required");
+		this.delegate = delegate;
 	}
 
+
 	@Override
 	public void onOpen(Session session, EndpointConfig config) {
-		this.wsSession = new TomcatWebSocketSession(session);
-
-		session.addMessageHandler(new MessageHandler.Whole<String>() {
-
-			@Override
-			public void onMessage(String message) {
-				WebSocketMessage wsMessage = toMessage(message);
-				wsSession.handleMessage(wsMessage.getType(), wsMessage);
-			}
+		this.session = new TomcatWebSocketSession(session);
 
+		session.addMessageHandler(String.class, message -> {
+			WebSocketMessage webSocketMessage = toMessage(message);
+			this.session.handleMessage(webSocketMessage.getType(), webSocketMessage);
 		});
-		session.addMessageHandler(new MessageHandler.Whole<ByteBuffer>() {
-
-			@Override
-			public void onMessage(ByteBuffer message) {
-				WebSocketMessage wsMessage = toMessage(message);
-				wsSession.handleMessage(wsMessage.getType(), wsMessage);
-			}
-
+		session.addMessageHandler(ByteBuffer.class, message -> {
+			WebSocketMessage webSocketMessage = toMessage(message);
+			this.session.handleMessage(webSocketMessage.getType(), webSocketMessage);
 		});
-		session.addMessageHandler(new MessageHandler.Whole<PongMessage>() {
-
-			@Override
-			public void onMessage(PongMessage message) {
-				WebSocketMessage wsMessage = toMessage(message);
-				wsSession.handleMessage(wsMessage.getType(), wsMessage);
-			}
-
+		session.addMessageHandler(PongMessage.class, message -> {
+			WebSocketMessage webSocketMessage = toMessage(message);
+			this.session.handleMessage(webSocketMessage.getType(), webSocketMessage);
 		});
 
 		HandlerResultSubscriber resultSubscriber = new HandlerResultSubscriber();
-		this.handler.handle(this.wsSession).subscribe(resultSubscriber);
+		this.delegate.handle(this.session).subscribe(resultSubscriber);
 	}
 
 	@Override
 	public void onClose(Session session, CloseReason reason) {
-		if (this.wsSession != null) {
-			this.wsSession.handleClose(
-					new CloseStatus(reason.getCloseCode().getCode(), reason.getReasonPhrase()));
+		if (this.session != null) {
+			int code = reason.getCloseCode().getCode();
+			this.session.handleClose(new CloseStatus(code, reason.getReasonPhrase()));
 		}
 	}
 
 	@Override
 	public void onError(Session session, Throwable exception) {
-		if (this.wsSession != null) {
-			this.wsSession.handleError(exception);
+		if (this.session != null) {
+			this.session.handleError(exception);
 		}
 	}
 
 	private <T> WebSocketMessage toMessage(T message) {
 		if (message instanceof String) {
-			return WebSocketMessage.create(Type.TEXT,
-					bufferFactory.wrap(((String) message).getBytes(StandardCharsets.UTF_8)));
+			byte[] bytes = ((String) message).getBytes(StandardCharsets.UTF_8);
+			return WebSocketMessage.create(Type.TEXT, this.bufferFactory.wrap(bytes));
 		}
 		else if (message instanceof ByteBuffer) {
-			return WebSocketMessage.create(Type.BINARY,
-					bufferFactory.wrap((ByteBuffer) message));
+			DataBuffer buffer = this.bufferFactory.wrap((ByteBuffer) message);
+			return WebSocketMessage.create(Type.BINARY, buffer);
 		}
 		else if (message instanceof PongMessage) {
-			return WebSocketMessage.create(Type.PONG,
-					bufferFactory.wrap(((PongMessage) message).getApplicationData()));
+			DataBuffer buffer = this.bufferFactory.wrap(((PongMessage) message).getApplicationData());
+			return WebSocketMessage.create(Type.PONG, buffer);
 		}
 		else {
 			throw new IllegalArgumentException("Unexpected message type: " + message);
 		}
 	}
 
+
 	private final class HandlerResultSubscriber implements Subscriber<Void> {
 
 		@Override
@@ -139,15 +127,15 @@ public class TomcatWebSocketHandlerAdapter extends Endpoint {
 
 		@Override
 		public void onError(Throwable ex) {
-			if (wsSession != null) {
-				wsSession.close(new CloseStatus(CloseStatus.SERVER_ERROR.getCode(), ex.getMessage()));
+			if (session != null) {
+				session.close(new CloseStatus(CloseStatus.SERVER_ERROR.getCode(), ex.getMessage()));
 			}
 		}
 
 		@Override
 		public void onComplete() {
-			if (wsSession != null) {
-				wsSession.close();
+			if (session != null) {
+				session.close();
 			}
 		}
 	}

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

@@ -17,6 +17,7 @@
 package org.springframework.web.reactive.socket.adapter;
 
 import java.io.IOException;
+import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import javax.websocket.CloseReason;
 import javax.websocket.CloseReason.CloseCodes;
@@ -59,22 +60,21 @@ public class TomcatWebSocketSession extends AbstractListenerWebSocketSession<Ses
 
 	@Override
 	protected boolean sendMessage(WebSocketMessage message) throws IOException {
+		ByteBuffer buffer = message.getPayload().asByteBuffer();
 		if (WebSocketMessage.Type.TEXT.equals(message.getType())) {
-			getSendProcessor().setReady(false);
-			getDelegate().getAsyncRemote().sendText(
-					new String(message.getPayload().asByteBuffer().array(), StandardCharsets.UTF_8),
-					new WebSocketMessageSendHandler());
+			getSendProcessor().setReadyToSend(false);
+			String text = new String(buffer.array(), StandardCharsets.UTF_8);
+			getDelegate().getAsyncRemote().sendText(text, new SendProcessorCallback());
 		}
 		else if (WebSocketMessage.Type.BINARY.equals(message.getType())) {
-			getSendProcessor().setReady(false);
-			getDelegate().getAsyncRemote().sendBinary(message.getPayload().asByteBuffer(),
-					new WebSocketMessageSendHandler());
+			getSendProcessor().setReadyToSend(false);
+			getDelegate().getAsyncRemote().sendBinary(buffer, new SendProcessorCallback());
 		}
 		else if (WebSocketMessage.Type.PING.equals(message.getType())) {
-			getDelegate().getAsyncRemote().sendPing(message.getPayload().asByteBuffer());
+			getDelegate().getAsyncRemote().sendPing(buffer);
 		}
 		else if (WebSocketMessage.Type.PONG.equals(message.getType())) {
-			getDelegate().getAsyncRemote().sendPong(message.getPayload().asByteBuffer());
+			getDelegate().getAsyncRemote().sendPong(buffer);
 		}
 		else {
 			throw new IllegalArgumentException("Unexpected message type: " + message.getType());
@@ -98,12 +98,12 @@ public class TomcatWebSocketSession extends AbstractListenerWebSocketSession<Ses
 	}
 
 
-	private final class WebSocketMessageSendHandler implements SendHandler {
+	private final class SendProcessorCallback implements SendHandler {
 
 		@Override
 		public void onResult(SendResult result) {
 			if (result.isOK()) {
-				getSendProcessor().setReady(true);
+				getSendProcessor().setReadyToSend(true);
 				getSendProcessor().onWritePossible();
 			}
 			else {

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

@@ -16,13 +16,14 @@
 
 package org.springframework.web.reactive.socket.adapter;
 
-import java.io.IOException;
 import java.net.URISyntaxException;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
+
+import org.springframework.core.io.buffer.DataBuffer;
 import org.springframework.core.io.buffer.DataBufferFactory;
 import org.springframework.core.io.buffer.DefaultDataBufferFactory;
 import org.springframework.util.Assert;
@@ -50,78 +51,78 @@ public class UndertowWebSocketHandlerAdapter implements WebSocketConnectionCallb
 
 	private final DataBufferFactory bufferFactory = new DefaultDataBufferFactory(false);
 
-	private final WebSocketHandler handler;
+	private final WebSocketHandler delegate;
+
+	private UndertowWebSocketSession session;
 
-	private UndertowWebSocketSession wsSession;
 
-	public UndertowWebSocketHandlerAdapter(WebSocketHandler handler) {
-		Assert.notNull("'handler' is required");
-		this.handler = handler;
+	public UndertowWebSocketHandlerAdapter(WebSocketHandler delegate) {
+		Assert.notNull("WebSocketHandler is required");
+		this.delegate = delegate;
 	}
 
+
 	@Override
 	public void onConnect(WebSocketHttpExchange exchange, WebSocketChannel channel) {
 		try {
-			this.wsSession = new UndertowWebSocketSession(channel);
+			this.session = new UndertowWebSocketSession(channel);
 		}
 		catch (URISyntaxException e) {
 			// TODO Auto-generated catch block
 			e.printStackTrace();
 		}
 
-		channel.getReceiveSetter().set(new ReceiveListener());
+		channel.getReceiveSetter().set(new UndertowReceiveListener());
 		channel.resumeReceives();
 
 		HandlerResultSubscriber resultSubscriber = new HandlerResultSubscriber();
-		this.handler.handle(this.wsSession).subscribe(resultSubscriber);
+		this.delegate.handle(this.session).subscribe(resultSubscriber);
 	}
 
-	private final class ReceiveListener extends AbstractReceiveListener {
+
+	private final class UndertowReceiveListener extends AbstractReceiveListener {
 
 		@Override
 		protected void onFullTextMessage(WebSocketChannel channel, BufferedTextMessage message) {
-			wsSession.handleMessage(Type.TEXT, toMessage(Type.TEXT, message.getData()));
+			session.handleMessage(Type.TEXT, toMessage(Type.TEXT, message.getData()));
 		}
 
 		@Override
-		protected void onFullBinaryMessage(WebSocketChannel channel,
-				BufferedBinaryMessage message) throws IOException {
-			wsSession.handleMessage(Type.BINARY, toMessage(Type.BINARY, message.getData().getResource()));
+		protected void onFullBinaryMessage(WebSocketChannel channel, BufferedBinaryMessage message) {
+			session.handleMessage(Type.BINARY, toMessage(Type.BINARY, message.getData().getResource()));
 			message.getData().free();
 		}
 
 		@Override
-		protected void onFullPongMessage(WebSocketChannel channel,
-				BufferedBinaryMessage message) throws IOException {
-			wsSession.handleMessage(Type.PONG, toMessage(Type.PONG, message.getData().getResource()));
+		protected void onFullPongMessage(WebSocketChannel channel, BufferedBinaryMessage message) {
+			session.handleMessage(Type.PONG, toMessage(Type.PONG, message.getData().getResource()));
 			message.getData().free();
 		}
 
 		@Override
-		protected void onFullCloseMessage(WebSocketChannel channel,
-				BufferedBinaryMessage message) throws IOException {
+		protected void onFullCloseMessage(WebSocketChannel channel, BufferedBinaryMessage message) {
 			CloseMessage closeMessage = new CloseMessage(message.getData().getResource());
-			wsSession.handleClose(new CloseStatus(closeMessage.getCode(), closeMessage.getReason()));
+			session.handleClose(new CloseStatus(closeMessage.getCode(), closeMessage.getReason()));
 			message.getData().free();
 		}
 
 		@Override
 		protected void onError(WebSocketChannel channel, Throwable error) {
-			wsSession.handleError(error);
+			session.handleError(error);
 		}
 
 		private <T> WebSocketMessage toMessage(Type type, T message) {
 			if (Type.TEXT.equals(type)) {
-				return WebSocketMessage.create(Type.TEXT,
-						bufferFactory.wrap(((String) message).getBytes(StandardCharsets.UTF_8)));
+				byte[] bytes = ((String) message).getBytes(StandardCharsets.UTF_8);
+				return WebSocketMessage.create(Type.TEXT, bufferFactory.wrap(bytes));
 			}
 			else if (Type.BINARY.equals(type)) {
-				return WebSocketMessage.create(Type.BINARY,
-						bufferFactory.allocateBuffer().write((ByteBuffer[]) message));
+				DataBuffer buffer = bufferFactory.allocateBuffer().write((ByteBuffer[]) message);
+				return WebSocketMessage.create(Type.BINARY, buffer);
 			}
 			else if (Type.PONG.equals(type)) {
-				return WebSocketMessage.create(Type.PONG,
-						bufferFactory.allocateBuffer().write((ByteBuffer[]) message));
+				DataBuffer buffer = bufferFactory.allocateBuffer().write((ByteBuffer[]) message);
+				return WebSocketMessage.create(Type.PONG, buffer);
 			}
 			else {
 				throw new IllegalArgumentException("Unexpected message type: " + message);
@@ -144,12 +145,12 @@ public class UndertowWebSocketHandlerAdapter implements WebSocketConnectionCallb
 
 		@Override
 		public void onError(Throwable ex) {
-			wsSession.close(new CloseStatus(CloseStatus.SERVER_ERROR.getCode(), ex.getMessage()));
+			session.close(new CloseStatus(CloseStatus.SERVER_ERROR.getCode(), ex.getMessage()));
 		}
 
 		@Override
 		public void onComplete() {
-			wsSession.close();
+			session.close();
 		}
 	}
 

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

@@ -19,6 +19,7 @@ package org.springframework.web.reactive.socket.adapter;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 
 import io.undertow.websockets.core.CloseMessage;
@@ -41,10 +42,12 @@ import org.springframework.web.reactive.socket.WebSocketSession;
  */
 public class UndertowWebSocketSession extends AbstractListenerWebSocketSession<WebSocketChannel> {
 
+
 	public UndertowWebSocketSession(WebSocketChannel channel) throws URISyntaxException {
 		super(channel, ObjectUtils.getIdentityHexString(channel), new URI(channel.getUrl()));
 	}
 
+
 	@Override
 	protected Mono<Void> closeInternal(CloseStatus status) {
 		CloseMessage cm = new CloseMessage(status.getCode(), status.getReason());
@@ -69,26 +72,23 @@ public class UndertowWebSocketSession extends AbstractListenerWebSocketSession<W
 
 	@Override
 	protected boolean sendMessage(WebSocketMessage message) throws IOException {
+		ByteBuffer buffer = message.getPayload().asByteBuffer();
 		if (WebSocketMessage.Type.TEXT.equals(message.getType())) {
-			getSendProcessor().setReady(false);
-			WebSockets.sendText(
-					new String(message.getPayload().asByteBuffer().array(), StandardCharsets.UTF_8),
-					getDelegate(), new WebSocketMessageSendHandler());
+			getSendProcessor().setReadyToSend(false);
+			String text = new String(buffer.array(), StandardCharsets.UTF_8);
+			WebSockets.sendText(text, getDelegate(), new SendProcessorCallback());
 		}
 		else if (WebSocketMessage.Type.BINARY.equals(message.getType())) {
-			getSendProcessor().setReady(false);
-			WebSockets.sendBinary(message.getPayload().asByteBuffer(),
-					getDelegate(), new WebSocketMessageSendHandler());
+			getSendProcessor().setReadyToSend(false);
+			WebSockets.sendBinary(buffer, getDelegate(), new SendProcessorCallback());
 		}
 		else if (WebSocketMessage.Type.PING.equals(message.getType())) {
-			getSendProcessor().setReady(false);
-			WebSockets.sendPing(message.getPayload().asByteBuffer(),
-					getDelegate(), new WebSocketMessageSendHandler());
+			getSendProcessor().setReadyToSend(false);
+			WebSockets.sendPing(buffer, getDelegate(), new SendProcessorCallback());
 		}
 		else if (WebSocketMessage.Type.PONG.equals(message.getType())) {
-			getSendProcessor().setReady(false);
-			WebSockets.sendPong(message.getPayload().asByteBuffer(),
-					getDelegate(), new WebSocketMessageSendHandler());
+			getSendProcessor().setReadyToSend(false);
+			WebSockets.sendPong(buffer, getDelegate(), new SendProcessorCallback());
 		}
 		else {
 			throw new IllegalArgumentException("Unexpected message type: " + message.getType());
@@ -96,21 +96,20 @@ public class UndertowWebSocketSession extends AbstractListenerWebSocketSession<W
 		return true;
 	}
 
-	private final class WebSocketMessageSendHandler implements WebSocketCallback<Void> {
+
+	private final class SendProcessorCallback implements WebSocketCallback<Void> {
 
 		@Override
 		public void complete(WebSocketChannel channel, Void context) {
-			getSendProcessor().setReady(true);
+			getSendProcessor().setReadyToSend(true);
 			getSendProcessor().onWritePossible();
 		}
 
 		@Override
-		public void onError(WebSocketChannel channel, Void context,
-				Throwable throwable) {
+		public void onError(WebSocketChannel channel, Void context, Throwable throwable) {
 			getSendProcessor().cancel();
 			getSendProcessor().onError(throwable);
 		}
-
 	}
 
 }