Spring/spring-framework
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Polish

Browse Source
This commit is contained in:
Rossen Stoyanchev
2016-12-23 21:55:40 -05:00
parent 1b8cdb8924
commit 22a57b9aed
8 changed files with 89 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/adapter/NettyWebSocketSessionSupport.java
View File

@@ -69,6 +69,7 @@ public abstract class NettyWebSocketSessionSupport<T> 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);

94
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/JettyWebSocketClient.java
View File

@@ -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<Void> 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<Void> completion,
Session session) {
private JettyWebSocketSession createJettySession(URI url, MonoProcessor<Void> 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();
}
}
}

36
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/StandardWebSocketClient.java
View File

@@ -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<Void> completion, DefaultConfigurator configurator) {
return new StandardWebSocketHandlerAdapter(handler,
session -> createSession(url, configurator.getResponseHeaders(), completion, session));
}
private StandardWebSocketSession createSession(URI url, HttpHeaders responseHeaders,
MonoProcessor<Void> 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);
});
}

38
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/UndertowWebSocketClient.java
View File

@@ -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<URI, ConnectionBuilder> 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<URI, ConnectionBuilder> 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<Void> executeInternal(URI url, HttpHeaders headers, WebSocketHandler handler) {
MonoProcessor<Void> completionMono = MonoProcessor.create();
MonoProcessor<Void> 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<Void> completion,
private void handleChannel(URI url, WebSocketHandler handler, MonoProcessor<Void> completion,
DefaultNegotiation negotiation, WebSocketChannel channel) {
HandshakeInfo info = afterHandshake(url, negotiation.getResponseHeaders());

2
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/WebSocketClient.java
View File

@@ -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

3
spring-web-reactive/src/main/java/org/springframework/web/reactive/socket/client/WebSocketClientSupport.java
View File

@@ -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());

10
spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/server/AbstractWebSocketIntegrationTests.java → spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/AbstractWebSocketIntegrationTests.java
View File

@@ -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
*/

6
spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/server/WebSocketIntegrationTests.java → spring-web-reactive/src/test/java/org/springframework/web/reactive/socket/WebSocketIntegrationTests.java
View File

@@ -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;