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

Introduce ListenableFuture to WebSocketClient

Browse Source 
Issue: SPR-10888
This commit is contained in:
Rossen Stoyanchev
2013-09-06 12:28:21 -04:00
parent 71e76196fe
commit 62921683fd
16 changed files with 234 additions and 149 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
spring-websocket/src/main/java/org/springframework/web/socket/client/AbstractWebSocketClient.java
View File

@@ -28,6 +28,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.util.Assert;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.util.UriComponentsBuilder;
@@ -60,8 +61,8 @@ public abstract class AbstractWebSocketClient implements WebSocketClient {
@Override
public WebSocketSession doHandshake(WebSocketHandler webSocketHandler, String uriTemplate,
Object... uriVars) throws WebSocketConnectFailureException {
public ListenableFuture<WebSocketSession> doHandshake(WebSocketHandler webSocketHandler,
String uriTemplate, Object... uriVars) {
Assert.notNull(uriTemplate, "uriTemplate must not be null");
URI uri = UriComponentsBuilder.fromUriString(uriTemplate).buildAndExpand(uriVars).encode().toUri();
@@ -69,8 +70,8 @@ public abstract class AbstractWebSocketClient implements WebSocketClient {
}
@Override
public final WebSocketSession doHandshake(WebSocketHandler webSocketHandler,
HttpHeaders headers, URI uri) throws WebSocketConnectFailureException {
public final ListenableFuture<WebSocketSession> doHandshake(WebSocketHandler webSocketHandler,
HttpHeaders headers, URI uri) {
Assert.notNull(webSocketHandler, "webSocketHandler must not be null");
Assert.notNull(uri, "uri must not be null");
@@ -111,12 +112,9 @@ public abstract class AbstractWebSocketClient implements WebSocketClient {
* @param handshakeAttributes attributes to make available via
* {@link WebSocketSession#getHandshakeAttributes()}; currently always an empty map.
*
* @return the established WebSocket session
*
* @throws WebSocketConnectFailureException
* @return the established WebSocket session wrapped in a ListenableFuture.
*/
protected abstract WebSocketSession doHandshakeInternal(WebSocketHandler webSocketHandler,
HttpHeaders headers, URI uri, List<String> subProtocols,
Map<String, Object> handshakeAttributes) throws WebSocketConnectFailureException;
protected abstract ListenableFuture<WebSocketSession> doHandshakeInternal(WebSocketHandler webSocketHandler,
HttpHeaders headers, URI uri, List<String> subProtocols, Map<String, Object> handshakeAttributes);
}

30
spring-websocket/src/main/java/org/springframework/web/socket/client/ConnectionManagerSupport.java
View File

@@ -21,8 +21,6 @@ import java.net.URI;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.web.util.UriComponentsBuilder;
/**
@@ -48,8 +46,6 @@ public abstract class ConnectionManagerSupport implements SmartLifecycle {
private int phase = Integer.MAX_VALUE;
private final TaskExecutor taskExecutor = new SimpleAsyncTaskExecutor("EndpointConnectionManager-");
private final Object lifecycleMonitor = new Object();
@@ -126,28 +122,16 @@ public abstract class ConnectionManagerSupport implements SmartLifecycle {
}
protected void startInternal() {
if (logger.isDebugEnabled()) {
logger.debug("Starting " + this.getClass().getSimpleName());
}
this.isRunning = true;
this.taskExecutor.execute(new Runnable() {
@Override
public void run() {
synchronized (lifecycleMonitor) {
try {
logger.info("Connecting to WebSocket at " + uri);
openConnection();
logger.info("Successfully connected");
}
catch (Throwable ex) {
logger.error("Failed to connect", ex);
}
}
synchronized (lifecycleMonitor) {
if (logger.isDebugEnabled()) {
logger.debug("Starting " + this.getClass().getSimpleName());
}
});
this.isRunning = true;
openConnection();
}
}
protected abstract void openConnection() throws Exception;
protected abstract void openConnection();
@Override
public final void stop() {

8
spring-websocket/src/main/java/org/springframework/web/socket/client/WebSocketClient.java
View File

@@ -19,6 +19,7 @@ package org.springframework.web.socket.client;
import java.net.URI;
import org.springframework.http.HttpHeaders;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.WebSocketSession;
@@ -34,10 +35,9 @@ import org.springframework.web.socket.WebSocketSession;
*/
public interface WebSocketClient {
WebSocketSession doHandshake(WebSocketHandler webSocketHandler,
String uriTemplate, Object... uriVariables) throws WebSocketConnectFailureException;
ListenableFuture<WebSocketSession> doHandshake(WebSocketHandler webSocketHandler,
String uriTemplate, Object... uriVariables);
WebSocketSession doHandshake(WebSocketHandler webSocketHandler, HttpHeaders headers, URI uri)
throws WebSocketConnectFailureException;
ListenableFuture<WebSocketSession> doHandshake(WebSocketHandler webSocketHandler, HttpHeaders headers, URI uri);
}

38
spring-websocket/src/main/java/org/springframework/web/socket/client/WebSocketConnectFailureException.java
View File

@@ -1,38 +0,0 @@
/*
* Copyright 2002-2013 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.web.socket.client;
import org.springframework.core.NestedRuntimeException;
/**
* Thrown when a WebSocket connection to a server could not be established.
*
* @author Rossen Stoyanchev
* @since 4.0
*/
@SuppressWarnings("serial")
public class WebSocketConnectFailureException extends NestedRuntimeException {
public WebSocketConnectFailureException(String msg, Throwable cause) {
super(msg, cause);
}
public WebSocketConnectFailureException(String msg) {
super(msg);
}
}

22
spring-websocket/src/main/java/org/springframework/web/socket/client/WebSocketConnectionManager.java
View File

@@ -20,6 +20,8 @@ import java.util.List;
import org.springframework.context.SmartLifecycle;
import org.springframework.http.HttpHeaders;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.support.LoggingWebSocketHandlerDecorator;
@@ -129,8 +131,24 @@ public class WebSocketConnectionManager extends ConnectionManagerSupport {
}
@Override
protected void openConnection() throws Exception {
this.webSocketSession = this.client.doHandshake(this.webSocketHandler, this.headers, getUri());
protected void openConnection() {
logger.info("Connecting to WebSocket at " + getUri());
ListenableFuture<WebSocketSession> future =
this.client.doHandshake(this.webSocketHandler, this.headers, getUri());
future.addCallback(new ListenableFutureCallback<WebSocketSession>() {
@Override
public void onSuccess(WebSocketSession result) {
webSocketSession = result;
logger.info("Successfully connected");
}
@Override
public void onFailure(Throwable t) {
logger.error("Failed to connect", t);
}
});
}
@Override

39
spring-websocket/src/main/java/org/springframework/web/socket/client/endpoint/AnnotatedEndpointConnectionManager.java
View File

@@ -24,6 +24,9 @@ import javax.websocket.server.ServerEndpoint;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.util.Assert;
import org.springframework.web.socket.client.ConnectionManagerSupport;
import org.springframework.web.socket.support.BeanCreatingHandlerProvider;
@@ -46,6 +49,8 @@ public class AnnotatedEndpointConnectionManager extends ConnectionManagerSupport
private Session session;
private TaskExecutor taskExecutor = new SimpleAsyncTaskExecutor("AnnotatedEndpointConnectionManager-");
public AnnotatedEndpointConnectionManager(Object endpoint, String uriTemplate, Object... uriVariables) {
super(uriTemplate, uriVariables);
@@ -75,10 +80,38 @@ public class AnnotatedEndpointConnectionManager extends ConnectionManagerSupport
}
}
/**
* Set a {@link TaskExecutor} to use to open the connection.
* By default {@link SimpleAsyncTaskExecutor} is used.
*/
public void setTaskExecutor(TaskExecutor taskExecutor) {
Assert.notNull(taskExecutor, "taskExecutor is required");
this.taskExecutor = taskExecutor;
}
/**
* Return the configured {@link TaskExecutor}.
*/
public TaskExecutor getTaskExecutor() {
return this.taskExecutor;
}
@Override
protected void openConnection() throws Exception {
Object endpoint = (this.endpoint != null) ? this.endpoint : this.endpointProvider.getHandler();
this.session = this.webSocketContainer.connectToServer(endpoint, getUri());
protected void openConnection() {
this.taskExecutor.execute(new Runnable() {
@Override
public void run() {
try {
logger.info("Connecting to WebSocket at " + getUri());
Object endpointToUse = (endpoint != null) ? endpoint : endpointProvider.getHandler();
session = webSocketContainer.connectToServer(endpointToUse, getUri());
logger.info("Successfully connected");
}
catch (Throwable ex) {
logger.error("Failed to connect", ex);
}
}
});
}
@Override

41
spring-websocket/src/main/java/org/springframework/web/socket/client/endpoint/EndpointConnectionManager.java
View File

@@ -32,6 +32,8 @@ import javax.websocket.WebSocketContainer;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.util.Assert;
import org.springframework.web.socket.client.ConnectionManagerSupport;
import org.springframework.web.socket.support.BeanCreatingHandlerProvider;
@@ -58,6 +60,8 @@ public class EndpointConnectionManager extends ConnectionManagerSupport implemen
private Session session;
private TaskExecutor taskExecutor = new SimpleAsyncTaskExecutor("EndpointConnectionManager-");
public EndpointConnectionManager(Endpoint endpoint, String uriTemplate, Object... uriVariables) {
super(uriTemplate, uriVariables);
@@ -109,11 +113,40 @@ public class EndpointConnectionManager extends ConnectionManagerSupport implemen
}
}
/**
* Set a {@link TaskExecutor} to use to open connections.
* By default {@link SimpleAsyncTaskExecutor} is used.
*/
public void setTaskExecutor(TaskExecutor taskExecutor) {
Assert.notNull(taskExecutor, "taskExecutor is required");
this.taskExecutor = taskExecutor;
}
/**
* Return the configured {@link TaskExecutor}.
*/
public TaskExecutor getTaskExecutor() {
return this.taskExecutor;
}
@Override
protected void openConnection() throws Exception {
Endpoint endpoint = (this.endpoint != null) ? this.endpoint : this.endpointProvider.getHandler();
ClientEndpointConfig endpointConfig = this.configBuilder.build();
this.session = getWebSocketContainer().connectToServer(endpoint, endpointConfig, getUri());
protected void openConnection() {
this.taskExecutor.execute(new Runnable() {
@Override
public void run() {
try {
logger.info("Connecting to WebSocket at " + getUri());
Endpoint endpointToUse = (endpoint != null) ? endpoint : endpointProvider.getHandler();
ClientEndpointConfig endpointConfig = configBuilder.build();
session = getWebSocketContainer().connectToServer(endpointToUse, endpointConfig, getUri());
logger.info("Successfully connected");
}
catch (Throwable ex) {
logger.error("Failed to connect", ex);
}
}
});
}
@Override

49
spring-websocket/src/main/java/org/springframework/web/socket/client/endpoint/StandardWebSocketClient.java
View File

@@ -23,6 +23,7 @@ import java.net.UnknownHostException;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.Callable;
import javax.websocket.ClientEndpointConfig;
import javax.websocket.ClientEndpointConfig.Configurator;
@@ -31,14 +32,17 @@ import javax.websocket.Endpoint;
import javax.websocket.HandshakeResponse;
import javax.websocket.WebSocketContainer;
import org.springframework.core.task.AsyncListenableTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.http.HttpHeaders;
import org.springframework.util.Assert;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.adapter.StandardWebSocketHandlerAdapter;
import org.springframework.web.socket.adapter.StandardWebSocketSession;
import org.springframework.web.socket.client.AbstractWebSocketClient;
import org.springframework.web.socket.client.WebSocketConnectFailureException;
/**
* Initiates WebSocket requests to a WebSocket server programatically through the standard
@@ -51,6 +55,9 @@ public class StandardWebSocketClient extends AbstractWebSocketClient {
private final WebSocketContainer webSocketContainer;
private AsyncListenableTaskExecutor taskExecutor =
new SimpleAsyncTaskExecutor("WebSocketClient-");
/**
* Default constructor that calls {@code ContainerProvider.getWebSocketContainer()} to
@@ -71,31 +78,45 @@ public class StandardWebSocketClient extends AbstractWebSocketClient {
this.webSocketContainer = webSocketContainer;
}
/**
* Set a {@link TaskExecutor} to use to open the connection.
* By default {@link SimpleAsyncTaskExecutor} is used.
*/
public void setTaskExecutor(AsyncListenableTaskExecutor taskExecutor) {
Assert.notNull(taskExecutor, "taskExecutor is required");
this.taskExecutor = taskExecutor;
}
/**
* Return the configured {@link TaskExecutor}.
*/
public AsyncListenableTaskExecutor getTaskExecutor() {
return this.taskExecutor;
}
@Override
protected WebSocketSession doHandshakeInternal(WebSocketHandler webSocketHandler,
HttpHeaders headers, URI uri, List<String> protocols,
Map<String, Object> handshakeAttributes) throws WebSocketConnectFailureException {
protected ListenableFuture<WebSocketSession> doHandshakeInternal(WebSocketHandler webSocketHandler,
HttpHeaders headers, final URI uri, List<String> protocols, Map<String, Object> handshakeAttributes) {
int port = getPort(uri);
InetSocketAddress localAddress = new InetSocketAddress(getLocalHost(), port);
InetSocketAddress remoteAddress = new InetSocketAddress(uri.getHost(), port);
StandardWebSocketSession session = new StandardWebSocketSession(headers,
final StandardWebSocketSession session = new StandardWebSocketSession(headers,
handshakeAttributes, localAddress, remoteAddress);
ClientEndpointConfig.Builder configBuidler = ClientEndpointConfig.Builder.create();
final ClientEndpointConfig.Builder configBuidler = ClientEndpointConfig.Builder.create();
configBuidler.configurator(new StandardWebSocketClientConfigurator(headers));
configBuidler.preferredSubprotocols(protocols);
final Endpoint endpoint = new StandardWebSocketHandlerAdapter(webSocketHandler, session);
try {
Endpoint endpoint = new StandardWebSocketHandlerAdapter(webSocketHandler, session);
this.webSocketContainer.connectToServer(endpoint, configBuidler.build(), uri);
return session;
}
catch (Exception e) {
throw new WebSocketConnectFailureException("Failed to connect to " + uri, e);
}
return this.taskExecutor.submitListenable(new Callable<WebSocketSession>() {
@Override
public WebSocketSession call() throws Exception {
webSocketContainer.connectToServer(endpoint, configBuidler.build(), uri);
return session;
}
});
}
private InetAddress getLocalHost() {

57
spring-websocket/src/main/java/org/springframework/web/socket/client/jetty/JettyWebSocketClient.java
View File

@@ -20,19 +20,24 @@ import java.net.URI;
import java.security.Principal;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
import org.eclipse.jetty.websocket.client.WebSocketClient;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.task.AsyncListenableTaskExecutor;
import org.springframework.core.task.SimpleAsyncTaskExecutor;
import org.springframework.core.task.TaskExecutor;
import org.springframework.http.HttpHeaders;
import org.springframework.util.Assert;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.socket.WebSocketHandler;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.adapter.JettyWebSocketHandlerAdapter;
import org.springframework.web.socket.adapter.JettyWebSocketSession;
import org.springframework.web.socket.client.AbstractWebSocketClient;
import org.springframework.web.socket.client.WebSocketConnectFailureException;
import org.springframework.web.util.UriComponents;
import org.springframework.web.util.UriComponentsBuilder;
@@ -53,6 +58,8 @@ public class JettyWebSocketClient extends AbstractWebSocketClient implements Sma
private final Object lifecycleMonitor = new Object();
private AsyncListenableTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor("WebSocketClient-");
/**
* Default constructor that creates an instance of
@@ -71,6 +78,22 @@ public class JettyWebSocketClient extends AbstractWebSocketClient implements Sma
}
/**
* Set a {@link TaskExecutor} to use to open the connection.
* By default {@link SimpleAsyncTaskExecutor} is used.
*/
public void setTaskExecutor(AsyncListenableTaskExecutor taskExecutor) {
Assert.notNull(taskExecutor, "taskExecutor is required");
this.taskExecutor = taskExecutor;
}
/**
* Return the configured {@link TaskExecutor}.
*/
public AsyncListenableTaskExecutor getTaskExecutor() {
return this.taskExecutor;
}
public void setAutoStartup(boolean autoStartup) {
this.autoStartup = autoStartup;
}
@@ -137,39 +160,37 @@ public class JettyWebSocketClient extends AbstractWebSocketClient implements Sma
}
@Override
public WebSocketSession doHandshake(WebSocketHandler webSocketHandler, String uriTemplate, Object... uriVars)
throws WebSocketConnectFailureException {
public ListenableFuture<WebSocketSession> doHandshake(WebSocketHandler webSocketHandler,
String uriTemplate, Object... uriVars) {
UriComponents uriComponents = UriComponentsBuilder.fromUriString(uriTemplate).buildAndExpand(uriVars).encode();
return doHandshake(webSocketHandler, null, uriComponents.toUri());
}
@Override
public WebSocketSession doHandshakeInternal(WebSocketHandler wsHandler, HttpHeaders headers,
URI uri, List<String> protocols, Map<String, Object> handshakeAttributes)
throws WebSocketConnectFailureException {
public ListenableFuture<WebSocketSession> doHandshakeInternal(WebSocketHandler wsHandler,
HttpHeaders headers, final URI uri, List<String> protocols, Map<String, Object> handshakeAttributes) {
ClientUpgradeRequest request = new ClientUpgradeRequest();
final ClientUpgradeRequest request = new ClientUpgradeRequest();
request.setSubProtocols(protocols);
for (String header : headers.keySet()) {
request.setHeader(header, headers.get(header));
}
Principal user = getUser();
JettyWebSocketSession wsSession = new JettyWebSocketSession(user, handshakeAttributes);
JettyWebSocketHandlerAdapter listener = new JettyWebSocketHandlerAdapter(wsHandler, wsSession);
final JettyWebSocketSession wsSession = new JettyWebSocketSession(user, handshakeAttributes);
final JettyWebSocketHandlerAdapter listener = new JettyWebSocketHandlerAdapter(wsHandler, wsSession);
try {
Future<Session> future = this.client.connect(listener, uri, request);
future.get();
return wsSession;
}
catch (Exception e) {
throw new WebSocketConnectFailureException("Failed to connect to " + uri, e);
}
return this.taskExecutor.submitListenable(new Callable<WebSocketSession>() {
@Override
public WebSocketSession call() throws Exception {
Future<Session> future = client.connect(listener, uri, request);
future.get();
return wsSession;
}
});
}
/**
* @return the user to make available through {@link WebSocketSession#getPrincipal()};
* by default this method returns {@code null}