From f9078c947fb5ef1fb4627015f2f5e355294c085e Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Mon, 22 Apr 2013 17:59:19 -0400 Subject: [PATCH] Add WebSocketMessage and WebSocketHandler sub-interfcs There is now a WebSocketMessage type with TextMessage and BinaryMessage sub-types. WebSocketHandler is also sub-divided into TextMessageHandler and BinaryMessageHandler, so that applications can choose to handle text, binary, or both. Also in this commit, the SockJsHandler and SockJsSession interfaces have been removed. SockJsService now accepts WebSocketHandler. --- ...upport.java => AbstractSockJsSession.java} | 51 ++++--- .../springframework/sockjs/SockJsHandler.java | 51 ------- .../springframework/sockjs/SockJsSession.java | 62 -------- .../sockjs/SockJsSessionFactory.java | 7 +- .../server/AbstractServerSockJsSession.java | 34 +++-- .../sockjs/server/AbstractSockJsService.java | 12 +- .../server/ConfigurableTransportHandler.java | 11 -- .../sockjs/server/SockJsService.java | 12 +- .../sockjs/server/TransportHandler.java | 6 +- .../server/support/DefaultSockJsService.java | 65 ++------- .../support/SockJsHttpRequestHandler.java | 22 ++- ...AbstractHttpReceivingTransportHandler.java | 8 +- .../AbstractHttpSendingTransportHandler.java | 14 +- .../AbstractHttpServerSockJsSession.java | 16 ++- .../AbstractStreamingTransportHandler.java | 6 +- .../JsonpPollingTransportHandler.java | 6 +- .../transport/JsonpTransportHandler.java | 4 +- .../transport/PollingServerSockJsSession.java | 14 +- .../transport/SockJsWebSocketHandler.java | 60 ++++---- .../StreamingServerSockJsSession.java | 10 +- .../WebSocketSockJsHandlerAdapter.java | 132 ------------------ .../transport/WebSocketTransportHandler.java | 54 +------ .../transport/XhrPollingTransportHandler.java | 6 +- .../websocket/BinaryMessage.java | 85 +++++++++++ .../BinaryMessageHandler.java} | 28 ++-- .../websocket/PartialMessageHandler.java | 28 ++++ ...HandshakeRequest.java => TextMessage.java} | 25 ++-- .../websocket/TextMessageHandler.java | 40 ++++++ .../websocket/WebSocketHandler.java | 16 +-- .../websocket/WebSocketHandlerAdapter.java | 20 ++- .../websocket/WebSocketMessage.java | 61 ++++++++ .../websocket/WebSocketSession.java | 16 +-- .../endpoint/StandardWebSocketSession.java | 27 ++-- .../endpoint/WebSocketHandlerEndpoint.java | 105 +++++++++----- .../server/DefaultHandshakeHandler.java | 6 - .../websocket/server/HandshakeHandler.java | 10 -- .../server/RequestUpgradeStrategy.java | 10 -- .../AbstractEndpointUpgradeStrategy.java | 32 +---- .../support/WebSocketHttpRequestHandler.java | 7 +- 39 files changed, 513 insertions(+), 666 deletions(-) rename spring-websocket/src/main/java/org/springframework/sockjs/{SockJsSessionSupport.java => AbstractSockJsSession.java} (74%) delete mode 100644 spring-websocket/src/main/java/org/springframework/sockjs/SockJsHandler.java delete mode 100644 spring-websocket/src/main/java/org/springframework/sockjs/SockJsSession.java delete mode 100644 spring-websocket/src/main/java/org/springframework/sockjs/server/transport/WebSocketSockJsHandlerAdapter.java create mode 100644 spring-websocket/src/main/java/org/springframework/websocket/BinaryMessage.java rename spring-websocket/src/main/java/org/springframework/{sockjs/SockJsHandlerAdapter.java => websocket/BinaryMessageHandler.java} (52%) create mode 100644 spring-websocket/src/main/java/org/springframework/websocket/PartialMessageHandler.java rename spring-websocket/src/main/java/org/springframework/websocket/{WebSocketHandshakeRequest.java => TextMessage.java} (63%) create mode 100644 spring-websocket/src/main/java/org/springframework/websocket/TextMessageHandler.java create mode 100644 spring-websocket/src/main/java/org/springframework/websocket/WebSocketMessage.java diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/SockJsSessionSupport.java b/spring-websocket/src/main/java/org/springframework/sockjs/AbstractSockJsSession.java similarity index 74% rename from spring-websocket/src/main/java/org/springframework/sockjs/SockJsSessionSupport.java rename to spring-websocket/src/main/java/org/springframework/sockjs/AbstractSockJsSession.java index 3cf7e61fc1..1f3e82541e 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/SockJsSessionSupport.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/AbstractSockJsSession.java @@ -16,12 +16,16 @@ package org.springframework.sockjs; -import java.io.IOException; +import java.net.URI; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.util.Assert; import org.springframework.websocket.CloseStatus; +import org.springframework.websocket.TextMessage; +import org.springframework.websocket.TextMessageHandler; +import org.springframework.websocket.WebSocketHandler; +import org.springframework.websocket.WebSocketSession; /** @@ -30,13 +34,13 @@ import org.springframework.websocket.CloseStatus; * @author Rossen Stoyanchev * @since 4.0 */ -public abstract class SockJsSessionSupport implements SockJsSession { +public abstract class AbstractSockJsSession implements WebSocketSession { protected Log logger = LogFactory.getLog(this.getClass()); private final String sessionId; - private final SockJsHandler sockJsHandler; + private final TextMessageHandler handler; private State state = State.NEW; @@ -48,19 +52,32 @@ public abstract class SockJsSessionSupport implements SockJsSession { /** * * @param sessionId - * @param sockJsHandler the recipient of SockJS messages + * @param handler the recipient of SockJS messages */ - public SockJsSessionSupport(String sessionId, SockJsHandler sockJsHandler) { + public AbstractSockJsSession(String sessionId, WebSocketHandler webSocketHandler) { Assert.notNull(sessionId, "sessionId is required"); - Assert.notNull(sockJsHandler, "sockJsHandler is required"); + Assert.notNull(webSocketHandler, "webSocketHandler is required"); + Assert.isInstanceOf(TextMessageHandler.class, webSocketHandler, "Expected a TextMessageHandler"); this.sessionId = sessionId; - this.sockJsHandler = sockJsHandler; + this.handler = (TextMessageHandler) webSocketHandler; } public String getId() { return this.sessionId; } + @Override + public boolean isSecure() { + // TODO + return false; + } + + @Override + public URI getURI() { + // TODO + return null; + } + public boolean isNew() { return State.NEW.equals(this.state); } @@ -104,26 +121,26 @@ public abstract class SockJsSessionSupport implements SockJsSession { public void delegateConnectionEstablished() throws Exception { this.state = State.OPEN; - this.sockJsHandler.afterConnectionEstablished(this); + this.handler.afterConnectionEstablished(this); } public void delegateMessages(String[] messages) throws Exception { for (String message : messages) { - this.sockJsHandler.handleMessage(message, this); + this.handler.handleTextMessage(new TextMessage(message), this); } } public void delegateError(Throwable ex) { - this.sockJsHandler.handleError(ex, this); + this.handler.handleError(ex, this); } /** * Invoked in reaction to the underlying connection being closed by the remote side * (or the WebSocket container) in order to perform cleanup and notify the - * {@link SockJsHandler}. This is in contrast to {@link #close()} that pro-actively + * {@link TextMessageHandler}. This is in contrast to {@link #close()} that pro-actively * closes the connection. */ - public final void delegateConnectionClosed(CloseStatus status) { + public final void delegateConnectionClosed(CloseStatus status) throws Exception { if (!isClosed()) { if (logger.isDebugEnabled()) { logger.debug(this + " was closed, " + status); @@ -133,7 +150,7 @@ public abstract class SockJsSessionSupport implements SockJsSession { } finally { this.state = State.CLOSED; - this.sockJsHandler.afterConnectionClosed(status, this); + this.handler.afterConnectionClosed(status, this); } } } @@ -145,7 +162,7 @@ public abstract class SockJsSessionSupport implements SockJsSession { * {@inheritDoc} *

Performs cleanup and notifies the {@link SockJsHandler}. */ - public final void close() throws IOException { + public final void close() throws Exception { close(CloseStatus.NORMAL); } @@ -153,7 +170,7 @@ public abstract class SockJsSessionSupport implements SockJsSession { * {@inheritDoc} *

Performs cleanup and notifies the {@link SockJsHandler}. */ - public final void close(CloseStatus status) throws IOException { + public final void close(CloseStatus status) throws Exception { if (!isClosed()) { if (logger.isDebugEnabled()) { logger.debug("Closing " + this + ", " + status); @@ -163,12 +180,12 @@ public abstract class SockJsSessionSupport implements SockJsSession { } finally { this.state = State.CLOSED; - this.sockJsHandler.afterConnectionClosed(status, this); + this.handler.afterConnectionClosed(status, this); } } } - protected abstract void closeInternal(CloseStatus status) throws IOException; + protected abstract void closeInternal(CloseStatus status) throws Exception; @Override diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/SockJsHandler.java b/spring-websocket/src/main/java/org/springframework/sockjs/SockJsHandler.java deleted file mode 100644 index e8c20e573f..0000000000 --- a/spring-websocket/src/main/java/org/springframework/sockjs/SockJsHandler.java +++ /dev/null @@ -1,51 +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.sockjs; - -import org.springframework.websocket.CloseStatus; - - - -/** - * A handler for SockJS messages. - * - * @author Rossen Stoyanchev - * @since 4.0 - */ -public interface SockJsHandler { - - /** - * A new connection was opened and is ready for use. - */ - void afterConnectionEstablished(SockJsSession session) throws Exception; - - /** - * Handle an incoming message. - */ - void handleMessage(String message, SockJsSession session) throws Exception; - - /** - * TODO - */ - void handleError(Throwable exception, SockJsSession session); - - /** - * A connection has been closed. - */ - void afterConnectionClosed(CloseStatus status, SockJsSession session); - -} diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/SockJsSession.java b/spring-websocket/src/main/java/org/springframework/sockjs/SockJsSession.java deleted file mode 100644 index e1e46155cd..0000000000 --- a/spring-websocket/src/main/java/org/springframework/sockjs/SockJsSession.java +++ /dev/null @@ -1,62 +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.sockjs; - -import java.io.IOException; - -import org.springframework.websocket.CloseStatus; - - - -/** - * Allows sending SockJS messages as well as closing the underlying connection. - * - * @author Rossen Stoyanchev - * @since 4.0 - */ -public interface SockJsSession { - - /** - * Return a unique SockJS session identifier. - */ - String getId(); - - /** - * Return whether the connection is still open. - */ - boolean isOpen(); - - /** - * Send a message. - */ - void sendMessage(String text) throws IOException; - - /** - * Close the underlying connection with status 1000, i.e. equivalent to: - * 

-	 * session.close(CloseStatus.NORMAL);
-	 *
- */ - void close() throws IOException; - - /** - * Close the underlying connection with the given close status. - */ - void close(CloseStatus status) throws IOException; - - -} diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/SockJsSessionFactory.java b/spring-websocket/src/main/java/org/springframework/sockjs/SockJsSessionFactory.java index 4e4950181d..c87ff5a7b9 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/SockJsSessionFactory.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/SockJsSessionFactory.java @@ -16,6 +16,9 @@ package org.springframework.sockjs; +import org.springframework.websocket.WebSocketHandler; +import org.springframework.websocket.WebSocketSession; + /** * A factory for creating a SockJS session. @@ -23,8 +26,8 @@ package org.springframework.sockjs; * @author Rossen Stoyanchev * @since 4.0 */ -public interface SockJsSessionFactory{ +public interface SockJsSessionFactory{ - S createSession(String sessionId, SockJsHandler sockJsHandler); + S createSession(String sessionId, WebSocketHandler webSocketHandler); } diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/AbstractServerSockJsSession.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/AbstractServerSockJsSession.java index 04fc33409d..51004660a6 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/AbstractServerSockJsSession.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/AbstractServerSockJsSession.java @@ -22,11 +22,12 @@ import java.net.SocketException; import java.util.Date; import java.util.concurrent.ScheduledFuture; -import org.springframework.sockjs.SockJsHandler; -import org.springframework.sockjs.SockJsSession; -import org.springframework.sockjs.SockJsSessionSupport; +import org.springframework.sockjs.AbstractSockJsSession; import org.springframework.util.Assert; import org.springframework.websocket.CloseStatus; +import org.springframework.websocket.TextMessage; +import org.springframework.websocket.WebSocketHandler; +import org.springframework.websocket.WebSocketMessage; /** @@ -36,15 +37,17 @@ import org.springframework.websocket.CloseStatus; * @author Rossen Stoyanchev * @since 4.0 */ -public abstract class AbstractServerSockJsSession extends SockJsSessionSupport { +public abstract class AbstractServerSockJsSession extends AbstractSockJsSession { private final SockJsConfiguration sockJsConfig; private ScheduledFuture heartbeatTask; - public AbstractServerSockJsSession(String sessionId, SockJsConfiguration config, SockJsHandler sockJsHandler) { - super(sessionId, sockJsHandler); + public AbstractServerSockJsSession(String sessionId, SockJsConfiguration config, + WebSocketHandler webSocketHandler) { + + super(sessionId, webSocketHandler); this.sockJsConfig = config; } @@ -52,12 +55,13 @@ public abstract class AbstractServerSockJsSession extends SockJsSessionSupport { return this.sockJsConfig; } - public final synchronized void sendMessage(String message) throws IOException { + public final synchronized void sendMessage(WebSocketMessage message) throws Exception { Assert.isTrue(!isClosed(), "Cannot send a message, session has been closed"); - sendMessageInternal(message); + Assert.isInstanceOf(TextMessage.class, message, "Expected text message: " + message); + sendMessageInternal(((TextMessage) message).getPayload()); } - protected abstract void sendMessageInternal(String message) throws IOException; + protected abstract void sendMessageInternal(String message) throws Exception; @Override @@ -67,7 +71,7 @@ public abstract class AbstractServerSockJsSession extends SockJsSessionSupport { } @Override - public final synchronized void closeInternal(CloseStatus status) throws IOException { + public final synchronized void closeInternal(CloseStatus status) throws Exception { if (isActive()) { // TODO: deliver messages "in flight" before sending close frame try { @@ -84,13 +88,13 @@ public abstract class AbstractServerSockJsSession extends SockJsSessionSupport { } // TODO: close status/reason - protected abstract void disconnect(CloseStatus status) throws IOException; + protected abstract void disconnect(CloseStatus status) throws Exception; /** * For internal use within a TransportHandler and the (TransportHandler-specific) * session sub-class. */ - protected void writeFrame(SockJsFrame frame) throws IOException { + protected void writeFrame(SockJsFrame frame) throws Exception { if (logger.isTraceEnabled()) { logger.trace("Preparing to write " + frame); } @@ -114,9 +118,9 @@ public abstract class AbstractServerSockJsSession extends SockJsSessionSupport { } } - protected abstract void writeFrameInternal(SockJsFrame frame) throws IOException; + protected abstract void writeFrameInternal(SockJsFrame frame) throws Exception; - public synchronized void sendHeartbeat() throws IOException { + public synchronized void sendHeartbeat() throws Exception { if (isActive()) { writeFrame(SockJsFrame.heartbeatFrame()); scheduleHeartbeat(); @@ -135,7 +139,7 @@ public abstract class AbstractServerSockJsSession extends SockJsSessionSupport { try { sendHeartbeat(); } - catch (IOException e) { + catch (Exception e) { // ignore } } diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/AbstractSockJsService.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/AbstractSockJsService.java index 6d84713727..24c3dacb06 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/AbstractSockJsService.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/AbstractSockJsService.java @@ -34,12 +34,12 @@ import org.springframework.http.server.ServerHttpRequest; import org.springframework.http.server.ServerHttpResponse; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; -import org.springframework.sockjs.SockJsHandler; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.DigestUtils; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; +import org.springframework.websocket.WebSocketHandler; /** @@ -218,7 +218,7 @@ public abstract class AbstractSockJsService * @throws Exception */ public final void handleRequest(ServerHttpRequest request, ServerHttpResponse response, - String sockJsPath, SockJsHandler sockJsHandler) throws Exception { + String sockJsPath, WebSocketHandler webSocketHandler) throws Exception { logger.debug(request.getMethod() + " [" + sockJsPath + "]"); @@ -244,7 +244,7 @@ public abstract class AbstractSockJsService return; } else if (sockJsPath.equals("/websocket")) { - handleRawWebSocketRequest(request, response, sockJsHandler); + handleRawWebSocketRequest(request, response, webSocketHandler); return; } @@ -264,7 +264,7 @@ public abstract class AbstractSockJsService return; } - handleTransportRequest(request, response, sessionId, TransportType.fromValue(transport), sockJsHandler); + handleTransportRequest(request, response, sessionId, TransportType.fromValue(transport), webSocketHandler); } finally { response.flush(); @@ -272,10 +272,10 @@ public abstract class AbstractSockJsService } protected abstract void handleRawWebSocketRequest(ServerHttpRequest request, ServerHttpResponse response, - SockJsHandler sockJsHandler) throws Exception; + WebSocketHandler webSocketHandler) throws Exception; protected abstract void handleTransportRequest(ServerHttpRequest request, ServerHttpResponse response, - String sessionId, TransportType transportType, SockJsHandler sockJsHandler) throws Exception; + String sessionId, TransportType transportType, WebSocketHandler webSocketHandler) throws Exception; protected boolean validateRequest(String serverId, String sessionId, String transport) { diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/ConfigurableTransportHandler.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/ConfigurableTransportHandler.java index 8ad26ef6fa..799ba50d89 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/ConfigurableTransportHandler.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/ConfigurableTransportHandler.java @@ -16,11 +16,6 @@ package org.springframework.sockjs.server; -import java.util.Collection; - -import org.springframework.sockjs.SockJsHandler; -import org.springframework.websocket.WebSocketHandler; - /** * @@ -31,10 +26,4 @@ public interface ConfigurableTransportHandler extends TransportHandler { void setSockJsConfiguration(SockJsConfiguration sockJsConfig); - /** - * Pre-register {@link SockJsHandler} instances so they can be adapted to - * {@link WebSocketHandler} and hence re-used at runtime. - */ - void registerSockJsHandlers(Collection sockJsHandlers); - } diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/SockJsService.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/SockJsService.java index 23c79be7d2..604ddcbca1 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/SockJsService.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/SockJsService.java @@ -16,11 +16,8 @@ package org.springframework.sockjs.server; -import java.util.Collection; - import org.springframework.http.server.ServerHttpRequest; import org.springframework.http.server.ServerHttpResponse; -import org.springframework.sockjs.SockJsHandler; import org.springframework.websocket.WebSocketHandler; @@ -31,15 +28,8 @@ import org.springframework.websocket.WebSocketHandler; */ public interface SockJsService { - /** - * Pre-register {@link SockJsHandler} instances so they can be adapted to - * {@link WebSocketHandler} and hence re-used at runtime when - * {@link #handleRequest(ServerHttpRequest, ServerHttpResponse, String, SockJsHandler) handleRequest} - * is called. - */ - void registerSockJsHandlers(Collection sockJsHandlers); void handleRequest(ServerHttpRequest request, ServerHttpResponse response, String sockJsPath, - SockJsHandler handler) throws Exception; + WebSocketHandler webSocketHandler) throws Exception; } diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/TransportHandler.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/TransportHandler.java index ea6208b836..8c662127d6 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/TransportHandler.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/TransportHandler.java @@ -17,8 +17,8 @@ package org.springframework.sockjs.server; import org.springframework.http.server.ServerHttpRequest; import org.springframework.http.server.ServerHttpResponse; -import org.springframework.sockjs.SockJsHandler; -import org.springframework.sockjs.SockJsSessionSupport; +import org.springframework.sockjs.AbstractSockJsSession; +import org.springframework.websocket.WebSocketHandler; /** @@ -31,6 +31,6 @@ public interface TransportHandler { TransportType getTransportType(); void handleRequest(ServerHttpRequest request, ServerHttpResponse response, - SockJsHandler sockJsHandler, SockJsSessionSupport session) throws Exception; + WebSocketHandler webSocketHandler, AbstractSockJsSession session) throws Exception; } diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/support/DefaultSockJsService.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/support/DefaultSockJsService.java index 60f8143f70..aa35115fe6 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/support/DefaultSockJsService.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/support/DefaultSockJsService.java @@ -16,7 +16,6 @@ package org.springframework.sockjs.server.support; import java.util.Arrays; -import java.util.Collection; import java.util.HashMap; import java.util.List; import java.util.Map; @@ -29,9 +28,8 @@ import org.springframework.http.HttpStatus; import org.springframework.http.server.ServerHttpRequest; import org.springframework.http.server.ServerHttpResponse; import org.springframework.scheduling.TaskScheduler; -import org.springframework.sockjs.SockJsHandler; +import org.springframework.sockjs.AbstractSockJsSession; import org.springframework.sockjs.SockJsSessionFactory; -import org.springframework.sockjs.SockJsSessionSupport; import org.springframework.sockjs.server.AbstractSockJsService; import org.springframework.sockjs.server.ConfigurableTransportHandler; import org.springframework.sockjs.server.TransportHandler; @@ -40,7 +38,6 @@ import org.springframework.sockjs.server.transport.EventSourceTransportHandler; import org.springframework.sockjs.server.transport.HtmlFileTransportHandler; import org.springframework.sockjs.server.transport.JsonpPollingTransportHandler; import org.springframework.sockjs.server.transport.JsonpTransportHandler; -import org.springframework.sockjs.server.transport.WebSocketSockJsHandlerAdapter; import org.springframework.sockjs.server.transport.WebSocketTransportHandler; import org.springframework.sockjs.server.transport.XhrPollingTransportHandler; import org.springframework.sockjs.server.transport.XhrStreamingTransportHandler; @@ -65,9 +62,7 @@ public class DefaultSockJsService extends AbstractSockJsService implements Initi private TaskSchedulerHolder sessionTimeoutSchedulerHolder; - private final Map sessions = new ConcurrentHashMap(); - - private final Map sockJsHandlers = new HashMap(); + private final Map sessions = new ConcurrentHashMap(); public DefaultSockJsService() { @@ -93,22 +88,6 @@ public class DefaultSockJsService extends AbstractSockJsService implements Initi } } - public void registerSockJsHandlers(Collection sockJsHandlers) { - for (SockJsHandler sockJsHandler : sockJsHandlers) { - if (!this.sockJsHandlers.containsKey(sockJsHandler)) { - this.sockJsHandlers.put(sockJsHandler, adaptSockJsHandler(sockJsHandler)); - } - } - configureTransportHandlers(); - } - - /** - * Adapt the {@link SockJsHandler} to the {@link WebSocketHandler} contract for - * raw WebSocket communication on SockJS path "/websocket". - */ - protected WebSocketSockJsHandlerAdapter adaptSockJsHandler(SockJsHandler sockJsHandler) { - return new WebSocketSockJsHandlerAdapter(this, sockJsHandler); - } @Override public void afterPropertiesSet() throws Exception { @@ -135,7 +114,11 @@ public class DefaultSockJsService extends AbstractSockJsService implements Initi } } - configureTransportHandlers(); + for (TransportHandler h : this.transportHandlers.values()) { + if (h instanceof ConfigurableTransportHandler) { + ((ConfigurableTransportHandler) h).setSockJsConfiguration(this); + } + } this.sessionTimeoutSchedulerHolder.initialize(); @@ -146,7 +129,7 @@ public class DefaultSockJsService extends AbstractSockJsService implements Initi if (logger.isTraceEnabled() && (count != 0)) { logger.trace("Checking " + count + " session(s) for timeouts [" + getName() + "]"); } - for (SockJsSessionSupport session : sessions.values()) { + for (AbstractSockJsSession session : sessions.values()) { if (session.getTimeSinceLastActive() > getDisconnectDelay()) { if (logger.isTraceEnabled()) { logger.trace("Removing " + session + " for [" + getName() + "]"); @@ -172,32 +155,14 @@ public class DefaultSockJsService extends AbstractSockJsService implements Initi this.sessionTimeoutSchedulerHolder.destroy(); } - private void configureTransportHandlers() { - for (TransportHandler h : this.transportHandlers.values()) { - if (h instanceof ConfigurableTransportHandler) { - ((ConfigurableTransportHandler) h).setSockJsConfiguration(this); - if (!this.sockJsHandlers.isEmpty()) { - ((ConfigurableTransportHandler) h).registerSockJsHandlers(this.sockJsHandlers.keySet()); - if (h instanceof HandshakeHandler) { - ((HandshakeHandler) h).registerWebSocketHandlers(this.sockJsHandlers.values()); - } - } - } - } - } - @Override protected void handleRawWebSocketRequest(ServerHttpRequest request, ServerHttpResponse response, - SockJsHandler sockJsHandler) throws Exception { + WebSocketHandler webSocketHandler) throws Exception { if (isWebSocketEnabled()) { TransportHandler transportHandler = this.transportHandlers.get(TransportType.WEBSOCKET); if (transportHandler != null) { if (transportHandler instanceof HandshakeHandler) { - WebSocketHandler webSocketHandler = this.sockJsHandlers.get(sockJsHandler); - if (webSocketHandler == null) { - webSocketHandler = adaptSockJsHandler(sockJsHandler); - } ((HandshakeHandler) transportHandler).doHandshake(request, response, webSocketHandler); return; } @@ -209,7 +174,7 @@ public class DefaultSockJsService extends AbstractSockJsService implements Initi @Override protected void handleTransportRequest(ServerHttpRequest request, ServerHttpResponse response, - String sessionId, TransportType transportType, SockJsHandler sockJsHandler) throws Exception { + String sessionId, TransportType transportType, WebSocketHandler webSocketHandler) throws Exception { TransportHandler transportHandler = this.transportHandlers.get(transportType); @@ -236,7 +201,7 @@ public class DefaultSockJsService extends AbstractSockJsService implements Initi return; } - SockJsSessionSupport session = getSockJsSession(sessionId, sockJsHandler, transportHandler); + AbstractSockJsSession session = getSockJsSession(sessionId, webSocketHandler, transportHandler); if (session != null) { if (transportType.setsNoCacheHeader()) { @@ -255,13 +220,13 @@ public class DefaultSockJsService extends AbstractSockJsService implements Initi } } - transportHandler.handleRequest(request, response, sockJsHandler, session); + transportHandler.handleRequest(request, response, webSocketHandler, session); } - public SockJsSessionSupport getSockJsSession(String sessionId, SockJsHandler sockJsHandler, + public AbstractSockJsSession getSockJsSession(String sessionId, WebSocketHandler webSocketHandler, TransportHandler transportHandler) { - SockJsSessionSupport session = this.sessions.get(sessionId); + AbstractSockJsSession session = this.sessions.get(sessionId); if (session != null) { return session; } @@ -275,7 +240,7 @@ public class DefaultSockJsService extends AbstractSockJsService implements Initi return session; } logger.debug("Creating new session with session id \"" + sessionId + "\""); - session = (SockJsSessionSupport) sessionFactory.createSession(sessionId, sockJsHandler); + session = (AbstractSockJsSession) sessionFactory.createSession(sessionId, webSocketHandler); this.sessions.put(sessionId, session); return session; } diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/support/SockJsHttpRequestHandler.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/support/SockJsHttpRequestHandler.java index 22d058e0eb..b5e2b49caf 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/support/SockJsHttpRequestHandler.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/support/SockJsHttpRequestHandler.java @@ -17,7 +17,6 @@ package org.springframework.sockjs.server.support; import java.io.IOException; -import java.util.Collections; import javax.servlet.ServletException; import javax.servlet.http.HttpServletRequest; @@ -30,13 +29,13 @@ import org.springframework.http.server.AsyncServletServerHttpRequest; import org.springframework.http.server.ServerHttpRequest; import org.springframework.http.server.ServerHttpResponse; import org.springframework.http.server.ServletServerHttpResponse; -import org.springframework.sockjs.SockJsHandler; import org.springframework.sockjs.server.SockJsService; import org.springframework.util.Assert; import org.springframework.web.HttpRequestHandler; import org.springframework.web.util.NestedServletException; import org.springframework.web.util.UrlPathHelper; import org.springframework.websocket.HandlerProvider; +import org.springframework.websocket.WebSocketHandler; /** @@ -50,7 +49,7 @@ public class SockJsHttpRequestHandler implements HttpRequestHandler, BeanFactory private final SockJsService sockJsService; - private final HandlerProvider handlerProvider; + private final HandlerProvider handlerProvider; private final UrlPathHelper urlPathHelper = new UrlPathHelper(); @@ -62,16 +61,15 @@ public class SockJsHttpRequestHandler implements HttpRequestHandler, BeanFactory * that begins with the specified prefix will be handled by this service. In a * Servlet container this is the path within the current servlet mapping. */ - public SockJsHttpRequestHandler(String prefix, SockJsService sockJsService, SockJsHandler sockJsHandler) { + public SockJsHttpRequestHandler(String prefix, SockJsService sockJsService, WebSocketHandler webSocketHandler) { Assert.hasText(prefix, "prefix is required"); Assert.notNull(sockJsService, "sockJsService is required"); - Assert.notNull(sockJsHandler, "sockJsHandler is required"); + Assert.notNull(webSocketHandler, "webSocketHandler is required"); this.prefix = prefix; this.sockJsService = sockJsService; - this.sockJsService.registerSockJsHandlers(Collections.singleton(sockJsHandler)); - this.handlerProvider = new HandlerProvider(sockJsHandler); + this.handlerProvider = new HandlerProvider(webSocketHandler); } /** @@ -82,15 +80,15 @@ public class SockJsHttpRequestHandler implements HttpRequestHandler, BeanFactory * Servlet container this is the path within the current servlet mapping. */ public SockJsHttpRequestHandler(String prefix, SockJsService sockJsService, - Class sockJsHandlerClass) { + Class webSocketHandlerClass) { Assert.hasText(prefix, "prefix is required"); Assert.notNull(sockJsService, "sockJsService is required"); - Assert.notNull(sockJsHandlerClass, "sockJsHandlerClass is required"); + Assert.notNull(webSocketHandlerClass, "webSocketHandlerClass is required"); this.prefix = prefix; this.sockJsService = sockJsService; - this.handlerProvider = new HandlerProvider(sockJsHandlerClass); + this.handlerProvider = new HandlerProvider(webSocketHandlerClass); } public String getPrefix() { @@ -121,8 +119,8 @@ public class SockJsHttpRequestHandler implements HttpRequestHandler, BeanFactory ServerHttpResponse httpResponse = new ServletServerHttpResponse(response); try { - SockJsHandler sockJsHandler = this.handlerProvider.getHandler(); - this.sockJsService.handleRequest(httpRequest, httpResponse, sockJsPath, sockJsHandler); + WebSocketHandler webSocketHandler = this.handlerProvider.getHandler(); + this.sockJsService.handleRequest(httpRequest, httpResponse, sockJsPath, webSocketHandler); } catch (Exception ex) { // TODO diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractHttpReceivingTransportHandler.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractHttpReceivingTransportHandler.java index 300adbd981..8b7e3604a9 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractHttpReceivingTransportHandler.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractHttpReceivingTransportHandler.java @@ -25,9 +25,9 @@ import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; import org.springframework.http.server.ServerHttpRequest; import org.springframework.http.server.ServerHttpResponse; -import org.springframework.sockjs.SockJsHandler; -import org.springframework.sockjs.SockJsSessionSupport; +import org.springframework.sockjs.AbstractSockJsSession; import org.springframework.sockjs.server.TransportHandler; +import org.springframework.websocket.WebSocketHandler; import com.fasterxml.jackson.databind.JsonMappingException; import com.fasterxml.jackson.databind.ObjectMapper; @@ -53,7 +53,7 @@ public abstract class AbstractHttpReceivingTransportHandler implements Transport @Override public final void handleRequest(ServerHttpRequest request, ServerHttpResponse response, - SockJsHandler sockJsHandler, SockJsSessionSupport session) throws Exception { + WebSocketHandler webSocketHandler, AbstractSockJsSession session) throws Exception { if (session == null) { response.setStatusCode(HttpStatus.NOT_FOUND); @@ -64,7 +64,7 @@ public abstract class AbstractHttpReceivingTransportHandler implements Transport } protected void handleRequestInternal(ServerHttpRequest request, ServerHttpResponse response, - SockJsSessionSupport session) throws Exception { + AbstractSockJsSession session) throws Exception { String[] messages = null; try { diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractHttpSendingTransportHandler.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractHttpSendingTransportHandler.java index 726fa76c87..978663598c 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractHttpSendingTransportHandler.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractHttpSendingTransportHandler.java @@ -16,20 +16,19 @@ package org.springframework.sockjs.server.transport; import java.io.IOException; -import java.util.Collection; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.http.MediaType; import org.springframework.http.server.ServerHttpRequest; import org.springframework.http.server.ServerHttpResponse; -import org.springframework.sockjs.SockJsHandler; +import org.springframework.sockjs.AbstractSockJsSession; import org.springframework.sockjs.SockJsSessionFactory; -import org.springframework.sockjs.SockJsSessionSupport; import org.springframework.sockjs.server.ConfigurableTransportHandler; import org.springframework.sockjs.server.SockJsConfiguration; import org.springframework.sockjs.server.SockJsFrame; import org.springframework.sockjs.server.SockJsFrame.FrameFormat; +import org.springframework.websocket.WebSocketHandler; /** * TODO @@ -38,7 +37,7 @@ import org.springframework.sockjs.server.SockJsFrame.FrameFormat; * @since 4.0 */ public abstract class AbstractHttpSendingTransportHandler - implements ConfigurableTransportHandler, SockJsSessionFactory { + implements ConfigurableTransportHandler, SockJsSessionFactory { protected final Log logger = LogFactory.getLog(this.getClass()); @@ -50,18 +49,13 @@ public abstract class AbstractHttpSendingTransportHandler this.sockJsConfig = sockJsConfig; } - @Override - public void registerSockJsHandlers(Collection sockJsHandlers) { - // ignore - } - public SockJsConfiguration getSockJsConfig() { return this.sockJsConfig; } @Override public final void handleRequest(ServerHttpRequest request, ServerHttpResponse response, - SockJsHandler sockJsHandler, SockJsSessionSupport session) throws Exception { + WebSocketHandler webSocketHandler, AbstractSockJsSession session) throws Exception { // Set content type before writing response.getHeaders().setContentType(getContentType()); diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractHttpServerSockJsSession.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractHttpServerSockJsSession.java index 0c9d624f56..e87006958b 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractHttpServerSockJsSession.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractHttpServerSockJsSession.java @@ -22,7 +22,6 @@ import java.util.concurrent.BlockingQueue; import org.springframework.http.server.AsyncServerHttpRequest; import org.springframework.http.server.ServerHttpRequest; import org.springframework.http.server.ServerHttpResponse; -import org.springframework.sockjs.SockJsHandler; import org.springframework.sockjs.server.AbstractServerSockJsSession; import org.springframework.sockjs.server.SockJsConfiguration; import org.springframework.sockjs.server.SockJsFrame; @@ -30,6 +29,7 @@ import org.springframework.sockjs.server.SockJsFrame.FrameFormat; import org.springframework.sockjs.server.TransportHandler; import org.springframework.util.Assert; import org.springframework.websocket.CloseStatus; +import org.springframework.websocket.WebSocketHandler; /** * An abstract base class for use with HTTP-based transports. @@ -48,8 +48,10 @@ public abstract class AbstractHttpServerSockJsSession extends AbstractServerSock private ServerHttpResponse response; - public AbstractHttpServerSockJsSession(String sessionId, SockJsConfiguration sockJsConfig, SockJsHandler sockJsHandler) { - super(sessionId, sockJsConfig, sockJsHandler); + public AbstractHttpServerSockJsSession(String sessionId, SockJsConfiguration sockJsConfig, + WebSocketHandler webSocketHandler) { + + super(sessionId, sockJsConfig, webSocketHandler); } public void setFrameFormat(FrameFormat frameFormat) { @@ -57,7 +59,7 @@ public abstract class AbstractHttpServerSockJsSession extends AbstractServerSock } public synchronized void setCurrentRequest(ServerHttpRequest request, ServerHttpResponse response, - FrameFormat frameFormat) throws IOException { + FrameFormat frameFormat) throws Exception { if (isClosed()) { logger.debug("connection already closed"); @@ -90,14 +92,14 @@ public abstract class AbstractHttpServerSockJsSession extends AbstractServerSock return this.response; } - protected final synchronized void sendMessageInternal(String message) throws IOException { + protected final synchronized void sendMessageInternal(String message) throws Exception { // assert close() was not called // threads: TH-Session-Endpoint or any other thread this.messageCache.add(message); tryFlushCache(); } - private void tryFlushCache() throws IOException { + private void tryFlushCache() throws Exception { if (isActive() && !getMessageCache().isEmpty()) { logger.trace("Flushing messages"); flushCache(); @@ -107,7 +109,7 @@ public abstract class AbstractHttpServerSockJsSession extends AbstractServerSock /** * Only called if the connection is currently active */ - protected abstract void flushCache() throws IOException; + protected abstract void flushCache() throws Exception; @Override protected void disconnect(CloseStatus status) { diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractStreamingTransportHandler.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractStreamingTransportHandler.java index f8dc637c16..0adfd867b1 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractStreamingTransportHandler.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractStreamingTransportHandler.java @@ -19,8 +19,8 @@ import java.io.IOException; import org.springframework.http.server.ServerHttpRequest; import org.springframework.http.server.ServerHttpResponse; -import org.springframework.sockjs.SockJsHandler; import org.springframework.util.Assert; +import org.springframework.websocket.WebSocketHandler; /** @@ -33,9 +33,9 @@ public abstract class AbstractStreamingTransportHandler extends AbstractHttpSend @Override - public StreamingServerSockJsSession createSession(String sessionId, SockJsHandler sockJsHandler) { + public StreamingServerSockJsSession createSession(String sessionId, WebSocketHandler webSocketHandler) { Assert.notNull(getSockJsConfig(), "This transport requires SockJsConfiguration"); - return new StreamingServerSockJsSession(sessionId, getSockJsConfig(), sockJsHandler); + return new StreamingServerSockJsSession(sessionId, getSockJsConfig(), webSocketHandler); } @Override diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/JsonpPollingTransportHandler.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/JsonpPollingTransportHandler.java index 23c9f07efb..fabd525393 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/JsonpPollingTransportHandler.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/JsonpPollingTransportHandler.java @@ -21,13 +21,13 @@ import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; import org.springframework.http.server.ServerHttpRequest; import org.springframework.http.server.ServerHttpResponse; -import org.springframework.sockjs.SockJsHandler; import org.springframework.sockjs.server.SockJsFrame; import org.springframework.sockjs.server.SockJsFrame.FrameFormat; import org.springframework.sockjs.server.TransportType; import org.springframework.util.Assert; import org.springframework.util.StringUtils; import org.springframework.web.util.JavaScriptUtils; +import org.springframework.websocket.WebSocketHandler; /** @@ -50,9 +50,9 @@ public class JsonpPollingTransportHandler extends AbstractHttpSendingTransportHa } @Override - public PollingServerSockJsSession createSession(String sessionId, SockJsHandler sockJsHandler) { + public PollingServerSockJsSession createSession(String sessionId, WebSocketHandler webSocketHandler) { Assert.notNull(getSockJsConfig(), "This transport requires SockJsConfiguration"); - return new PollingServerSockJsSession(sessionId, getSockJsConfig(), sockJsHandler); + return new PollingServerSockJsSession(sessionId, getSockJsConfig(), webSocketHandler); } @Override diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/JsonpTransportHandler.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/JsonpTransportHandler.java index 95e90dccc0..8d39e2d125 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/JsonpTransportHandler.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/JsonpTransportHandler.java @@ -21,7 +21,7 @@ import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; import org.springframework.http.server.ServerHttpRequest; import org.springframework.http.server.ServerHttpResponse; -import org.springframework.sockjs.SockJsSessionSupport; +import org.springframework.sockjs.AbstractSockJsSession; import org.springframework.sockjs.server.TransportType; public class JsonpTransportHandler extends AbstractHttpReceivingTransportHandler { @@ -34,7 +34,7 @@ public class JsonpTransportHandler extends AbstractHttpReceivingTransportHandler @Override public void handleRequestInternal(ServerHttpRequest request, ServerHttpResponse response, - SockJsSessionSupport sockJsSession) throws Exception { + AbstractSockJsSession sockJsSession) throws Exception { if (MediaType.APPLICATION_FORM_URLENCODED.equals(request.getHeaders().getContentType())) { if (request.getQueryParams().getFirst("d") == null) { diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/PollingServerSockJsSession.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/PollingServerSockJsSession.java index baac23d587..c9814f09f2 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/PollingServerSockJsSession.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/PollingServerSockJsSession.java @@ -15,21 +15,21 @@ */ package org.springframework.sockjs.server.transport; -import java.io.IOException; - -import org.springframework.sockjs.SockJsHandler; import org.springframework.sockjs.server.SockJsConfiguration; import org.springframework.sockjs.server.SockJsFrame; +import org.springframework.websocket.WebSocketHandler; public class PollingServerSockJsSession extends AbstractHttpServerSockJsSession { - public PollingServerSockJsSession(String sessionId, SockJsConfiguration sockJsConfig, SockJsHandler sockJsHandler) { - super(sessionId, sockJsConfig, sockJsHandler); + public PollingServerSockJsSession(String sessionId, SockJsConfiguration sockJsConfig, + WebSocketHandler webSocketHandler) { + + super(sessionId, sockJsConfig, webSocketHandler); } @Override - protected void flushCache() throws IOException { + protected void flushCache() throws Exception { cancelHeartbeat(); String[] messages = getMessageCache().toArray(new String[getMessageCache().size()]); getMessageCache().clear(); @@ -37,7 +37,7 @@ public class PollingServerSockJsSession extends AbstractHttpServerSockJsSession } @Override - protected void writeFrame(SockJsFrame frame) throws IOException { + protected void writeFrame(SockJsFrame frame) throws Exception { super.writeFrame(frame); resetRequest(); } diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/SockJsWebSocketHandler.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/SockJsWebSocketHandler.java index e616455a2c..4e4dc0987e 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/SockJsWebSocketHandler.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/SockJsWebSocketHandler.java @@ -22,14 +22,15 @@ import java.util.concurrent.ConcurrentHashMap; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.sockjs.SockJsHandler; -import org.springframework.sockjs.SockJsSessionSupport; +import org.springframework.sockjs.AbstractSockJsSession; import org.springframework.sockjs.server.AbstractServerSockJsSession; import org.springframework.sockjs.server.SockJsConfiguration; import org.springframework.sockjs.server.SockJsFrame; import org.springframework.util.Assert; import org.springframework.util.StringUtils; import org.springframework.websocket.CloseStatus; +import org.springframework.websocket.TextMessage; +import org.springframework.websocket.TextMessageHandler; import org.springframework.websocket.WebSocketHandler; import org.springframework.websocket.WebSocketSession; @@ -43,55 +44,52 @@ import com.fasterxml.jackson.databind.ObjectMapper; * @author Rossen Stoyanchev * @since 4.0 */ -public class SockJsWebSocketHandler implements WebSocketHandler { +public class SockJsWebSocketHandler implements TextMessageHandler { private static final Log logger = LogFactory.getLog(SockJsWebSocketHandler.class); private final SockJsConfiguration sockJsConfig; - private final SockJsHandler sockJsHandler; + private final WebSocketHandler webSocketHandler; - private final Map sessions = - new ConcurrentHashMap(); + private final Map sessions = + new ConcurrentHashMap(); // TODO: JSON library used must be configurable private final ObjectMapper objectMapper = new ObjectMapper(); - public SockJsWebSocketHandler(SockJsConfiguration sockJsConfig, SockJsHandler sockJsHandler) { + public SockJsWebSocketHandler(SockJsConfiguration sockJsConfig, WebSocketHandler webSocketHandler) { Assert.notNull(sockJsConfig, "sockJsConfig is required"); - Assert.notNull(sockJsHandler, "sockJsHandler is required"); + Assert.notNull(webSocketHandler, "webSocketHandler is required"); this.sockJsConfig = sockJsConfig; - this.sockJsHandler = sockJsHandler; + this.webSocketHandler = webSocketHandler; } protected SockJsConfiguration getSockJsConfig() { return this.sockJsConfig; } - protected SockJsHandler getSockJsHandler() { - return this.sockJsHandler; - } - - protected SockJsSessionSupport getSockJsSession(WebSocketSession wsSession) { + protected AbstractSockJsSession getSockJsSession(WebSocketSession wsSession) { return this.sessions.get(wsSession); } @Override public void afterConnectionEstablished(WebSocketSession wsSession) throws Exception { - SockJsSessionSupport session = new WebSocketServerSockJsSession(wsSession, getSockJsConfig()); + AbstractSockJsSession session = new WebSocketServerSockJsSession(wsSession, getSockJsConfig()); this.sessions.put(wsSession, session); } @Override - public void handleTextMessage(String message, WebSocketSession wsSession) throws Exception { - if (StringUtils.isEmpty(message)) { + public void handleTextMessage(TextMessage message, WebSocketSession wsSession) throws Exception { + String payload = message.getPayload(); + if (StringUtils.isEmpty(payload)) { logger.trace("Ignoring empty message"); return; } try { - String[] messages = this.objectMapper.readValue(message, String[].class); - SockJsSessionSupport session = getSockJsSession(wsSession); + String[] messages = this.objectMapper.readValue(payload, String[].class); + AbstractSockJsSession session = getSockJsSession(wsSession); session.delegateMessages(messages); } catch (IOException e) { @@ -100,21 +98,15 @@ public class SockJsWebSocketHandler implements WebSocketHandler { } } - @Override - public void handleBinaryMessage(byte[] message, WebSocketSession session) throws Exception { - logger.warn("Unexpected binary message for " + session); - session.close(CloseStatus.NOT_ACCEPTABLE); - } - @Override public void afterConnectionClosed(CloseStatus status, WebSocketSession wsSession) throws Exception { - SockJsSessionSupport session = this.sessions.remove(wsSession); + AbstractSockJsSession session = this.sessions.remove(wsSession); session.delegateConnectionClosed(status); } @Override public void handleError(Throwable exception, WebSocketSession webSocketSession) { - SockJsSessionSupport session = getSockJsSession(webSocketSession); + AbstractSockJsSession session = getSockJsSession(webSocketSession); session.delegateError(exception); } @@ -135,9 +127,10 @@ public class SockJsWebSocketHandler implements WebSocketHandler { public WebSocketServerSockJsSession(WebSocketSession wsSession, SockJsConfiguration sockJsConfig) throws Exception { - super(getSockJsSessionId(wsSession), sockJsConfig, getSockJsHandler()); + super(getSockJsSessionId(wsSession), sockJsConfig, SockJsWebSocketHandler.this.webSocketHandler); this.wsSession = wsSession; - this.wsSession.sendTextMessage(SockJsFrame.openFrame().getContent()); + TextMessage message = new TextMessage(SockJsFrame.openFrame().getContent()); + this.wsSession.sendMessage(message); scheduleHeartbeat(); delegateConnectionEstablished(); } @@ -148,22 +141,23 @@ public class SockJsWebSocketHandler implements WebSocketHandler { } @Override - public void sendMessageInternal(String message) throws IOException { + public void sendMessageInternal(String message) throws Exception { cancelHeartbeat(); writeFrame(SockJsFrame.messageFrame(message)); scheduleHeartbeat(); } @Override - protected void writeFrameInternal(SockJsFrame frame) throws IOException { + protected void writeFrameInternal(SockJsFrame frame) throws Exception { if (logger.isTraceEnabled()) { logger.trace("Write " + frame); } - this.wsSession.sendTextMessage(frame.getContent()); + TextMessage message = new TextMessage(frame.getContent()); + this.wsSession.sendMessage(message); } @Override - protected void disconnect(CloseStatus status) throws IOException { + protected void disconnect(CloseStatus status) throws Exception { this.wsSession.close(status); } } diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/StreamingServerSockJsSession.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/StreamingServerSockJsSession.java index e0293a7372..77cacca780 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/StreamingServerSockJsSession.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/StreamingServerSockJsSession.java @@ -18,9 +18,9 @@ package org.springframework.sockjs.server.transport; import java.io.IOException; import org.springframework.http.server.ServerHttpResponse; -import org.springframework.sockjs.SockJsHandler; import org.springframework.sockjs.server.SockJsConfiguration; import org.springframework.sockjs.server.SockJsFrame; +import org.springframework.websocket.WebSocketHandler; public class StreamingServerSockJsSession extends AbstractHttpServerSockJsSession { @@ -28,11 +28,13 @@ public class StreamingServerSockJsSession extends AbstractHttpServerSockJsSessio private int byteCount; - public StreamingServerSockJsSession(String sessionId, SockJsConfiguration sockJsConfig, SockJsHandler sockJsHandler) { - super(sessionId, sockJsConfig, sockJsHandler); + public StreamingServerSockJsSession(String sessionId, SockJsConfiguration sockJsConfig, + WebSocketHandler webSocketHandler) { + + super(sessionId, sockJsConfig, webSocketHandler); } - protected void flushCache() throws IOException { + protected void flushCache() throws Exception { cancelHeartbeat(); diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/WebSocketSockJsHandlerAdapter.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/WebSocketSockJsHandlerAdapter.java deleted file mode 100644 index 12a5bcd282..0000000000 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/WebSocketSockJsHandlerAdapter.java +++ /dev/null @@ -1,132 +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.sockjs.server.transport; - -import java.io.IOException; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.springframework.sockjs.SockJsHandler; -import org.springframework.sockjs.SockJsSessionSupport; -import org.springframework.sockjs.server.SockJsConfiguration; -import org.springframework.util.Assert; -import org.springframework.websocket.CloseStatus; -import org.springframework.websocket.WebSocketHandler; -import org.springframework.websocket.WebSocketSession; - - -/** - * A plain {@link WebSocketHandler} to {@link SockJsHandler} adapter that merely delegates - * without any additional SockJS message framing. Used for raw WebSocket communication at - * SockJS path "/websocket". - * - * @author Rossen Stoyanchev - * @since 4.0 - */ -public class WebSocketSockJsHandlerAdapter implements WebSocketHandler { - - private static final Log logger = LogFactory.getLog(WebSocketSockJsHandlerAdapter.class); - - private final SockJsConfiguration sockJsConfig; - - private final SockJsHandler sockJsHandler; - - private final Map sessions = - new ConcurrentHashMap(); - - - public WebSocketSockJsHandlerAdapter(SockJsConfiguration sockJsConfig, SockJsHandler sockJsHandler) { - Assert.notNull(sockJsConfig, "sockJsConfig is required"); - Assert.notNull(sockJsHandler, "sockJsHandler is required"); - this.sockJsConfig = sockJsConfig; - this.sockJsHandler = sockJsHandler; - } - - protected SockJsConfiguration getSockJsConfig() { - return this.sockJsConfig; - } - - protected SockJsHandler getSockJsHandler() { - return this.sockJsHandler; - } - - protected SockJsSessionSupport getSockJsSession(WebSocketSession wsSession) { - return this.sessions.get(wsSession); - } - - @Override - public void afterConnectionEstablished(WebSocketSession wsSession) throws Exception { - SockJsSessionSupport session = new SockJsWebSocketSessionAdapter(wsSession); - this.sessions.put(wsSession, session); - } - - @Override - public void handleTextMessage(String message, WebSocketSession wsSession) throws Exception { - SockJsSessionSupport session = getSockJsSession(wsSession); - session.delegateMessages(new String[] { message }); - } - - @Override - public void handleBinaryMessage(byte[] message, WebSocketSession session) throws Exception { - logger.warn("Unexpected binary message for " + session); - session.close(CloseStatus.NOT_ACCEPTABLE); - } - - @Override - public void afterConnectionClosed(CloseStatus status, WebSocketSession wsSession) throws Exception { - SockJsSessionSupport session = this.sessions.remove(wsSession); - session.delegateConnectionClosed(status); - } - - @Override - public void handleError(Throwable exception, WebSocketSession wsSession) { - logger.error("Error for " + wsSession); - SockJsSessionSupport session = getSockJsSession(wsSession); - session.delegateError(exception); - } - - - private class SockJsWebSocketSessionAdapter extends SockJsSessionSupport { - - private final WebSocketSession wsSession; - - - public SockJsWebSocketSessionAdapter(WebSocketSession wsSession) throws Exception { - super(wsSession.getId(), getSockJsHandler()); - this.wsSession = wsSession; - delegateConnectionEstablished(); - } - - @Override - public boolean isActive() { - return this.wsSession.isOpen(); - } - - @Override - public void sendMessage(String message) throws IOException { - this.wsSession.sendTextMessage(message); - } - - @Override - public void closeInternal(CloseStatus status) throws IOException { - this.wsSession.close(status); - } - } - -} diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/WebSocketTransportHandler.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/WebSocketTransportHandler.java index 32702d00d9..836e7331f5 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/WebSocketTransportHandler.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/WebSocketTransportHandler.java @@ -16,17 +16,9 @@ package org.springframework.sockjs.server.transport; -import java.util.ArrayList; -import java.util.Collection; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; - import org.springframework.http.server.ServerHttpRequest; import org.springframework.http.server.ServerHttpResponse; -import org.springframework.sockjs.SockJsHandler; -import org.springframework.sockjs.SockJsSessionSupport; +import org.springframework.sockjs.AbstractSockJsSession; import org.springframework.sockjs.server.ConfigurableTransportHandler; import org.springframework.sockjs.server.SockJsConfiguration; import org.springframework.sockjs.server.TransportHandler; @@ -50,10 +42,6 @@ public class WebSocketTransportHandler implements ConfigurableTransportHandler, private SockJsConfiguration sockJsConfig; - private final Map sockJsHandlerCache = new HashMap(); - - private final Collection rawWebSocketHandlerCache = new ArrayList(); - public WebSocketTransportHandler(HandshakeHandler handshakeHandler) { Assert.notNull(handshakeHandler, "handshakeHandler is required"); @@ -71,50 +59,22 @@ public class WebSocketTransportHandler implements ConfigurableTransportHandler, } @Override - public void registerSockJsHandlers(Collection sockJsHandlers) { - this.sockJsHandlerCache.clear(); - for (SockJsHandler sockJsHandler : sockJsHandlers) { - this.sockJsHandlerCache.put(sockJsHandler, adaptSockJsHandler(sockJsHandler)); - } - this.handshakeHandler.registerWebSocketHandlers(getAllWebSocketHandlers()); + public void handleRequest(ServerHttpRequest request, ServerHttpResponse response, + WebSocketHandler webSocketHandler, AbstractSockJsSession session) throws Exception { + + this.handshakeHandler.doHandshake(request, response, adaptSockJsHandler(webSocketHandler)); } /** * Adapt the {@link SockJsHandler} to the {@link WebSocketHandler} contract for * exchanging SockJS message over WebSocket. */ - protected WebSocketHandler adaptSockJsHandler(SockJsHandler sockJsHandler) { - return new SockJsWebSocketHandler(this.sockJsConfig, sockJsHandler); - } - - private Collection getAllWebSocketHandlers() { - Set handlers = new HashSet(); - handlers.addAll(this.sockJsHandlerCache.values()); - handlers.addAll(this.rawWebSocketHandlerCache); - return handlers; - } - - @Override - public void handleRequest(ServerHttpRequest request, ServerHttpResponse response, - SockJsHandler sockJsHandler, SockJsSessionSupport session) throws Exception { - - WebSocketHandler webSocketHandler = this.sockJsHandlerCache.get(sockJsHandler); - if (webSocketHandler == null) { - webSocketHandler = adaptSockJsHandler(sockJsHandler); - } - - this.handshakeHandler.doHandshake(request, response, webSocketHandler); + protected WebSocketHandler adaptSockJsHandler(WebSocketHandler handler) { + return new SockJsWebSocketHandler(this.sockJsConfig, handler); } // HandshakeHandler methods - @Override - public void registerWebSocketHandlers(Collection webSocketHandlers) { - this.rawWebSocketHandlerCache.clear(); - this.rawWebSocketHandlerCache.addAll(webSocketHandlers); - this.handshakeHandler.registerWebSocketHandlers(getAllWebSocketHandlers()); - } - @Override public boolean doHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler webSocketHandler) throws Exception { diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/XhrPollingTransportHandler.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/XhrPollingTransportHandler.java index e424bac91a..57b5fd4e94 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/XhrPollingTransportHandler.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/transport/XhrPollingTransportHandler.java @@ -19,11 +19,11 @@ import java.nio.charset.Charset; import org.springframework.http.MediaType; import org.springframework.http.server.ServerHttpRequest; -import org.springframework.sockjs.SockJsHandler; import org.springframework.sockjs.server.SockJsFrame.DefaultFrameFormat; import org.springframework.sockjs.server.SockJsFrame.FrameFormat; import org.springframework.sockjs.server.TransportType; import org.springframework.util.Assert; +import org.springframework.websocket.WebSocketHandler; /** @@ -50,9 +50,9 @@ public class XhrPollingTransportHandler extends AbstractHttpSendingTransportHand return new DefaultFrameFormat("%s\n"); } - public PollingServerSockJsSession createSession(String sessionId, SockJsHandler sockJsHandler) { + public PollingServerSockJsSession createSession(String sessionId, WebSocketHandler webSocketHandler) { Assert.notNull(getSockJsConfig(), "This transport requires SockJsConfiguration"); - return new PollingServerSockJsSession(sessionId, getSockJsConfig(), sockJsHandler); + return new PollingServerSockJsSession(sessionId, getSockJsConfig(), webSocketHandler); } } diff --git a/spring-websocket/src/main/java/org/springframework/websocket/BinaryMessage.java b/spring-websocket/src/main/java/org/springframework/websocket/BinaryMessage.java new file mode 100644 index 0000000000..5f136bc31e --- /dev/null +++ b/spring-websocket/src/main/java/org/springframework/websocket/BinaryMessage.java @@ -0,0 +1,85 @@ +/* + * 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; + +import java.io.ByteArrayInputStream; +import java.io.InputStream; +import java.nio.ByteBuffer; + + +/** + * Represents a binary WebSocket message. + * + * @author Rossen Stoyanchev + * @since 4.0 + */ +public final class BinaryMessage extends WebSocketMessage { + + private final byte[] bytes; + + private final boolean last; + + + public BinaryMessage(ByteBuffer payload) { + this(payload, true); + } + + public BinaryMessage(ByteBuffer payload, boolean isLast) { + super(payload); + this.bytes = null; + this.last = isLast; + } + + public BinaryMessage(byte[] payload) { + this(payload, true); + } + + public BinaryMessage(byte[] payload, boolean isLast) { + super((payload != null) ? ByteBuffer.wrap(payload) : null); + this.bytes = payload; + this.last = isLast; + } + + public boolean isLast() { + return this.last; + } + + public byte[] getByteArray() { + if (this.bytes != null) { + return this.bytes; + } + else if (getPayload() != null){ + byte[] result = new byte[getPayload().remaining()]; + getPayload().get(result); + return result; + } + else { + return null; + } + } + + public InputStream getInputStream() { + byte[] array = getByteArray(); + return (array != null) ? new ByteArrayInputStream(array) : null; + } + + @Override + public String toString() { + int size = (getPayload() != null) ? getPayload().remaining() : 0; + return "WebSocket binary message size=" + size; + } + +} diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/SockJsHandlerAdapter.java b/spring-websocket/src/main/java/org/springframework/websocket/BinaryMessageHandler.java similarity index 52% rename from spring-websocket/src/main/java/org/springframework/sockjs/SockJsHandlerAdapter.java rename to spring-websocket/src/main/java/org/springframework/websocket/BinaryMessageHandler.java index 3523f6c0dc..50940e34da 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/SockJsHandlerAdapter.java +++ b/spring-websocket/src/main/java/org/springframework/websocket/BinaryMessageHandler.java @@ -5,7 +5,7 @@ * 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 + * 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, @@ -13,33 +13,23 @@ * See the License for the specific language governing permissions and * limitations under the License. */ +package org.springframework.websocket; -package org.springframework.sockjs; - -import org.springframework.websocket.CloseStatus; /** + * A handler for WebSocket binary messages. * * @author Rossen Stoyanchev * @since 4.0 */ -public class SockJsHandlerAdapter implements SockJsHandler { +public interface BinaryMessageHandler extends WebSocketHandler { - @Override - public void afterConnectionEstablished(SockJsSession session) throws Exception { - } - @Override - public void handleMessage(String message, SockJsSession session) throws Exception { - } - - @Override - public void handleError(Throwable exception, SockJsSession session) { - } - - @Override - public void afterConnectionClosed(CloseStatus status, SockJsSession session) { - } + /** + * Handle an incoming binary message. + */ + void handleBinaryMessage(BinaryMessage message, WebSocketSession session) + throws Exception; } diff --git a/spring-websocket/src/main/java/org/springframework/websocket/PartialMessageHandler.java b/spring-websocket/src/main/java/org/springframework/websocket/PartialMessageHandler.java new file mode 100644 index 0000000000..7752baf6cc --- /dev/null +++ b/spring-websocket/src/main/java/org/springframework/websocket/PartialMessageHandler.java @@ -0,0 +1,28 @@ +/* + * 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; + + +/** + * A "marker" interface for {@link BinaryMessageHandler} types that wish to handle partial + * messages. + * + * @author Rossen Stoyanchev + * @since 4.0 + */ +public interface PartialMessageHandler { + +} diff --git a/spring-websocket/src/main/java/org/springframework/websocket/WebSocketHandshakeRequest.java b/spring-websocket/src/main/java/org/springframework/websocket/TextMessage.java similarity index 63% rename from spring-websocket/src/main/java/org/springframework/websocket/WebSocketHandshakeRequest.java rename to spring-websocket/src/main/java/org/springframework/websocket/TextMessage.java index 762cc2fe8b..b254ae7f80 100644 --- a/spring-websocket/src/main/java/org/springframework/websocket/WebSocketHandshakeRequest.java +++ b/spring-websocket/src/main/java/org/springframework/websocket/TextMessage.java @@ -15,34 +15,25 @@ */ package org.springframework.websocket; -import java.net.URI; - -import org.springframework.http.HttpHeaders; +import java.io.Reader; +import java.io.StringReader; /** + * Represents a text WebSocket message. * * @author Rossen Stoyanchev * @since 4.0 */ -public class WebSocketHandshakeRequest { - - private final URI uri; - - private final HttpHeaders headers; +public final class TextMessage extends WebSocketMessage { - public WebSocketHandshakeRequest(HttpHeaders headers, URI uri) { - this.headers = HttpHeaders.readOnlyHttpHeaders(headers); - this.uri = uri; + public TextMessage(String payload) { + super(payload); } - public URI getUri() { - return this.uri; - } - - public HttpHeaders getHeaders() { - return this.headers; + public Reader getReader() { + return new StringReader(getPayload()); } } diff --git a/spring-websocket/src/main/java/org/springframework/websocket/TextMessageHandler.java b/spring-websocket/src/main/java/org/springframework/websocket/TextMessageHandler.java new file mode 100644 index 0000000000..ea073eaf3d --- /dev/null +++ b/spring-websocket/src/main/java/org/springframework/websocket/TextMessageHandler.java @@ -0,0 +1,40 @@ +/* + * 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; + +import org.springframework.websocket.WebSocketHandlerAdapter.TextAndBinaryMessageHandlerAdapter; +import org.springframework.websocket.WebSocketHandlerAdapter.TextMessageHandlerAdapter; + + +/** + * A handler for WebSocket text messages. + * + * @author Rossen Stoyanchev + * @since 4.0 + * + * @see TextMessageHandlerAdapter + * @see TextAndBinaryMessageHandlerAdapter + */ +public interface TextMessageHandler extends WebSocketHandler { + + + /** + * Handle an incoming text message. + */ + void handleTextMessage(TextMessage message, WebSocketSession session) + throws Exception; + +} diff --git a/spring-websocket/src/main/java/org/springframework/websocket/WebSocketHandler.java b/spring-websocket/src/main/java/org/springframework/websocket/WebSocketHandler.java index 783e9b66d1..b420e83940 100644 --- a/spring-websocket/src/main/java/org/springframework/websocket/WebSocketHandler.java +++ b/spring-websocket/src/main/java/org/springframework/websocket/WebSocketHandler.java @@ -19,7 +19,7 @@ package org.springframework.websocket; /** - * A handler for WebSocket messages. + * A handler for WebSocket sessions. * * @author Rossen Stoyanchev * @since 4.0 @@ -32,23 +32,13 @@ public interface WebSocketHandler { void afterConnectionEstablished(WebSocketSession session) throws Exception; /** - * Handle an incoming text message. + * A WebSocket connection has been closed. */ - void handleTextMessage(String message, WebSocketSession session) throws Exception; - - /** - * Handle an incoming binary message. - */ - void handleBinaryMessage(byte[] bytes, WebSocketSession session) throws Exception; + void afterConnectionClosed(CloseStatus closeStatus, WebSocketSession session) throws Exception; /** * TODO */ void handleError(Throwable exception, WebSocketSession session); - /** - * A WebSocket connection has been closed. - */ - void afterConnectionClosed(CloseStatus closeStatus, WebSocketSession session) throws Exception; - } diff --git a/spring-websocket/src/main/java/org/springframework/websocket/WebSocketHandlerAdapter.java b/spring-websocket/src/main/java/org/springframework/websocket/WebSocketHandlerAdapter.java index af03f63b1c..4914f3767a 100644 --- a/spring-websocket/src/main/java/org/springframework/websocket/WebSocketHandlerAdapter.java +++ b/spring-websocket/src/main/java/org/springframework/websocket/WebSocketHandlerAdapter.java @@ -18,6 +18,7 @@ package org.springframework.websocket; /** + * A {@link WebSocketHandler} with empty methods. * * @author Rossen Stoyanchev * @since 4.0 @@ -29,19 +30,24 @@ public class WebSocketHandlerAdapter implements WebSocketHandler { } @Override - public void handleTextMessage(String message, WebSocketSession session) throws Exception { - } - - @Override - public void handleBinaryMessage(byte[] message, WebSocketSession session) throws Exception { + public void afterConnectionClosed(CloseStatus status, WebSocketSession session) throws Exception { } @Override public void handleError(Throwable exception, WebSocketSession session) { } - @Override - public void afterConnectionClosed(CloseStatus status, WebSocketSession session) throws Exception { + + public static abstract class TextMessageHandlerAdapter + extends WebSocketHandlerAdapter implements TextMessageHandler { + } + + public static abstract class BinaryMessageHandlerAdapter + extends WebSocketHandlerAdapter implements BinaryMessageHandler { + } + + public static abstract class TextAndBinaryMessageHandlerAdapter extends WebSocketHandlerAdapter + implements TextMessageHandler, BinaryMessageHandler { } } diff --git a/spring-websocket/src/main/java/org/springframework/websocket/WebSocketMessage.java b/spring-websocket/src/main/java/org/springframework/websocket/WebSocketMessage.java new file mode 100644 index 0000000000..19b79eda74 --- /dev/null +++ b/spring-websocket/src/main/java/org/springframework/websocket/WebSocketMessage.java @@ -0,0 +1,61 @@ +/* + * 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; + +import org.springframework.util.ObjectUtils; + + +/** + * + * @author Rossen Stoyanchev + * @since 4.0 + */ +public abstract class WebSocketMessage { + + private final T payload; + + + WebSocketMessage(T payload) { + this.payload = payload; + } + + public T getPayload() { + return this.payload; + } + + @Override + public String toString() { + return getClass().getSimpleName() + " [payload=" + this.payload + "]"; + } + + @Override + public int hashCode() { + return WebSocketMessage.class.hashCode() * 13 + ObjectUtils.nullSafeHashCode(this.payload); + } + + @Override + public boolean equals(Object other) { + if (this == other) { + return true; + } + if (!(other instanceof WebSocketMessage)) { + return false; + } + WebSocketMessage otherMessage = (WebSocketMessage) other; + return ObjectUtils.nullSafeEquals(this.payload, otherMessage.payload); + } + +} diff --git a/spring-websocket/src/main/java/org/springframework/websocket/WebSocketSession.java b/spring-websocket/src/main/java/org/springframework/websocket/WebSocketSession.java index d1d3c27a35..2c9d4355f5 100644 --- a/spring-websocket/src/main/java/org/springframework/websocket/WebSocketSession.java +++ b/spring-websocket/src/main/java/org/springframework/websocket/WebSocketSession.java @@ -16,9 +16,7 @@ package org.springframework.websocket; -import java.io.IOException; import java.net.URI; -import java.nio.ByteBuffer; @@ -51,14 +49,10 @@ public interface WebSocketSession { URI getURI(); /** - * Send a text message. + * Send a WebSocket message either {@link TextMessage} or + * {@link BinaryMessage}. */ - void sendTextMessage(String message) throws IOException; - - /** - * Send a binary message. - */ - void sendBinaryMessage(ByteBuffer message) throws IOException; + void sendMessage(WebSocketMessage message) throws Exception; /** * Close the WebSocket connection with status 1000, i.e. equivalent to: @@ -66,11 +60,11 @@ public interface WebSocketSession { * session.close(CloseStatus.NORMAL); * */ - void close() throws IOException; + void close() throws Exception; /** * Close the WebSocket connection with the given close status. */ - void close(CloseStatus status) throws IOException; + void close(CloseStatus status) throws Exception; } diff --git a/spring-websocket/src/main/java/org/springframework/websocket/endpoint/StandardWebSocketSession.java b/spring-websocket/src/main/java/org/springframework/websocket/endpoint/StandardWebSocketSession.java index 64f4d4d7fd..0da50203ee 100644 --- a/spring-websocket/src/main/java/org/springframework/websocket/endpoint/StandardWebSocketSession.java +++ b/spring-websocket/src/main/java/org/springframework/websocket/endpoint/StandardWebSocketSession.java @@ -18,15 +18,18 @@ package org.springframework.websocket.endpoint; import java.io.IOException; import java.net.URI; -import java.nio.ByteBuffer; import javax.websocket.CloseReason; import javax.websocket.CloseReason.CloseCodes; +import javax.websocket.RemoteEndpoint.Basic; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.util.Assert; +import org.springframework.websocket.BinaryMessage; import org.springframework.websocket.CloseStatus; +import org.springframework.websocket.TextMessage; +import org.springframework.websocket.WebSocketMessage; import org.springframework.websocket.WebSocketSession; @@ -70,21 +73,21 @@ public class StandardWebSocketSession implements WebSocketSession { } @Override - public void sendTextMessage(String text) throws IOException { + public void sendMessage(WebSocketMessage message) throws IOException { if (logger.isTraceEnabled()) { - logger.trace("Sending text message: " + text + ", " + this); + logger.trace("Sending: " + message + ", " + this); } Assert.isTrue(isOpen(), "Cannot send message after connection closed."); - this.session.getBasicRemote().sendText(text); - } - - @Override - public void sendBinaryMessage(ByteBuffer message) throws IOException { - if (logger.isTraceEnabled()) { - logger.trace("Sending binary message, " + this); + Basic remote = this.session.getBasicRemote(); + if (message instanceof TextMessage) { + remote.sendText(((TextMessage) message).getPayload()); + } + else if (message instanceof BinaryMessage) { + remote.sendBinary(((BinaryMessage) message).getPayload()); + } + else { + throw new IllegalStateException("Unexpected WebSocketMessage type: " + message); } - Assert.isTrue(isOpen(), "Cannot send message after connection closed."); - this.session.getBasicRemote().sendBinary(message); } @Override diff --git a/spring-websocket/src/main/java/org/springframework/websocket/endpoint/WebSocketHandlerEndpoint.java b/spring-websocket/src/main/java/org/springframework/websocket/endpoint/WebSocketHandlerEndpoint.java index 7593adbfd0..82d75c098d 100644 --- a/spring-websocket/src/main/java/org/springframework/websocket/endpoint/WebSocketHandlerEndpoint.java +++ b/spring-websocket/src/main/java/org/springframework/websocket/endpoint/WebSocketHandlerEndpoint.java @@ -28,8 +28,13 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.util.Assert; import org.springframework.websocket.CloseStatus; -import org.springframework.websocket.WebSocketSession; +import org.springframework.websocket.BinaryMessage; +import org.springframework.websocket.BinaryMessageHandler; +import org.springframework.websocket.PartialMessageHandler; import org.springframework.websocket.WebSocketHandler; +import org.springframework.websocket.WebSocketSession; +import org.springframework.websocket.TextMessage; +import org.springframework.websocket.TextMessageHandler; /** @@ -47,19 +52,50 @@ public class WebSocketHandlerEndpoint extends Endpoint { private final Map sessions = new ConcurrentHashMap(); - public WebSocketHandlerEndpoint(WebSocketHandler webSocketHandler) { - this.webSocketHandler = webSocketHandler; + public WebSocketHandlerEndpoint(WebSocketHandler handler) { + Assert.notNull(handler, "webSocketHandler is required"); + this.webSocketHandler = handler; } @Override - public void onOpen(javax.websocket.Session session, EndpointConfig config) { + public void onOpen(final javax.websocket.Session session, EndpointConfig config) { if (logger.isDebugEnabled()) { logger.debug("Client connected, WebSocket session id=" + session.getId() + ", uri=" + session.getRequestURI()); } try { WebSocketSession webSocketSession = new StandardWebSocketSession(session); this.sessions.put(session.getId(), webSocketSession); - session.addMessageHandler(new StandardMessageHandler(session)); + + if (this.webSocketHandler instanceof TextMessageHandler) { + session.addMessageHandler(new MessageHandler.Whole() { + @Override + public void onMessage(String message) { + handleTextMessage(session, message); + } + }); + } + else if (this.webSocketHandler instanceof BinaryMessageHandler) { + if (this.webSocketHandler instanceof PartialMessageHandler) { + session.addMessageHandler(new MessageHandler.Partial() { + @Override + public void onMessage(byte[] messagePart, boolean isLast) { + handleBinaryMessage(session, messagePart, isLast); + } + }); + } + else { + session.addMessageHandler(new MessageHandler.Whole() { + @Override + public void onMessage(byte[] message) { + handleBinaryMessage(session, message, true); + } + }); + } + } + else { + logger.warn("WebSocketHandler handles neither text nor binary messages: " + this.webSocketHandler); + } + this.webSocketHandler.afterConnectionEstablished(webSocketSession); } catch (Throwable ex) { @@ -68,6 +104,38 @@ public class WebSocketHandlerEndpoint extends Endpoint { } } + private void handleTextMessage(javax.websocket.Session session, String message) { + if (logger.isTraceEnabled()) { + logger.trace("Received message for WebSocket session id=" + session.getId() + ": " + message); + } + WebSocketSession wsSession = getWebSocketSession(session); + Assert.notNull(wsSession, "WebSocketSession not found"); + try { + TextMessage textMessage = new TextMessage(message); + ((TextMessageHandler) webSocketHandler).handleTextMessage(textMessage, wsSession); + } + catch (Throwable ex) { + // TODO + logger.error("Error while processing message", ex); + } + } + + private void handleBinaryMessage(javax.websocket.Session session, byte[] message, boolean isLast) { + if (logger.isTraceEnabled()) { + logger.trace("Received binary data for WebSocket session id=" + session.getId()); + } + WebSocketSession wsSession = getWebSocketSession(session); + Assert.notNull(wsSession, "WebSocketSession not found"); + try { + BinaryMessage binaryMessage = new BinaryMessage(message, isLast); + ((BinaryMessageHandler) webSocketHandler).handleBinaryMessage(binaryMessage, wsSession); + } + catch (Throwable ex) { + // TODO + logger.error("Error while processing message", ex); + } + } + @Override public void onClose(javax.websocket.Session session, CloseReason reason) { if (logger.isDebugEnabled()) { @@ -111,31 +179,4 @@ public class WebSocketHandlerEndpoint extends Endpoint { return this.sessions.get(session.getId()); } - - private class StandardMessageHandler implements MessageHandler.Whole { - - private final javax.websocket.Session session; - - public StandardMessageHandler(javax.websocket.Session session) { - this.session = session; - } - - @Override - public void onMessage(String message) { - if (logger.isTraceEnabled()) { - logger.trace("Received message for WebSocket session id=" + this.session.getId() + ": " + message); - } - WebSocketSession wsSession = getWebSocketSession(this.session); - Assert.notNull(wsSession, "WebSocketSession not found"); - try { - WebSocketHandlerEndpoint.this.webSocketHandler.handleTextMessage(message, wsSession); - } - catch (Throwable ex) { - // TODO - logger.error("Error while processing message", ex); - } - } - - } - } diff --git a/spring-websocket/src/main/java/org/springframework/websocket/server/DefaultHandshakeHandler.java b/spring-websocket/src/main/java/org/springframework/websocket/server/DefaultHandshakeHandler.java index cf4e4046f6..c7a3fdcb87 100644 --- a/spring-websocket/src/main/java/org/springframework/websocket/server/DefaultHandshakeHandler.java +++ b/spring-websocket/src/main/java/org/springframework/websocket/server/DefaultHandshakeHandler.java @@ -21,7 +21,6 @@ import java.nio.charset.Charset; import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; import java.util.Arrays; -import java.util.Collection; import java.util.Collections; import java.util.List; @@ -86,11 +85,6 @@ public class DefaultHandshakeHandler implements HandshakeHandler { return this.supportedProtocols.toArray(new String[this.supportedProtocols.size()]); } - @Override - public void registerWebSocketHandlers(Collection handlers) { - this.requestUpgradeStrategy.registerWebSocketHandlers(handlers); - } - @Override public final boolean doHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler webSocketHandler) throws Exception { diff --git a/spring-websocket/src/main/java/org/springframework/websocket/server/HandshakeHandler.java b/spring-websocket/src/main/java/org/springframework/websocket/server/HandshakeHandler.java index 8222e7258e..2af525c606 100644 --- a/spring-websocket/src/main/java/org/springframework/websocket/server/HandshakeHandler.java +++ b/spring-websocket/src/main/java/org/springframework/websocket/server/HandshakeHandler.java @@ -16,8 +16,6 @@ package org.springframework.websocket.server; -import java.util.Collection; - import org.springframework.http.server.ServerHttpRequest; import org.springframework.http.server.ServerHttpResponse; import org.springframework.websocket.WebSocketHandler; @@ -31,14 +29,6 @@ import org.springframework.websocket.WebSocketHandler; */ public interface HandshakeHandler { - /** - * Pre-register {@link WebSocketHandler} instances so they can be adapted to the - * underlying runtime and hence re-used at runtime when - * {@link #doHandshake(ServerHttpRequest, ServerHttpResponse, WebSocketHandler) doHandshake} - * is called. - */ - void registerWebSocketHandlers(Collection webSocketHandlers); - /** * * @param request the HTTP request diff --git a/spring-websocket/src/main/java/org/springframework/websocket/server/RequestUpgradeStrategy.java b/spring-websocket/src/main/java/org/springframework/websocket/server/RequestUpgradeStrategy.java index d0cf29fa17..81c238c423 100644 --- a/spring-websocket/src/main/java/org/springframework/websocket/server/RequestUpgradeStrategy.java +++ b/spring-websocket/src/main/java/org/springframework/websocket/server/RequestUpgradeStrategy.java @@ -16,8 +16,6 @@ package org.springframework.websocket.server; -import java.util.Collection; - import org.springframework.http.server.ServerHttpRequest; import org.springframework.http.server.ServerHttpResponse; import org.springframework.websocket.WebSocketHandler; @@ -37,14 +35,6 @@ public interface RequestUpgradeStrategy { */ String[] getSupportedVersions(); - /** - * Pre-register {@link WebSocketHandler} instances so they can be adapted to the - * underlying runtime and hence re-used at runtime when - * {@link #upgrade(ServerHttpRequest, ServerHttpResponse, String, WebSocketHandler) - * upgrade} is called. - */ - void registerWebSocketHandlers(Collection webSocketHandlers); - /** * Perform runtime specific steps to complete the upgrade. * Invoked only if the handshake is successful. diff --git a/spring-websocket/src/main/java/org/springframework/websocket/server/support/AbstractEndpointUpgradeStrategy.java b/spring-websocket/src/main/java/org/springframework/websocket/server/support/AbstractEndpointUpgradeStrategy.java index 29b2372f91..b16f3d746d 100644 --- a/spring-websocket/src/main/java/org/springframework/websocket/server/support/AbstractEndpointUpgradeStrategy.java +++ b/spring-websocket/src/main/java/org/springframework/websocket/server/support/AbstractEndpointUpgradeStrategy.java @@ -16,10 +16,6 @@ package org.springframework.websocket.server.support; -import java.util.Collection; -import java.util.HashMap; -import java.util.Map; - import javax.websocket.Endpoint; import org.apache.commons.logging.Log; @@ -42,38 +38,18 @@ public abstract class AbstractEndpointUpgradeStrategy implements RequestUpgradeS protected final Log logger = LogFactory.getLog(getClass()); - private final Map webSocketHandlers = new HashMap(); - @Override - public void registerWebSocketHandlers(Collection webSocketHandlers) { - for (WebSocketHandler webSocketHandler : webSocketHandlers) { - if (!this.webSocketHandlers.containsKey(webSocketHandler)) { - this.webSocketHandlers.put(webSocketHandler, adaptWebSocketHandler(webSocketHandler)); - } - } + public void upgrade(ServerHttpRequest request, ServerHttpResponse response, + String protocol, WebSocketHandler webSocketHandler) throws Exception { + + upgradeInternal(request, response, protocol, adaptWebSocketHandler(webSocketHandler)); } protected Endpoint adaptWebSocketHandler(WebSocketHandler handler) { return new WebSocketHandlerEndpoint(handler); } - @Override - public void upgrade(ServerHttpRequest request, ServerHttpResponse response, - String protocol, WebSocketHandler webSocketHandler) throws Exception { - - Endpoint endpoint = this.webSocketHandlers.get(webSocketHandler); - if (endpoint == null) { - endpoint = adaptWebSocketHandler(webSocketHandler); - } - - if (logger.isTraceEnabled()) { - logger.trace("Upgrading with " + endpoint); - } - - upgradeInternal(request, response, protocol, endpoint); - } - protected abstract void upgradeInternal(ServerHttpRequest request, ServerHttpResponse response, String selectedProtocol, Endpoint endpoint) throws Exception; diff --git a/spring-websocket/src/main/java/org/springframework/websocket/server/support/WebSocketHttpRequestHandler.java b/spring-websocket/src/main/java/org/springframework/websocket/server/support/WebSocketHttpRequestHandler.java index 6e641eaba0..9418afc689 100644 --- a/spring-websocket/src/main/java/org/springframework/websocket/server/support/WebSocketHttpRequestHandler.java +++ b/spring-websocket/src/main/java/org/springframework/websocket/server/support/WebSocketHttpRequestHandler.java @@ -17,7 +17,6 @@ package org.springframework.websocket.server.support; import java.io.IOException; -import java.util.Collections; import javax.servlet.ServletException; import javax.servlet.http.HttpServletRequest; @@ -35,8 +34,8 @@ import org.springframework.web.HttpRequestHandler; import org.springframework.web.util.NestedServletException; import org.springframework.websocket.HandlerProvider; import org.springframework.websocket.WebSocketHandler; -import org.springframework.websocket.server.HandshakeHandler; import org.springframework.websocket.server.DefaultHandshakeHandler; +import org.springframework.websocket.server.HandshakeHandler; /** @@ -56,7 +55,6 @@ public class WebSocketHttpRequestHandler implements HttpRequestHandler, BeanFact Assert.notNull(webSocketHandler, "webSocketHandler is required"); this.handlerProvider = new HandlerProvider(webSocketHandler); this.handshakeHandler = new DefaultHandshakeHandler(); - this.handshakeHandler.registerWebSocketHandlers(Collections.singleton(webSocketHandler)); } public WebSocketHttpRequestHandler( Class webSocketHandlerClass) { @@ -67,9 +65,6 @@ public class WebSocketHttpRequestHandler implements HttpRequestHandler, BeanFact public void setHandshakeHandler(HandshakeHandler handshakeHandler) { Assert.notNull(handshakeHandler, "handshakeHandler is required"); this.handshakeHandler = handshakeHandler; - if (this.handlerProvider.isSingleton()) { - this.handshakeHandler.registerWebSocketHandlers(Collections.singleton(this.handlerProvider.getHandler())); - } } @Override