Add JettyWebSocketClient

Also split out JSR-356 related configuration and load it conditionally.

README-WEBSOCKET.md

@@ -54,6 +54,17 @@ Run the ant build:
 
 A usable Tomcat installation can be found in `output/build`
 
+### Jetty 9
+
+Download and use the latest Jetty (currently 9.0.2.v20130417). It does not support JSR-356 yet but that's not an issue, since we're using the Jetty 9 native WebSocket API.
+
+If using Java-based Servlet configuration instead of web.xml, add the following options to Jetty's start.ini:
+
+    OPTIONS=plus
+    etc/jetty-plus.xml
+    OPTIONS=annotations
+    etc/jetty-annotations.xml
+
 ### Glassfish
 
 Glassfish also provides JSR-356 support based on Tyrus (the reference implementation).

build.gradle

@@ -532,6 +532,7 @@ project("spring-websocket") {
 			exclude group: "org.eclipse.jetty.orbit", module: "javax.servlet"
 		}
 		optional("org.eclipse.jetty.websocket:websocket-server:9.0.1.v20130408")
+		optional("org.eclipse.jetty.websocket:websocket-client:9.0.1.v20130408")
 
 		optional("com.fasterxml.jackson.core:jackson-databind:2.0.1") // required for SockJS support currently
 

spring-websocket/src/main/java/org/springframework/websocket/client/AbstractWebSocketConnectionManager.java

@@ -40,6 +40,8 @@ public abstract class AbstractWebSocketConnectionManager implements SmartLifecyc
 
 	private boolean autoStartup = false;
 
+	private boolean isRunning = false;
+
 	private int phase = Integer.MAX_VALUE;
 
 	private TaskExecutor taskExecutor = new SimpleAsyncTaskExecutor("EndpointConnectionManager-");
@@ -92,6 +94,15 @@ public abstract class AbstractWebSocketConnectionManager implements SmartLifecyc
 		return this.uri;
 	}
 
+	/**
+	 * Return whether this ConnectionManager has been started.
+	 */
+	public boolean isRunning() {
+		synchronized (this.lifecycleMonitor) {
+			return this.isRunning;
+		}
+	}
+
 	/**
 	 * Connect to the configured {@link #setDefaultUri(URI) default URI}. If already
 	 * connected, the method has no impact.
@@ -99,61 +110,69 @@ public abstract class AbstractWebSocketConnectionManager implements SmartLifecyc
 	public final void start() {
 		synchronized (this.lifecycleMonitor) {
 			if (!isRunning()) {
-				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);
-							}
-						}
-					}
-				});
+				startInternal();
 			}
 		}
 	}
 
+	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);
+					}
+				}
+			}
+		});
+	}
+
 	protected abstract void openConnection() throws Exception;
 
-	/**
-	 * Closes the configured message WebSocket connection.
-	 */
 	public final void stop() {
 		synchronized (this.lifecycleMonitor) {
 			if (isRunning()) {
-				try {
-					closeConnection();
-				}
-				catch (Throwable e) {
-					logger.error("Failed to stop WebSocket connection", e);
-				}
+				stopInternal();
 			}
 		}
 	}
 
+	protected void stopInternal() {
+		if (logger.isDebugEnabled()) {
+			logger.debug("Stopping " + this.getClass().getSimpleName());
+		}
+		try {
+			if (isConnected()) {
+				closeConnection();
+			}
+		}
+		catch (Throwable e) {
+			logger.error("Failed to stop WebSocket connection", e);
+		}
+		finally {
+			this.isRunning = false;
+		}
+	}
+
+	protected abstract boolean isConnected();
+
 	protected abstract void closeConnection() throws Exception;
 
-	public void stop(Runnable callback) {
+	public final void stop(Runnable callback) {
 		synchronized (this.lifecycleMonitor) {
 			this.stop();
 			callback.run();
 		}
 	}
 
-	/**
-	 * Return whether the configured message endpoint is currently active.
-	 */
-	public boolean isRunning() {
-		synchronized (this.lifecycleMonitor) {
-			return isConnected();
-		}
-	}
-
-	protected abstract boolean isConnected();
-
 }

spring-websocket/src/main/java/org/springframework/websocket/client/WebSocketClient.java

