root
/
spring-framework
mirror of https://github.com/spring-projects/spring-framework.git
1
0
Fork 0
Code Issues Actions Packages Projects Releases Wiki Activity

Add WebSocketMessage and WebSocketHandler sub-interfcs 

There is now a WebSocketMessage type with TextMessage and BinaryMessage
sub-types. WebSocketHandler is also sub-divided into TextMessageHandler
and BinaryMessageHandler, so that applications can choose to handle
text, binary, or both.

Also in this commit, the SockJsHandler and SockJsSession interfaces
have been removed. SockJsService now accepts WebSocketHandler.
This commit is contained in:
Rossen Stoyanchev 2013-04-22 17:59:19 -04:00
parent 2046629945
commit f9078c947f
39 changed files with 513 additions and 666 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
spring-websocket/src/main/java/org/springframework/sockjs/SockJsSessionSupport.java → spring-websocket/src/main/java/org/springframework/sockjs/AbstractSockJsSession.java
View File

@ -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}
* <p>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}
* <p>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

51
spring-websocket/src/main/java/org/springframework/sockjs/SockJsHandler.java
View File

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

62
spring-websocket/src/main/java/org/springframework/sockjs/SockJsSession.java
View File

@ -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:
* <pre>
* session.close(CloseStatus.NORMAL);
* </pre>
*/
void close() throws IOException;
/**
* Close the underlying connection with the given close status.
*/
void close(CloseStatus status) throws IOException;
}

7
spring-websocket/src/main/java/org/springframework/sockjs/SockJsSessionFactory.java
View File

@ -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<S extends SockJsSession>{
public interface SockJsSessionFactory<S extends WebSocketSession>{
S createSession(String sessionId, SockJsHandler sockJsHandler);
S createSession(String sessionId, WebSocketHandler webSocketHandler);
}

34
spring-websocket/src/main/java/org/springframework/sockjs/server/AbstractServerSockJsSession.java
View File

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

12
spring-websocket/src/main/java/org/springframework/sockjs/server/AbstractSockJsService.java
View File

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

11
spring-websocket/src/main/java/org/springframework/sockjs/server/ConfigurableTransportHandler.java
View File

@ -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<SockJsHandler> sockJsHandlers);
}

12
spring-websocket/src/main/java/org/springframework/sockjs/server/SockJsService.java
View File

@ -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<SockJsHandler> sockJsHandlers);
void handleRequest(ServerHttpRequest request, ServerHttpResponse response, String sockJsPath,
SockJsHandler handler) throws Exception;
WebSocketHandler webSocketHandler) throws Exception;
}

6
spring-websocket/src/main/java/org/springframework/sockjs/server/TransportHandler.java
View File

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

65
spring-websocket/src/main/java/org/springframework/sockjs/server/support/DefaultSockJsService.java
View File

@ -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<String, SockJsSessionSupport> sessions = new ConcurrentHashMap<String, SockJsSessionSupport>();
private final Map<SockJsHandler, WebSocketHandler> sockJsHandlers = new HashMap<SockJsHandler, WebSocketHandler>();
private final Map<String, AbstractSockJsSession> sessions = new ConcurrentHashMap<String, AbstractSockJsSession>();
public DefaultSockJsService() {
@ -93,22 +88,6 @@ public class DefaultSockJsService extends AbstractSockJsService implements Initi
}
}
public void registerSockJsHandlers(Collection<SockJsHandler> 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
* <em>raw WebSocket</em> 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;
}

22
spring-websocket/src/main/java/org/springframework/sockjs/server/support/SockJsHttpRequestHandler.java
View File

@ -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<SockJsHandler> handlerProvider;
private final HandlerProvider<WebSocketHandler> 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>(sockJsHandler);
this.handlerProvider = new HandlerProvider<WebSocketHandler>(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<? extends SockJsHandler> sockJsHandlerClass) {
Class<? extends WebSocketHandler> 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<SockJsHandler>(sockJsHandlerClass);
this.handlerProvider = new HandlerProvider<WebSocketHandler>(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

8
spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractHttpReceivingTransportHandler.java
View File

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

14
spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractHttpSendingTransportHandler.java
View File

@ -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<SockJsSessionSupport> {
implements ConfigurableTransportHandler, SockJsSessionFactory<AbstractSockJsSession> {
protected final Log logger = LogFactory.getLog(this.getClass());
@ -50,18 +49,13 @@ public abstract class AbstractHttpSendingTransportHandler
this.sockJsConfig = sockJsConfig;
}
@Override
public void registerSockJsHandlers(Collection<SockJsHandler> 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());

16
spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractHttpServerSockJsSession.java
View File

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

6
spring-websocket/src/main/java/org/springframework/sockjs/server/transport/AbstractStreamingTransportHandler.java
View File

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

6
spring-websocket/src/main/java/org/springframework/sockjs/server/transport/JsonpPollingTransportHandler.java
View File

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

4
spring-websocket/src/main/java/org/springframework/sockjs/server/transport/JsonpTransportHandler.java
View File

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

14
spring-websocket/src/main/java/org/springframework/sockjs/server/transport/PollingServerSockJsSession.java
View File

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

60
spring-websocket/src/main/java/org/springframework/sockjs/server/transport/SockJsWebSocketHandler.java
View File

@ -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<WebSocketSession, SockJsSessionSupport> sessions =
new ConcurrentHashMap<WebSocketSession, SockJsSessionSupport>();
private final Map<WebSocketSession, AbstractSockJsSession> sessions =
new ConcurrentHashMap<WebSocketSession, AbstractSockJsSession>();
// 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);
}
}

10
spring-websocket/src/main/java/org/springframework/sockjs/server/transport/StreamingServerSockJsSession.java
View File

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

132
spring-websocket/src/main/java/org/springframework/sockjs/server/transport/WebSocketSockJsHandlerAdapter.java
View File

@ -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<WebSocketSession, SockJsSessionSupport> sessions =
new ConcurrentHashMap<WebSocketSession, SockJsSessionSupport>();
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);
}
}
}

