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