@@ -22,10 +22,10 @@ import org.springframework.websocket.WebSocketHandler;
 import org.springframework.websocket.WebSocketSession;
 
 /**
- * Contract for starting a WebSocket handshake request.
- *
- * <p>To automatically start a WebSocket connection when the application starts, see
- * {@link WebSocketConnectionManager}.
+ * Contract for programmatically starting a WebSocket handshake request. For most cases it
+ * would be more convenient to use the declarative style
+ * {@link WebSocketConnectionManager} that starts a WebSocket connection to a
+ * pre-configured URI when the application starts.
  *
  * @author Rossen Stoyanchev
  * @since 4.0

spring-websocket/src/main/java/org/springframework/websocket/client/WebSocketConnectionManager.java

@@ -19,6 +19,7 @@ package org.springframework.websocket.client;
 import java.util.ArrayList;
 import java.util.List;
 
+import org.springframework.context.SmartLifecycle;
 import org.springframework.http.HttpHeaders;
 import org.springframework.util.CollectionUtils;
 import org.springframework.websocket.WebSocketHandler;
@@ -40,13 +41,16 @@ public class WebSocketConnectionManager extends AbstractWebSocketConnectionManag
 
 	private final List<String> subProtocols = new ArrayList<String>();
 
+	private final boolean syncClientLifecycle;
 
-	public WebSocketConnectionManager(WebSocketClient webSocketClient,
+
+	public WebSocketConnectionManager(WebSocketClient client,
 			WebSocketHandler webSocketHandler, String uriTemplate, Object... uriVariables) {
 
 		super(uriTemplate, uriVariables);
-		this.client = webSocketClient;
+		this.client = client;
 		this.webSocketHandler = decorateWebSocketHandler(webSocketHandler);
+		this.syncClientLifecycle = ((client instanceof SmartLifecycle) && !((SmartLifecycle) client).isRunning());
 	}
 
 	/**
@@ -71,6 +75,22 @@ public class WebSocketConnectionManager extends AbstractWebSocketConnectionManag
 		return this.subProtocols;
 	}
 
+	@Override
+	public void startInternal() {
+		if (this.syncClientLifecycle) {
+			((SmartLifecycle) this.client).start();
+		}
+		super.startInternal();
+	}
+
+	@Override
+	public void stopInternal() {
+		if (this.syncClientLifecycle) {
+			((SmartLifecycle) client).stop();
+		}
+		super.stopInternal();
+	}
+
 	@Override
 	protected void openConnection() throws Exception {
 		HttpHeaders headers = new HttpHeaders();

spring-websocket/src/main/java/org/springframework/websocket/client/jetty/JettyWebSocketClient.java

@@ -0,0 +1,150 @@
+/*
+ * 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.websocket.client.jetty;
+
+import java.net.URI;
+import java.util.concurrent.Future;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.eclipse.jetty.websocket.api.Session;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.http.HttpHeaders;
+import org.springframework.web.util.UriComponentsBuilder;
+import org.springframework.websocket.WebSocketHandler;
+import org.springframework.websocket.WebSocketSession;
+import org.springframework.websocket.adapter.JettyWebSocketListenerAdapter;
+import org.springframework.websocket.adapter.JettyWebSocketSessionAdapter;
+import org.springframework.websocket.client.WebSocketClient;
+import org.springframework.websocket.client.WebSocketConnectFailureException;
+
+
+/**
+ * @author Rossen Stoyanchev
+ * @since 4.0
+ */
+public class JettyWebSocketClient implements WebSocketClient, SmartLifecycle {
+
+	private static final Log logger = LogFactory.getLog(JettyWebSocketClient.class);
+
+	private final org.eclipse.jetty.websocket.client.WebSocketClient client;
+
+	private boolean autoStartup = true;
+
+	private int phase = Integer.MAX_VALUE;
+
+	private final Object lifecycleMonitor = new Object();
+
+
+	public JettyWebSocketClient() {
+		this.client = new org.eclipse.jetty.websocket.client.WebSocketClient();
+	}
+
+
+	// TODO: configure Jetty WebSocketClient properties
+
+	public void setAutoStartup(boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
+	@Override
+	public boolean isAutoStartup() {
+		return this.autoStartup;
+	}
+
+	public void setPhase(int phase) {
+		this.phase = phase;
+	}
+
+	@Override
+	public int getPhase() {
+		return this.phase;
+	}
+
+	@Override
+	public boolean isRunning() {
+		synchronized (this.lifecycleMonitor) {
+			return this.client.isStarted();
+		}
+	}
+
+	@Override
+	public void start() {
+		synchronized (this.lifecycleMonitor) {
+			if (!isRunning()) {
+				try {
+					if (logger.isDebugEnabled()) {
+						logger.debug("Starting Jetty WebSocketClient");
+					}
+					this.client.start();
+				}
+				catch (Exception e) {
+					throw new IllegalStateException("Failed to start Jetty client", e);
+				}
+			}
+		}
+	}
+
+	@Override
+	public void stop() {
+		synchronized (this.lifecycleMonitor) {
+			if (isRunning()) {
+				try {
+					if (logger.isDebugEnabled()) {
+						logger.debug("Stopping Jetty WebSocketClient");
+					}
+					this.client.stop();
+				}
+				catch (Exception e) {
+					logger.error("Error stopping Jetty WebSocketClient", e);
+				}
+			}
+		}
+	}
+
+	@Override
+	public void stop(Runnable callback) {
+		this.stop();
+		callback.run();
+	}
+
+	@Override
+	public WebSocketSession doHandshake(WebSocketHandler webSocketHandler, String uriTemplate, Object... uriVariables)
+			throws WebSocketConnectFailureException {
+
+		URI uri = UriComponentsBuilder.fromUriString(uriTemplate).buildAndExpand(uriVariables).encode().toUri();
+		return doHandshake(webSocketHandler, null, uri);
+	}
+
+	@Override
+	public WebSocketSession doHandshake(WebSocketHandler webSocketHandler, HttpHeaders headers, URI uri)
+			throws WebSocketConnectFailureException {
+
+		JettyWebSocketListenerAdapter listener = new JettyWebSocketListenerAdapter(webSocketHandler);
+
+		try {
+			// block for now
+			Future<org.eclipse.jetty.websocket.api.Session> future = this.client.connect(listener, uri);
+			Session session = future.get();
+			return new JettyWebSocketSessionAdapter(session);
+		}
+		catch (Exception e) {
+			throw new WebSocketConnectFailureException("Failed to connect to " + uri, e);
+		}
+	}
+
+}

spring-websocket/src/main/java/org/springframework/websocket/server/support/JettyRequestUpgradeStrategy.java

@@ -100,10 +100,14 @@ public class JettyRequestUpgradeStrategy implements RequestUpgradeStrategy {
 	private void upgrade(HttpServletRequest request, HttpServletResponse response,
 			String selectedProtocol, final WebSocketHandler webSocketHandler) throws IOException {
 
-		Assert.state(this.factory.isUpgradeRequest(request, response), "Not a suitable WebSocket upgrade request");
-		Assert.state(this.factory.acceptWebSocket(request, response), "Unable to accept WebSocket");
+		Assert.state(this.factory.isUpgradeRequest(request, response), "Expected websocket upgrade request");
 
 		request.setAttribute(HANDLER_PROVIDER_ATTR_NAME, webSocketHandler);
+
+		if (!this.factory.acceptWebSocket(request, response)) {
+			// should never happen
+			throw new IllegalStateException("WebSocket request not accepted by Jetty");
+		}
 	}
 
 }

spring-websocket/src/main/java/org/springframework/websocket/support/PerConnectionWebSocketHandlerProxy.java

@@ -125,4 +125,9 @@ public class PerConnectionWebSocketHandlerProxy implements WebSocketHandler, Bea
 		}
 	}
 
+	@Override
+	public String toString() {
+		return "PerConnectionWebSocketHandlerProxy [handlerType=" + this.provider.getHandlerType() + "]";
+	}
+
 }

spring-websocket/src/main/java/org/springframework/websocket/support/WebSocketHandlerDecorator.java

@@ -67,4 +67,9 @@ public class WebSocketHandlerDecorator implements WebSocketHandler {
 		return this.delegate.isStreaming();
 	}
 
+	@Override
+	public String toString() {
+		return getClass().getSimpleName() + " [delegate=" + this.delegate + "]";
+	}
+
 }