From 22a57b9aedf0039e30a0cab427998ad55f9a80dc Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Fri, 23 Dec 2016 21:55:40 -0500 Subject: [PATCH] Polish --- .../adapter/NettyWebSocketSessionSupport.java | 1 + .../socket/client/JettyWebSocketClient.java | 94 +++++++++---------- .../client/StandardWebSocketClient.java | 36 ++++--- .../client/UndertowWebSocketClient.java | 38 ++++---- .../socket/client/WebSocketClient.java | 2 +- .../socket/client/WebSocketClientSupport.java | 3 +- .../AbstractWebSocketIntegrationTests.java | 10 +- .../WebSocketIntegrationTests.java | 6 +- 8 files changed, 89 insertions(+), 101 deletions(-) rename spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/{server => }/AbstractWebSocketIntegrationTests.java (95%) rename spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/{server => }/WebSocketIntegrationTests.java (95%) diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/adapter/NettyWebSocketSessionSupport.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/adapter/NettyWebSocketSessionSupport.java index 2f8724019a..149a136b48 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/adapter/NettyWebSocketSessionSupport.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/adapter/NettyWebSocketSessionSupport.java @@ -69,6 +69,7 @@ public abstract class NettyWebSocketSessionSupport extends AbstractWebSocketS return (NettyDataBufferFactory) super.bufferFactory(); } + protected WebSocketMessage toMessage(WebSocketFrame frame) { DataBuffer payload = bufferFactory().wrap(frame.content()); return new WebSocketMessage(MESSAGE_TYPES.get(frame.getClass()), payload); diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/JettyWebSocketClient.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/JettyWebSocketClient.java index 397590742c..d8d6df08a5 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/JettyWebSocketClient.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/JettyWebSocketClient.java @@ -34,7 +34,7 @@ import org.springframework.web.reactive.socket.adapter.JettyWebSocketHandlerAdap import org.springframework.web.reactive.socket.adapter.JettyWebSocketSession; /** - * A Jetty based implementation of {@link WebSocketClient}. + * Jetty based implementation of {@link WebSocketClient}. * * @author Violeta Georgieva * @author Rossen Stoyanchev @@ -42,9 +42,9 @@ import org.springframework.web.reactive.socket.adapter.JettyWebSocketSession; */ public class JettyWebSocketClient extends WebSocketClientSupport implements WebSocketClient, Lifecycle { - private final DataBufferFactory bufferFactory = new DefaultDataBufferFactory(); + private final org.eclipse.jetty.websocket.client.WebSocketClient jettyClient; - private final org.eclipse.jetty.websocket.client.WebSocketClient wsClient; + private final DataBufferFactory bufferFactory = new DefaultDataBufferFactory(); private final Object lifecycleMonitor = new Object(); @@ -60,10 +60,46 @@ public class JettyWebSocketClient extends WebSocketClientSupport implements WebS /** * Constructor that accepts an existing * {@link org.eclipse.jetty.websocket.client.WebSocketClient} instance. - * @param wsClient a web socket client + * @param jettyClient a web socket client */ - public JettyWebSocketClient(org.eclipse.jetty.websocket.client.WebSocketClient wsClient) { - this.wsClient = wsClient; + public JettyWebSocketClient(org.eclipse.jetty.websocket.client.WebSocketClient jettyClient) { + this.jettyClient = jettyClient; + } + + + @Override + public void start() { + synchronized (this.lifecycleMonitor) { + if (!isRunning()) { + try { + this.jettyClient.start(); + } + catch (Exception ex) { + throw new IllegalStateException("Failed to start Jetty WebSocketClient", ex); + } + } + } + } + + @Override + public void stop() { + synchronized (this.lifecycleMonitor) { + if (isRunning()) { + try { + this.jettyClient.stop(); + } + catch (Exception ex) { + throw new IllegalStateException("Error stopping Jetty WebSocketClient", ex); + } + } + } + } + + @Override + public boolean isRunning() { + synchronized (this.lifecycleMonitor) { + return this.jettyClient.isStarted(); + } } @@ -84,19 +120,17 @@ public class JettyWebSocketClient extends WebSocketClientSupport implements WebS String[] protocols = beforeHandshake(url, headers, handler); ClientUpgradeRequest upgradeRequest = createRequest(headers, protocols); Object jettyHandler = createJettyHandler(url, handler, completionMono); - return this.wsClient.connect(jettyHandler, url, upgradeRequest); + return this.jettyClient.connect(jettyHandler, url, upgradeRequest); }) .then(completionMono); } private Object createJettyHandler(URI url, WebSocketHandler handler, MonoProcessor completion) { - return new JettyWebSocketHandlerAdapter( - handler, session -> createJettySession(url, completion, session)); + return new JettyWebSocketHandlerAdapter(handler, + session -> createJettySession(url, completion, session)); } - private JettyWebSocketSession createJettySession(URI url, MonoProcessor completion, - Session session) { - + private JettyWebSocketSession createJettySession(URI url, MonoProcessor completion, Session session) { UpgradeResponse response = session.getUpgradeResponse(); HttpHeaders responseHeaders = new HttpHeaders(); response.getHeaders().forEach(responseHeaders::put); @@ -111,40 +145,4 @@ public class JettyWebSocketClient extends WebSocketClientSupport implements WebS return request; } - - @Override - public void start() { - synchronized (this.lifecycleMonitor) { - if (!isRunning()) { - try { - this.wsClient.start(); - } - catch (Exception ex) { - throw new IllegalStateException("Failed to start Jetty WebSocketClient", ex); - } - } - } - } - - @Override - public void stop() { - synchronized (this.lifecycleMonitor) { - if (isRunning()) { - try { - this.wsClient.stop(); - } - catch (Exception ex) { - throw new IllegalStateException("Error stopping Jetty WebSocketClient", ex); - } - } - } - } - - @Override - public boolean isRunning() { - synchronized (this.lifecycleMonitor) { - return this.wsClient.isStarted(); - } - } - } \ No newline at end of file diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/StandardWebSocketClient.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/StandardWebSocketClient.java index 98ff388630..7605632eae 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/StandardWebSocketClient.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/StandardWebSocketClient.java @@ -25,7 +25,6 @@ import javax.websocket.ClientEndpointConfig.Configurator; import javax.websocket.ContainerProvider; import javax.websocket.Endpoint; import javax.websocket.HandshakeResponse; -import javax.websocket.Session; import javax.websocket.WebSocketContainer; import reactor.core.publisher.Mono; @@ -41,22 +40,23 @@ import org.springframework.web.reactive.socket.adapter.StandardWebSocketHandlerA import org.springframework.web.reactive.socket.adapter.StandardWebSocketSession; /** - * A Java WebSocket API (JSR-356) based implementation of - * {@link WebSocketClient}. + * Java WebSocket API (JSR-356) implementation of {@link WebSocketClient}. * * @author Violeta Georgieva + * @author Rossen Stoyanchev * @since 5.0 */ public class StandardWebSocketClient extends WebSocketClientSupport implements WebSocketClient { private final DataBufferFactory bufferFactory = new DefaultDataBufferFactory(); - private final WebSocketContainer wsContainer; + private final WebSocketContainer webSocketContainer; /** - * Default constructor that calls {@code ContainerProvider.getWebSocketContainer()} - * to obtain a (new) {@link WebSocketContainer} instance. + * Default constructor that calls + * {@code ContainerProvider.getWebSocketContainer()} to obtain a (new) + * {@link WebSocketContainer} instance. */ public StandardWebSocketClient() { this(ContainerProvider.getWebSocketContainer()); @@ -64,10 +64,10 @@ public class StandardWebSocketClient extends WebSocketClientSupport implements W /** * Constructor accepting an existing {@link WebSocketContainer} instance. - * @param wsContainer a web socket container + * @param webSocketContainer a web socket container */ - public StandardWebSocketClient(WebSocketContainer wsContainer) { - this.wsContainer = wsContainer; + public StandardWebSocketClient(WebSocketContainer webSocketContainer) { + this.webSocketContainer = webSocketContainer; } @@ -87,9 +87,9 @@ public class StandardWebSocketClient extends WebSocketClientSupport implements W () -> { String[] subProtocols = beforeHandshake(url, requestHeaders, handler); DefaultConfigurator configurator = new DefaultConfigurator(requestHeaders); - ClientEndpointConfig config = createEndpointConfig(configurator, subProtocols); Endpoint endpoint = createEndpoint(url, handler, completionMono, configurator); - return this.wsContainer.connectToServer(endpoint, config, url); + ClientEndpointConfig config = createEndpointConfig(configurator, subProtocols); + return this.webSocketContainer.connectToServer(endpoint, config, url); }) .subscribeOn(Schedulers.elastic()) // connectToServer is blocking .then(completionMono); @@ -105,15 +105,11 @@ public class StandardWebSocketClient extends WebSocketClientSupport implements W private StandardWebSocketHandlerAdapter createEndpoint(URI url, WebSocketHandler handler, MonoProcessor completion, DefaultConfigurator configurator) { - return new StandardWebSocketHandlerAdapter(handler, - session -> createSession(url, configurator.getResponseHeaders(), completion, session)); - } - - private StandardWebSocketSession createSession(URI url, HttpHeaders responseHeaders, - MonoProcessor completion, Session session) { - - HandshakeInfo info = afterHandshake(url, responseHeaders); - return new StandardWebSocketSession(session, info, this.bufferFactory, completion); + return new StandardWebSocketHandlerAdapter(handler, session -> { + HttpHeaders responseHeaders = configurator.getResponseHeaders(); + HandshakeInfo info = afterHandshake(url, responseHeaders); + return new StandardWebSocketSession(session, info, this.bufferFactory, completion); + }); } diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/UndertowWebSocketClient.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/UndertowWebSocketClient.java index 4a2ac6314f..a48b2c71a3 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/UndertowWebSocketClient.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/UndertowWebSocketClient.java @@ -27,6 +27,7 @@ import java.util.concurrent.CancellationException; import java.util.function.Function; import javax.net.ssl.SSLContext; +import io.undertow.connector.ByteBufferPool; import io.undertow.protocols.ssl.UndertowXnioSsl; import io.undertow.server.DefaultByteBufferPool; import io.undertow.websockets.client.WebSocketClient.ConnectionBuilder; @@ -37,6 +38,7 @@ import org.xnio.OptionMap; import org.xnio.Options; import org.xnio.Xnio; import org.xnio.XnioWorker; +import org.xnio.ssl.XnioSsl; import reactor.core.publisher.Mono; import reactor.core.publisher.MonoProcessor; @@ -49,7 +51,7 @@ import org.springframework.web.reactive.socket.adapter.UndertowWebSocketHandlerA import org.springframework.web.reactive.socket.adapter.UndertowWebSocketSession; /** - * An Undertow based implementation of {@link WebSocketClient}. + * Undertow based implementation of {@link WebSocketClient}. * * @author Violeta Georgieva * @author Rossen Stoyanchev @@ -79,24 +81,23 @@ public class UndertowWebSocketClient extends WebSocketClientSupport implements W } - private final DataBufferFactory bufferFactory = new DefaultDataBufferFactory(); - private final Function builder; + private final DataBufferFactory bufferFactory = new DefaultDataBufferFactory(); + /** * Default constructor that uses * {@link io.undertow.websockets.client.WebSocketClient#connectionBuilder(XnioWorker, ByteBufferPool, URI)} - * to create a web socket connection. + * to create WebSocket connections. */ public UndertowWebSocketClient() { this(UndertowWebSocketClient::createDefaultConnectionBuilder); } /** - * Constructor that accepts an existing - * {@link io.undertow.websockets.client.WebSocketClient#connectionBuilder(XnioWorker, ByteBufferPool, URI)} - * instance. + * Constructor that accepts a {@link Function} to prepare a + * {@link ConnectionBuilder} for WebSocket connections. * @param builder a connection builder that can be used to create a web socket connection. */ public UndertowWebSocketClient(Function builder) { @@ -105,15 +106,13 @@ public class UndertowWebSocketClient extends WebSocketClientSupport implements W private static ConnectionBuilder createDefaultConnectionBuilder(URI url) { - ConnectionBuilder builder = - io.undertow.websockets.client.WebSocketClient.connectionBuilder( + ConnectionBuilder builder = io.undertow.websockets.client.WebSocketClient.connectionBuilder( worker, new DefaultByteBufferPool(false, DEFAULT_BUFFER_SIZE), url); boolean secure = "wss".equals(url.getScheme()); if (secure) { try { - UndertowXnioSsl ssl = new UndertowXnioSsl(Xnio.getInstance(), - OptionMap.EMPTY, SSLContext.getDefault()); + XnioSsl ssl = new UndertowXnioSsl(Xnio.getInstance(), OptionMap.EMPTY, SSLContext.getDefault()); builder.setSsl(ssl); } catch (NoSuchAlgorithmException ex) { @@ -136,7 +135,7 @@ public class UndertowWebSocketClient extends WebSocketClientSupport implements W } private Mono executeInternal(URI url, HttpHeaders headers, WebSocketHandler handler) { - MonoProcessor completionMono = MonoProcessor.create(); + MonoProcessor completion = MonoProcessor.create(); return Mono.fromCallable( () -> { String[] subProtocols = beforeHandshake(url, headers, handler); @@ -147,29 +146,26 @@ public class UndertowWebSocketClient extends WebSocketClientSupport implements W .connect() .addNotifier((future, attachment) -> { if (Status.DONE.equals(future.getStatus())) { - WebSocketChannel channel; try { - channel = future.get(); + handleChannel(url, handler, completion, negotiation, future.get()); } catch (CancellationException | IOException ex) { - completionMono.onError(ex); - return; + completion.onError(ex); } - handleWebSocket(url, handler, completionMono, negotiation, channel); } else if (Status.FAILED.equals(future.getStatus())) { - completionMono.onError(future.getException()); + completion.onError(future.getException()); } else { String message = "Failed to connect" + future.getStatus(); - completionMono.onError(new IllegalStateException(message)); + completion.onError(new IllegalStateException(message)); } }, null); }) - .then(completionMono); + .then(completion); } - private void handleWebSocket(URI url, WebSocketHandler handler, MonoProcessor completion, + private void handleChannel(URI url, WebSocketHandler handler, MonoProcessor completion, DefaultNegotiation negotiation, WebSocketChannel channel) { HandshakeInfo info = afterHandshake(url, negotiation.getResponseHeaders()); diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/WebSocketClient.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/WebSocketClient.java index 5821f984e3..9db20271a5 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/WebSocketClient.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/WebSocketClient.java @@ -23,7 +23,7 @@ import org.springframework.http.HttpHeaders; import org.springframework.web.reactive.socket.WebSocketHandler; /** - * Contract for connecting and handling a WebSocket session. + * Contract for reactive-style handling of a WebSocket session. * * @author Rossen Stoyanchev * @since 5.0 diff --git a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/WebSocketClientSupport.java b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/WebSocketClientSupport.java index 1173cc354e..37619ea81c 100644 --- a/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/WebSocketClientSupport.java +++ b/spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/WebSocketClientSupport.java @@ -23,7 +23,6 @@ import org.apache.commons.logging.LogFactory; import reactor.core.publisher.Mono; import org.springframework.http.HttpHeaders; -import org.springframework.util.Assert; import org.springframework.web.reactive.socket.HandshakeInfo; import org.springframework.web.reactive.socket.WebSocketHandler; @@ -35,7 +34,7 @@ import org.springframework.web.reactive.socket.WebSocketHandler; */ public class WebSocketClientSupport { - protected static final String SEC_WEBSOCKET_PROTOCOL = "Sec-WebSocket-Protocol"; + private static final String SEC_WEBSOCKET_PROTOCOL = "Sec-WebSocket-Protocol"; protected final Log logger = LogFactory.getLog(getClass()); diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/server/AbstractWebSocketIntegrationTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/AbstractWebSocketIntegrationTests.java similarity index 95% rename from spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/server/AbstractWebSocketIntegrationTests.java rename to spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/AbstractWebSocketIntegrationTests.java index 0e8903349c..0993470d6e 100644 --- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/server/AbstractWebSocketIntegrationTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/AbstractWebSocketIntegrationTests.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.web.reactive.socket.server; +package org.springframework.web.reactive.socket; import java.io.File; import java.net.URI; @@ -49,6 +49,8 @@ import org.springframework.web.reactive.socket.client.RxNettyWebSocketClient; import org.springframework.web.reactive.socket.client.StandardWebSocketClient; import org.springframework.web.reactive.socket.client.UndertowWebSocketClient; import org.springframework.web.reactive.socket.client.WebSocketClient; +import org.springframework.web.reactive.socket.server.RequestUpgradeStrategy; +import org.springframework.web.reactive.socket.server.WebSocketService; import org.springframework.web.reactive.socket.server.support.HandshakeWebSocketService; import org.springframework.web.reactive.socket.server.support.WebSocketHandlerAdapter; import org.springframework.web.reactive.socket.server.upgrade.JettyRequestUpgradeStrategy; @@ -60,9 +62,9 @@ import org.springframework.web.reactive.socket.server.upgrade.UndertowRequestUpg import static org.junit.Assume.assumeFalse; /** - * Base class for WebSocket integration tests. - * Sub-classes must implement {@link #getWebConfigClass()} to return Spring - * config class with handler mappings to {@code WebSocketHandler}'s. + * Base class for WebSocket integration tests. Sub-classes must implement + * {@link #getWebConfigClass()} to return Spring config class with (server-side) + * handler mappings to {@code WebSocketHandler}'s. * * @author Rossen Stoyanchev */ diff --git a/spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/server/WebSocketIntegrationTests.java b/spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/WebSocketIntegrationTests.java similarity index 95% rename from spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/server/WebSocketIntegrationTests.java rename to spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/WebSocketIntegrationTests.java index e8b0d18eb9..f70163cd62 100644 --- a/spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/server/WebSocketIntegrationTests.java +++ b/spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/WebSocketIntegrationTests.java @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.web.reactive.socket.server; +package org.springframework.web.reactive.socket; import java.util.HashMap; import java.util.Map; @@ -32,10 +32,6 @@ import org.springframework.context.annotation.Configuration; import org.springframework.http.HttpHeaders; import org.springframework.web.reactive.HandlerMapping; import org.springframework.web.reactive.handler.SimpleUrlHandlerMapping; -import org.springframework.web.reactive.socket.HandshakeInfo; -import org.springframework.web.reactive.socket.WebSocketHandler; -import org.springframework.web.reactive.socket.WebSocketMessage; -import org.springframework.web.reactive.socket.WebSocketSession; import org.springframework.web.reactive.socket.client.JettyWebSocketClient; import org.springframework.web.reactive.socket.client.ReactorNettyWebSocketClient;