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 0c179b9dd0..a39e5d6789 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 @@ -129,13 +129,13 @@ public abstract class AbstractServerSockJsSession extends AbstractSockJsSession } protected void scheduleHeartbeat() { - Assert.notNull(getSockJsConfig().getHeartbeatScheduler(), "heartbeatScheduler not configured"); + Assert.notNull(getSockJsConfig().getTaskScheduler(), "heartbeatScheduler not configured"); cancelHeartbeat(); if (!isActive()) { return; } Date time = new Date(System.currentTimeMillis() + getSockJsConfig().getHeartbeatTime()); - this.heartbeatTask = getSockJsConfig().getHeartbeatScheduler().schedule(new Runnable() { + this.heartbeatTask = getSockJsConfig().getTaskScheduler().schedule(new Runnable() { public void run() { try { sendHeartbeat(); 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 0319ff41bc..e44c44cec0 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 @@ -25,15 +25,12 @@ import java.util.Random; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.beans.factory.DisposableBean; -import org.springframework.beans.factory.InitializingBean; import org.springframework.http.HttpMethod; 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.scheduling.TaskScheduler; -import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.DigestUtils; @@ -49,8 +46,7 @@ import org.springframework.websocket.WebSocketHandler; * @author Rossen Stoyanchev * @since 4.0 */ -public abstract class AbstractSockJsService - implements SockJsService, SockJsConfiguration, InitializingBean, DisposableBean { +public abstract class AbstractSockJsService implements SockJsService, SockJsConfiguration { protected final Log logger = LogFactory.getLog(getClass()); @@ -71,17 +67,13 @@ public abstract class AbstractSockJsService private boolean webSocketsEnabled = true; - private final TaskSchedulerHolder heartbeatSchedulerHolder; + private final TaskScheduler taskScheduler; - public AbstractSockJsService() { - this.heartbeatSchedulerHolder = new TaskSchedulerHolder("SockJs-heartbeat-"); - } - - public AbstractSockJsService(TaskScheduler heartbeatScheduler) { - Assert.notNull(heartbeatScheduler, "heartbeatScheduler is required"); - this.heartbeatSchedulerHolder = new TaskSchedulerHolder(heartbeatScheduler); + public AbstractSockJsService(TaskScheduler scheduler) { + Assert.notNull(scheduler, "scheduler is required"); + this.taskScheduler = scheduler; } /** @@ -159,8 +151,8 @@ public abstract class AbstractSockJsService return this.heartbeatTime; } - public TaskScheduler getHeartbeatScheduler() { - return this.heartbeatSchedulerHolder.getScheduler(); + public TaskScheduler getTaskScheduler() { + return this.taskScheduler; } /** @@ -199,16 +191,6 @@ public abstract class AbstractSockJsService return this.webSocketsEnabled; } - @Override - public void afterPropertiesSet() throws Exception { - this.heartbeatSchedulerHolder.initialize(); - } - - @Override - public void destroy() throws Exception { - this.heartbeatSchedulerHolder.destroy(); - } - /** * TODO * @@ -423,46 +405,4 @@ public abstract class AbstractSockJsService } }; - - /** - * Holds an externally provided or an internally managed TaskScheduler. Provides - * initialize and destroy methods have no effect if the scheduler is externally - * managed. - */ - protected static class TaskSchedulerHolder { - - private final TaskScheduler taskScheduler; - - private final boolean isDefaultTaskScheduler; - - public TaskSchedulerHolder(TaskScheduler taskScheduler) { - Assert.notNull(taskScheduler, "taskScheduler is required"); - this.taskScheduler = taskScheduler; - this.isDefaultTaskScheduler = false; - } - - public TaskSchedulerHolder(String threadNamePrefix) { - ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); - scheduler.setThreadNamePrefix(threadNamePrefix); - this.taskScheduler = scheduler; - this.isDefaultTaskScheduler = true; - } - - public TaskScheduler getScheduler() { - return this.taskScheduler; - } - - public void initialize() { - if (this.isDefaultTaskScheduler) { - ((ThreadPoolTaskScheduler) this.taskScheduler).afterPropertiesSet(); - } - } - - public void destroy() { - if (this.isDefaultTaskScheduler) { - ((ThreadPoolTaskScheduler) this.taskScheduler).shutdown(); - } - } - } - } diff --git a/spring-websocket/src/main/java/org/springframework/sockjs/server/SockJsConfiguration.java b/spring-websocket/src/main/java/org/springframework/sockjs/server/SockJsConfiguration.java index 4bf1db79e8..1fa79db883 100644 --- a/spring-websocket/src/main/java/org/springframework/sockjs/server/SockJsConfiguration.java +++ b/spring-websocket/src/main/java/org/springframework/sockjs/server/SockJsConfiguration.java @@ -48,10 +48,8 @@ public interface SockJsConfiguration { public long getHeartbeatTime(); /** - * A scheduler instance to use for scheduling heartbeat frames. - *

- * By default a {@link ThreadPoolTaskScheduler} with default settings is used. + * A scheduler instance to use for scheduling heart-beat messages. */ - public TaskScheduler getHeartbeatScheduler(); + public TaskScheduler getTaskScheduler(); } 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 f54b540eb1..5a4baadf3f 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,10 +16,15 @@ package org.springframework.sockjs.server.support; import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; import java.util.HashMap; +import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ScheduledFuture; import org.springframework.http.Cookie; import org.springframework.http.HttpMethod; @@ -41,7 +46,7 @@ import org.springframework.sockjs.server.transport.WebSocketTransportHandler; import org.springframework.sockjs.server.transport.XhrPollingTransportHandler; import org.springframework.sockjs.server.transport.XhrStreamingTransportHandler; import org.springframework.sockjs.server.transport.XhrTransportHandler; -import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; import org.springframework.websocket.HandlerProvider; import org.springframework.websocket.WebSocketHandler; import org.springframework.websocket.server.DefaultHandshakeHandler; @@ -58,101 +63,58 @@ public class DefaultSockJsService extends AbstractSockJsService { private final Map transportHandlers = new HashMap(); - private final Map transportHandlerOverrides = new HashMap(); - - private TaskSchedulerHolder sessionTimeoutSchedulerHolder; - private final Map sessions = new ConcurrentHashMap(); + private ScheduledFuture sessionCleanupTask; - public DefaultSockJsService() { - this.sessionTimeoutSchedulerHolder = new TaskSchedulerHolder("SockJs-sessionTimeout-"); + + public DefaultSockJsService(TaskScheduler taskScheduler) { + this(taskScheduler, null); } - public DefaultSockJsService(TaskScheduler heartbeatScheduler, TaskScheduler sessionTimeoutScheduler) { - Assert.notNull(sessionTimeoutScheduler, "sessionTimeoutScheduler is required"); - this.sessionTimeoutSchedulerHolder = new TaskSchedulerHolder(sessionTimeoutScheduler); + public DefaultSockJsService(TaskScheduler taskScheduler, Set transportHandlers, + TransportHandler... transportHandlerOverrides) { + + super(taskScheduler); + + transportHandlers = CollectionUtils.isEmpty(transportHandlers) ? getDefaultTransportHandlers() : transportHandlers; + addTransportHandlers(transportHandlers); + addTransportHandlers(Arrays.asList(transportHandlerOverrides)); } - public void setTransportHandlers(TransportHandler... handlers) { - this.transportHandlers.clear(); + protected Set getDefaultTransportHandlers() { + Set result = new HashSet(); + result.add(new XhrPollingTransportHandler()); + result.add(new XhrTransportHandler()); + result.add(new JsonpPollingTransportHandler()); + result.add(new JsonpTransportHandler()); + result.add(new XhrStreamingTransportHandler()); + result.add(new EventSourceTransportHandler()); + result.add(new HtmlFileTransportHandler()); + if (isWebSocketEnabled()) { + try { + result.add(new WebSocketTransportHandler(new DefaultHandshakeHandler())); + } + catch (Exception ex) { + if (logger.isWarnEnabled()) { + logger.warn("Failed to add default WebSocketTransportHandler: " + ex.getMessage()); + } + } + } + return result; + } + + protected void addTransportHandlers(Collection handlers) { for (TransportHandler handler : handlers) { + if (handler instanceof ConfigurableTransportHandler) { + ((ConfigurableTransportHandler) handler).setSockJsConfiguration(this); + } this.transportHandlers.put(handler.getTransportType(), handler); } } - public void setTransportHandlerOverrides(TransportHandler... handlers) { - this.transportHandlerOverrides.clear(); - for (TransportHandler handler : handlers) { - this.transportHandlerOverrides.put(handler.getTransportType(), handler); - } - } - - - @Override - public void afterPropertiesSet() throws Exception { - - super.afterPropertiesSet(); - - if (this.transportHandlers.isEmpty()) { - if (isWebSocketEnabled() && (this.transportHandlerOverrides.get(TransportType.WEBSOCKET) == null)) { - this.transportHandlers.put(TransportType.WEBSOCKET, - new WebSocketTransportHandler(new DefaultHandshakeHandler())); - } - this.transportHandlers.put(TransportType.XHR, new XhrPollingTransportHandler()); - this.transportHandlers.put(TransportType.XHR_SEND, new XhrTransportHandler()); - this.transportHandlers.put(TransportType.JSONP, new JsonpPollingTransportHandler()); - this.transportHandlers.put(TransportType.JSONP_SEND, new JsonpTransportHandler()); - this.transportHandlers.put(TransportType.XHR_STREAMING, new XhrStreamingTransportHandler()); - this.transportHandlers.put(TransportType.EVENT_SOURCE, new EventSourceTransportHandler()); - this.transportHandlers.put(TransportType.HTML_FILE, new HtmlFileTransportHandler()); - } - - if (!this.transportHandlerOverrides.isEmpty()) { - for (TransportHandler transportHandler : this.transportHandlerOverrides.values()) { - this.transportHandlers.put(transportHandler.getTransportType(), transportHandler); - } - } - - for (TransportHandler h : this.transportHandlers.values()) { - if (h instanceof ConfigurableTransportHandler) { - ((ConfigurableTransportHandler) h).setSockJsConfiguration(this); - } - } - - this.sessionTimeoutSchedulerHolder.initialize(); - - this.sessionTimeoutSchedulerHolder.getScheduler().scheduleAtFixedRate(new Runnable() { - public void run() { - try { - int count = sessions.size(); - if (logger.isTraceEnabled() && (count != 0)) { - logger.trace("Checking " + count + " session(s) for timeouts [" + getName() + "]"); - } - for (AbstractSockJsSession session : sessions.values()) { - if (session.getTimeSinceLastActive() > getDisconnectDelay()) { - if (logger.isTraceEnabled()) { - logger.trace("Removing " + session + " for [" + getName() + "]"); - } - session.close(); - sessions.remove(session.getId()); - } - } - if (logger.isTraceEnabled() && (count != 0)) { - logger.trace(sessions.size() + " remaining session(s) [" + getName() + "]"); - } - } - catch (Throwable t) { - logger.error("Failed to complete session timeout checks for [" + getName() + "]", t); - } - } - }, getDisconnectDelay()); - } - - @Override - public void destroy() throws Exception { - super.destroy(); - this.sessionTimeoutSchedulerHolder.destroy(); + public Map getTransportHandlers() { + return Collections.unmodifiableMap(this.transportHandlers); } @Override @@ -239,6 +201,9 @@ public class DefaultSockJsService extends AbstractSockJsService { if (session != null) { return session; } + if (this.sessionCleanupTask == null) { + scheduleSessionTask(); + } logger.debug("Creating new session with session id \"" + sessionId + "\""); session = (AbstractSockJsSession) sessionFactory.createSession(sessionId, handler); this.sessions.put(sessionId, session); @@ -249,4 +214,32 @@ public class DefaultSockJsService extends AbstractSockJsService { return null; } + private void scheduleSessionTask() { + this.sessionCleanupTask = getTaskScheduler().scheduleAtFixedRate(new Runnable() { + public void run() { + try { + int count = sessions.size(); + if (logger.isTraceEnabled() && (count != 0)) { + logger.trace("Checking " + count + " session(s) for timeouts [" + getName() + "]"); + } + for (AbstractSockJsSession session : sessions.values()) { + if (session.getTimeSinceLastActive() > getDisconnectDelay()) { + if (logger.isTraceEnabled()) { + logger.trace("Removing " + session + " for [" + getName() + "]"); + } + session.close(); + sessions.remove(session.getId()); + } + } + if (logger.isTraceEnabled() && (count != 0)) { + logger.trace(sessions.size() + " remaining session(s) [" + getName() + "]"); + } + } + catch (Throwable t) { + logger.error("Failed to complete session timeout checks for [" + getName() + "]", t); + } + } + }, getDisconnectDelay()); + } + } diff --git a/spring-websocket/src/main/java/org/springframework/websocket/client/WebSocketConnectionManager.java b/spring-websocket/src/main/java/org/springframework/websocket/client/WebSocketConnectionManager.java index 29b465e477..05eddadd77 100644 --- a/spring-websocket/src/main/java/org/springframework/websocket/client/WebSocketConnectionManager.java +++ b/spring-websocket/src/main/java/org/springframework/websocket/client/WebSocketConnectionManager.java @@ -15,9 +15,11 @@ */ package org.springframework.websocket.client; +import java.util.ArrayList; import java.util.List; import org.springframework.http.HttpHeaders; +import org.springframework.util.CollectionUtils; import org.springframework.websocket.HandlerProvider; import org.springframework.websocket.WebSocketHandler; import org.springframework.websocket.WebSocketSession; @@ -37,7 +39,7 @@ public class WebSocketConnectionManager extends AbstractWebSocketConnectionManag private WebSocketSession webSocketSession; - private List subProtocols; + private final List subProtocols = new ArrayList(); public WebSocketConnectionManager(WebSocketClient webSocketClient, @@ -57,7 +59,10 @@ public class WebSocketConnectionManager extends AbstractWebSocketConnectionManag } public void setSubProtocols(List subProtocols) { - this.subProtocols = subProtocols; + this.subProtocols.clear(); + if (!CollectionUtils.isEmpty(subProtocols)) { + this.subProtocols.addAll(subProtocols); + } } public List getSubProtocols() { diff --git a/spring-websocket/src/test/java/org/springframework/sockjs/server/support/DefaultSockJsServiceTests.java b/spring-websocket/src/test/java/org/springframework/sockjs/server/support/DefaultSockJsServiceTests.java new file mode 100644 index 0000000000..77ee2edf15 --- /dev/null +++ b/spring-websocket/src/test/java/org/springframework/sockjs/server/support/DefaultSockJsServiceTests.java @@ -0,0 +1,55 @@ +/* + * 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.support; + +import java.util.Map; + +import org.junit.Test; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import org.springframework.sockjs.server.TransportHandler; +import org.springframework.sockjs.server.TransportType; + +import static org.junit.Assert.*; + + +/** + * Test fixture for {@link DefaultSockJsService}. + * + * @author Rossen Stoyanchev + */ +public class DefaultSockJsServiceTests { + + + @Test + public void testDefaultTransportHandlers() { + + DefaultSockJsService sockJsService = new DefaultSockJsService(new ThreadPoolTaskScheduler()); + Map handlers = sockJsService.getTransportHandlers(); + + assertEquals(8, handlers.size()); + assertNotNull(handlers.get(TransportType.WEBSOCKET)); + assertNotNull(handlers.get(TransportType.XHR)); + assertNotNull(handlers.get(TransportType.XHR_SEND)); + assertNotNull(handlers.get(TransportType.XHR_STREAMING)); + assertNotNull(handlers.get(TransportType.JSONP)); + assertNotNull(handlers.get(TransportType.JSONP_SEND)); + assertNotNull(handlers.get(TransportType.HTML_FILE)); + assertNotNull(handlers.get(TransportType.EVENT_SOURCE)); + } + + +}