diff --git a/spring-context/src/main/java/org/springframework/scheduling/concurrent/ScheduledExecutorFactoryBean.java b/spring-context/src/main/java/org/springframework/scheduling/concurrent/ScheduledExecutorFactoryBean.java index 2990e8b3a2..7513226ae2 100644 --- a/spring-context/src/main/java/org/springframework/scheduling/concurrent/ScheduledExecutorFactoryBean.java +++ b/spring-context/src/main/java/org/springframework/scheduling/concurrent/ScheduledExecutorFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -24,9 +24,11 @@ import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.ThreadFactory; import org.springframework.beans.factory.FactoryBean; +import org.springframework.lang.UsesJava7; import org.springframework.scheduling.support.DelegatingErrorHandlingRunnable; import org.springframework.scheduling.support.TaskUtils; import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; import org.springframework.util.ObjectUtils; /** @@ -56,6 +58,7 @@ import org.springframework.util.ObjectUtils; * @author Juergen Hoeller * @since 2.0 * @see #setPoolSize + * @see #setRemoveOnCancelPolicy(boolean) * @see #setThreadFactory * @see ScheduledExecutorTask * @see java.util.concurrent.ScheduledExecutorService @@ -65,8 +68,15 @@ import org.springframework.util.ObjectUtils; public class ScheduledExecutorFactoryBean extends ExecutorConfigurationSupport implements FactoryBean { + // Check for setRemoveOnCancelPolicy method - available on JDK 7 and higher + private static boolean hasRemoveOnCancelPolicyMethod = ClassUtils.hasMethod( + ScheduledThreadPoolExecutor.class, "setRemoveOnCancelPolicy", boolean.class); + + private int poolSize = 1; + private Boolean removeOnCancelPolicy; + private ScheduledExecutorTask[] scheduledExecutorTasks; private boolean continueScheduledExecutionAfterException = false; @@ -85,6 +95,15 @@ public class ScheduledExecutorFactoryBean extends ExecutorConfigurationSupport this.poolSize = poolSize; } + /** + * Set the same property on ScheduledExecutorService available in JDK 1.7 or + * higher. This property is ignored on JDK 1.6. + * Default is false. + */ + public void setRemoveOnCancelPolicy(boolean removeOnCancelPolicy) { + this.removeOnCancelPolicy = removeOnCancelPolicy; + } + /** * Register a list of ScheduledExecutorTask objects with the ScheduledExecutorService * that this FactoryBean creates. Depending on each ScheduledExecutorTask's settings, @@ -130,6 +149,10 @@ public class ScheduledExecutorFactoryBean extends ExecutorConfigurationSupport ScheduledExecutorService executor = createExecutor(this.poolSize, threadFactory, rejectedExecutionHandler); + if (executor instanceof ScheduledThreadPoolExecutor) { + configureRemoveOnCancelPolicy(((ScheduledThreadPoolExecutor) executor)); + } + // Register specified ScheduledExecutorTasks, if necessary. if (!ObjectUtils.isEmpty(this.scheduledExecutorTasks)) { registerTasks(this.scheduledExecutorTasks, executor); @@ -199,6 +222,13 @@ public class ScheduledExecutorFactoryBean extends ExecutorConfigurationSupport new DelegatingErrorHandlingRunnable(task.getRunnable(), TaskUtils.LOG_AND_PROPAGATE_ERROR_HANDLER)); } + @UsesJava7 // guard setting removeOnCancelPolicy (safe with 1.6 due to hasRemoveOnCancelPolicyMethod check) + private void configureRemoveOnCancelPolicy(ScheduledThreadPoolExecutor service) { + if (hasRemoveOnCancelPolicyMethod && this.removeOnCancelPolicy != null) { + service.setRemoveOnCancelPolicy(true); + } + } + @Override public ScheduledExecutorService getObject() { diff --git a/spring-context/src/main/java/org/springframework/scheduling/concurrent/ThreadPoolTaskScheduler.java b/spring-context/src/main/java/org/springframework/scheduling/concurrent/ThreadPoolTaskScheduler.java index be5f525df7..ff23a2e75d 100644 --- a/spring-context/src/main/java/org/springframework/scheduling/concurrent/ThreadPoolTaskScheduler.java +++ b/spring-context/src/main/java/org/springframework/scheduling/concurrent/ThreadPoolTaskScheduler.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2014 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. @@ -31,11 +31,13 @@ import java.util.concurrent.TimeUnit; import org.springframework.core.task.AsyncListenableTaskExecutor; import org.springframework.core.task.TaskRejectedException; +import org.springframework.lang.UsesJava7; import org.springframework.scheduling.SchedulingTaskExecutor; import org.springframework.scheduling.TaskScheduler; import org.springframework.scheduling.Trigger; import org.springframework.scheduling.support.TaskUtils; import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; import org.springframework.util.ErrorHandler; import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFutureTask; @@ -48,6 +50,7 @@ import org.springframework.util.concurrent.ListenableFutureTask; * @author Mark Fisher * @since 3.0 * @see #setPoolSize + * @see #setRemoveOnCancelPolicy(boolean) * @see #setThreadFactory * @see #setErrorHandler */ @@ -55,8 +58,15 @@ import org.springframework.util.concurrent.ListenableFutureTask; public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport implements AsyncListenableTaskExecutor, SchedulingTaskExecutor, TaskScheduler { + // Check for setRemoveOnCancelPolicy method - available on JDK 7 and higher + private static boolean hasRemoveOnCancelPolicyMethod = ClassUtils.hasMethod( + ScheduledThreadPoolExecutor.class, "setRemoveOnCancelPolicy", boolean.class); + + private volatile int poolSize = 1; + private volatile Boolean removeOnCancelPolicy; + private volatile ScheduledExecutorService scheduledExecutor; private volatile ErrorHandler errorHandler; @@ -75,6 +85,27 @@ public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport } } + /** + * Set the same property on ScheduledExecutorService available in JDK 1.7 or + * higher. This property is ignored on JDK 1.6. + * Default is false. + *

This setting can be modified at runtime, for example through JMX. + */ + @UsesJava7 // guard setting removeOnCancelPolicy (safe with 1.6 due to hasRemoveOnCancelPolicyMethod check) + public void setRemoveOnCancelPolicy(boolean removeOnCancelPolicy) { + this.removeOnCancelPolicy = removeOnCancelPolicy; + if (this.scheduledExecutor instanceof ScheduledThreadPoolExecutor) { + configureRemoveOnCancelPolicy((ScheduledThreadPoolExecutor) this.scheduledExecutor); + } + } + + @UsesJava7 // guard setting removeOnCancelPolicy (safe with 1.6 due to hasRemoveOnCancelPolicyMethod check) + private void configureRemoveOnCancelPolicy(ScheduledThreadPoolExecutor service) { + if (hasRemoveOnCancelPolicyMethod && this.removeOnCancelPolicy != null) { + service.setRemoveOnCancelPolicy(true); + } + } + /** * Set a custom {@link ErrorHandler} strategy. */ @@ -88,6 +119,11 @@ public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) { this.scheduledExecutor = createExecutor(this.poolSize, threadFactory, rejectedExecutionHandler); + + if (this.scheduledExecutor instanceof ScheduledThreadPoolExecutor) { + configureRemoveOnCancelPolicy(((ScheduledThreadPoolExecutor) this.scheduledExecutor)); + } + return this.scheduledExecutor; } @@ -145,6 +181,14 @@ public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupport return getScheduledThreadPoolExecutor().getPoolSize(); } + /** + * Return the current setting of removeOnCancelPolicy. + *

Requires an underlying {@link ScheduledThreadPoolExecutor} and JDK 1.7+. + */ + public boolean isRemoveOnCancelPolicy() { + return this.removeOnCancelPolicy; + } + /** * Return the number of currently active threads. *

Requires an underlying {@link ScheduledThreadPoolExecutor}. diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/config/WebSocketNamespaceUtils.java b/spring-websocket/src/main/java/org/springframework/web/socket/config/WebSocketNamespaceUtils.java index d1372cdca9..daad1af288 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/config/WebSocketNamespaceUtils.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/config/WebSocketNamespaceUtils.java @@ -16,7 +16,7 @@ package org.springframework.web.socket.config; -import org.springframework.web.socket.sockjs.transport.SockJsThreadPoolTaskScheduler; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.w3c.dom.Element; import org.springframework.beans.factory.config.BeanDefinition; @@ -135,11 +135,12 @@ class WebSocketNamespaceUtils { ParserContext parserContext, Object source) { if (!parserContext.getRegistry().containsBeanDefinition(schedulerName)) { - RootBeanDefinition taskSchedulerDef = new RootBeanDefinition(SockJsThreadPoolTaskScheduler.class); + RootBeanDefinition taskSchedulerDef = new RootBeanDefinition(ThreadPoolTaskScheduler.class); taskSchedulerDef.setSource(source); taskSchedulerDef.setRole(BeanDefinition.ROLE_INFRASTRUCTURE); taskSchedulerDef.getPropertyValues().add("poolSize", Runtime.getRuntime().availableProcessors()); taskSchedulerDef.getPropertyValues().add("threadNamePrefix", schedulerName + "-"); + taskSchedulerDef.getPropertyValues().add("removeOnCancelPolicy", true); parserContext.getRegistry().registerBeanDefinition(schedulerName, taskSchedulerDef); parserContext.registerComponent(new BeanComponentDefinition(taskSchedulerDef, schedulerName)); } diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/WebSocketConfigurationSupport.java b/spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/WebSocketConfigurationSupport.java index bd343628df..6fb4c75541 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/WebSocketConfigurationSupport.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/WebSocketConfigurationSupport.java @@ -19,8 +19,6 @@ package org.springframework.web.socket.config.annotation; import org.springframework.context.annotation.Bean; import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.web.servlet.HandlerMapping; -import org.springframework.web.servlet.handler.AbstractHandlerMapping; -import org.springframework.web.socket.sockjs.transport.SockJsThreadPoolTaskScheduler; /** * Configuration support for WebSocket request handling. @@ -34,8 +32,7 @@ public class WebSocketConfigurationSupport { public HandlerMapping webSocketHandlerMapping() { ServletWebSocketHandlerRegistry registry = new ServletWebSocketHandlerRegistry(defaultSockJsTaskScheduler()); registerWebSocketHandlers(registry); - AbstractHandlerMapping hm = registry.getHandlerMapping(); - return hm; + return registry.getHandlerMapping(); } protected void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { @@ -60,8 +57,10 @@ public class WebSocketConfigurationSupport { */ @Bean public ThreadPoolTaskScheduler defaultSockJsTaskScheduler() { - ThreadPoolTaskScheduler scheduler = new SockJsThreadPoolTaskScheduler(); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); scheduler.setThreadNamePrefix("SockJS-"); + scheduler.setPoolSize(Runtime.getRuntime().availableProcessors()); + scheduler.setRemoveOnCancelPolicy(true); return scheduler; } diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/WebSocketMessageBrokerConfigurationSupport.java b/spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/WebSocketMessageBrokerConfigurationSupport.java index 1d170b81f9..0af443a470 100644 --- a/spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/WebSocketMessageBrokerConfigurationSupport.java +++ b/spring-websocket/src/main/java/org/springframework/web/socket/config/annotation/WebSocketMessageBrokerConfigurationSupport.java @@ -26,7 +26,6 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; import org.springframework.web.servlet.HandlerMapping; import org.springframework.web.socket.WebSocketHandler; import org.springframework.web.socket.messaging.SubProtocolWebSocketHandler; -import org.springframework.web.socket.sockjs.transport.SockJsThreadPoolTaskScheduler; /** * Extends {@link AbstractMessageBrokerConfiguration} and adds configuration for @@ -96,8 +95,10 @@ public abstract class WebSocketMessageBrokerConfigurationSupport extends Abstrac */ @Bean public ThreadPoolTaskScheduler messageBrokerSockJsTaskScheduler() { - ThreadPoolTaskScheduler scheduler = new SockJsThreadPoolTaskScheduler(); + ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler(); scheduler.setThreadNamePrefix("MessageBrokerSockJS-"); + scheduler.setPoolSize(Runtime.getRuntime().availableProcessors()); + scheduler.setRemoveOnCancelPolicy(true); return scheduler; } diff --git a/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/SockJsThreadPoolTaskScheduler.java b/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/SockJsThreadPoolTaskScheduler.java deleted file mode 100644 index 7e81997b9c..0000000000 --- a/spring-websocket/src/main/java/org/springframework/web/socket/sockjs/transport/SockJsThreadPoolTaskScheduler.java +++ /dev/null @@ -1,66 +0,0 @@ -/* - * Copyright 2002-2014 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.web.socket.sockjs.transport; - -import org.springframework.lang.UsesJava7; -import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; -import org.springframework.util.ClassUtils; - -import java.util.concurrent.ExecutorService; -import java.util.concurrent.RejectedExecutionHandler; -import java.util.concurrent.ScheduledThreadPoolExecutor; -import java.util.concurrent.ThreadFactory; - -/** - * An extension of ThreadPoolTaskScheduler optimized for managing a large number - * of task, e.g. setting he pool size to the number of available processors and - * setting the setRemoveOnCancelPolicy property of - * {@link java.util.concurrent.ScheduledThreadPoolExecutor} available in JDK 1.7 - * or higher in order to avoid keeping cancelled tasks around. - * - * @author Rossen Stoyanchev - * @since 4.1 - */ -@SuppressWarnings("serial") -public class SockJsThreadPoolTaskScheduler extends ThreadPoolTaskScheduler { - - // Check for setRemoveOnCancelPolicy method - available on JDK 7 and higher - private static boolean hasRemoveOnCancelPolicyMethod = ClassUtils.hasMethod( - ScheduledThreadPoolExecutor.class, "setRemoveOnCancelPolicy", boolean.class); - - - public SockJsThreadPoolTaskScheduler() { - setThreadNamePrefix("SockJS-"); - setPoolSize(Runtime.getRuntime().availableProcessors()); - } - - - @Override - protected ExecutorService initializeExecutor(ThreadFactory factory, RejectedExecutionHandler handler) { - ExecutorService service = super.initializeExecutor(factory, handler); - configureRemoveOnCancelPolicy((ScheduledThreadPoolExecutor) service); - return service; - } - - @UsesJava7 // guard setting removeOnCancelPolicy (safe with 1.6 due to hasRemoveOnCancelPolicyMethod check) - private void configureRemoveOnCancelPolicy(ScheduledThreadPoolExecutor service) { - if (hasRemoveOnCancelPolicyMethod) { - service.setRemoveOnCancelPolicy(true); - } - } - -} diff --git a/spring-websocket/src/test/java/org/springframework/web/socket/messaging/SubProtocolWebSocketHandlerTests.java b/spring-websocket/src/test/java/org/springframework/web/socket/messaging/SubProtocolWebSocketHandlerTests.java index 8075197908..8f315b44b8 100644 --- a/spring-websocket/src/test/java/org/springframework/web/socket/messaging/SubProtocolWebSocketHandlerTests.java +++ b/spring-websocket/src/test/java/org/springframework/web/socket/messaging/SubProtocolWebSocketHandlerTests.java @@ -170,7 +170,7 @@ public class SubProtocolWebSocketHandlerTests { assertTrue(session1.isOpen()); assertFalse(session2.isOpen()); assertNull(session1.getCloseStatus()); - assertEquals(CloseStatus.PROTOCOL_ERROR, session2.getCloseStatus()); + assertEquals(CloseStatus.SESSION_NOT_RELIABLE, session2.getCloseStatus()); }