54
spring-websocket/src/main/java/org/springframework/sockjs/server/transport/WebSocketTransportHandler.java
View File

@ -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<SockJsHandler, WebSocketHandler> sockJsHandlerCache = new HashMap<SockJsHandler, WebSocketHandler>();
private final Collection<WebSocketHandler> rawWebSocketHandlerCache = new ArrayList<WebSocketHandler>();
public WebSocketTransportHandler(HandshakeHandler handshakeHandler) {
Assert.notNull(handshakeHandler, "handshakeHandler is required");
@ -71,50 +59,22 @@ public class WebSocketTransportHandler implements ConfigurableTransportHandler,
}
@Override
public void registerSockJsHandlers(Collection<SockJsHandler> 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<WebSocketHandler> getAllWebSocketHandlers() {
Set<WebSocketHandler> handlers = new HashSet<WebSocketHandler>();
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<WebSocketHandler> webSocketHandlers) {
this.rawWebSocketHandlerCache.clear();
this.rawWebSocketHandlerCache.addAll(webSocketHandlers);
this.handshakeHandler.registerWebSocketHandlers(getAllWebSocketHandlers());
}
@Override
public boolean doHandshake(ServerHttpRequest request, ServerHttpResponse response,
WebSocketHandler webSocketHandler) throws Exception {

6
spring-websocket/src/main/java/org/springframework/sockjs/server/transport/XhrPollingTransportHandler.java
View File

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

85
spring-websocket/src/main/java/org/springframework/websocket/BinaryMessage.java Normal file
View File

@ -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<ByteBuffer> {
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;
}
}

28
spring-websocket/src/main/java/org/springframework/sockjs/SockJsHandlerAdapter.java → spring-websocket/src/main/java/org/springframework/websocket/BinaryMessageHandler.java
View File

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

28
spring-websocket/src/main/java/org/springframework/websocket/PartialMessageHandler.java Normal file
View File

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

25
spring-websocket/src/main/java/org/springframework/websocket/WebSocketHandshakeRequest.java → spring-websocket/src/main/java/org/springframework/websocket/TextMessage.java
View File

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

40
spring-websocket/src/main/java/org/springframework/websocket/TextMessageHandler.java Normal file
View File

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

16
spring-websocket/src/main/java/org/springframework/websocket/WebSocketHandler.java
View File

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

20
spring-websocket/src/main/java/org/springframework/websocket/WebSocketHandlerAdapter.java
View File

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

61
spring-websocket/src/main/java/org/springframework/websocket/WebSocketMessage.java Normal file
View File

@ -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<T> {
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);
}
}

16
spring-websocket/src/main/java/org/springframework/websocket/WebSocketSession.java
View File

@ -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);
* </pre>
*/
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;
}

27
spring-websocket/src/main/java/org/springframework/websocket/endpoint/StandardWebSocketSession.java
View File

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

105
spring-websocket/src/main/java/org/springframework/websocket/endpoint/WebSocketHandlerEndpoint.java
View File

@ -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<String, WebSocketSession> sessions = new ConcurrentHashMap<String, WebSocketSession>();
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<String>() {
@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<byte[]>() {
@Override
public void onMessage(byte[] messagePart, boolean isLast) {
handleBinaryMessage(session, messagePart, isLast);
}
});
}
else {
session.addMessageHandler(new MessageHandler.Whole<byte[]>() {
@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<String> {
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);
}
}
}
}

6
spring-websocket/src/main/java/org/springframework/websocket/server/DefaultHandshakeHandler.java
View File

@ -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<WebSocketHandler> handlers) {
this.requestUpgradeStrategy.registerWebSocketHandlers(handlers);
}
@Override
public final boolean doHandshake(ServerHttpRequest request, ServerHttpResponse response,
WebSocketHandler webSocketHandler) throws Exception {

10
spring-websocket/src/main/java/org/springframework/websocket/server/HandshakeHandler.java
View File

@ -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<WebSocketHandler> webSocketHandlers);
/**
*
* @param request the HTTP request

10
spring-websocket/src/main/java/org/springframework/websocket/server/RequestUpgradeStrategy.java
View File

@ -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<WebSocketHandler> webSocketHandlers);
/**
* Perform runtime specific steps to complete the upgrade.
* Invoked only if the handshake is successful.

32
spring-websocket/src/main/java/org/springframework/websocket/server/support/AbstractEndpointUpgradeStrategy.java
View File

@ -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<WebSocketHandler, Endpoint> webSocketHandlers = new HashMap<WebSocketHandler, Endpoint>();
@Override
public void registerWebSocketHandlers(Collection<WebSocketHandler> 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;

7
spring-websocket/src/main/java/org/springframework/websocket/server/support/WebSocketHttpRequestHandler.java
View File

@ -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>(webSocketHandler);
this.handshakeHandler = new DefaultHandshakeHandler();
this.handshakeHandler.registerWebSocketHandlers(Collections.singleton(webSocketHandler));
}
public WebSocketHttpRequestHandler( Class<? extends WebSocketHandler> 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