Refactored MessageBus to accept a MessagingTaskScheduler instead of setting the ScheduledExecutorService directly. The SimpleMessagingTaskScheduler does not shutdown the executor until 'destroy' is called. The 'stop' call only prevents task submission until the next invocation of 'start' on the scheduler.

org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java

@@ -29,6 +29,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
@@ -76,8 +77,8 @@ import org.springframework.util.Assert;
  * @author Mark Fisher
  * @author Marius Bogoevici
  */
-public class MessageBus implements ChannelRegistry, EndpointRegistry, ApplicationContextAware, ApplicationListener,
-		Lifecycle {
+public class MessageBus implements ChannelRegistry, EndpointRegistry,
+		ApplicationContextAware, ApplicationListener, Lifecycle, DisposableBean {
 
 	public static final String ERROR_CHANNEL_NAME = "errorChannel";
 
@@ -99,8 +100,6 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 
 	private volatile MessagingTaskScheduler taskScheduler;
 
-	private volatile ScheduledExecutorService executor;
-
 	private volatile boolean configureAsyncEventMulticaster = false;
 
 	private volatile boolean autoCreateChannels = false;
@@ -138,10 +137,10 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 	}
 
 	/**
-	 * Set the {@link ScheduledExecutorService} to use for scheduling message dispatchers.
+	 * Set the {@link MessagingTaskScheduler} to use for scheduling message dispatchers.
 	 */
-	public void setScheduledExecutorService(ScheduledExecutorService executor) {
-		this.executor = executor;
+	public void setMessagingTaskScheduler(MessagingTaskScheduler messagingTaskScheduler) {
+		this.taskScheduler = messagingTaskScheduler;
 	}
 
 	/**
@@ -205,10 +204,10 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 				return;
 			}
 			this.initializing = true;
-			if (this.executor == null) {
-				this.executor = new ScheduledThreadPoolExecutor(DEFAULT_DISPATCHER_POOL_SIZE);
+			if (this.taskScheduler == null) {
+				this.taskScheduler = new SimpleMessagingTaskScheduler(
+						new ScheduledThreadPoolExecutor(DEFAULT_DISPATCHER_POOL_SIZE));
 			}
-			this.taskScheduler = new SimpleMessagingTaskScheduler(this.executor);
 			if (this.getErrorChannel() == null) {
 				this.setErrorChannel(new DefaultErrorChannel());
 			}
@@ -501,6 +500,12 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 		}
 	}
 
+	public void destroy() throws Exception {
+		if (this.taskScheduler instanceof DisposableBean) {
+			((DisposableBean) this.taskScheduler).destroy();
+		}
+	}
+
 	public void onApplicationEvent(ApplicationEvent event) {
 		if (event instanceof ContextRefreshedEvent) {
 			ApplicationContext context = ((ContextRefreshedEvent) event).getApplicationContext();

org.springframework.integration/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2007 the original author or authors.
+ * Copyright 2002-2008 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.

org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java

@@ -25,6 +25,7 @@ import java.util.concurrent.TimeUnit;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
+import org.springframework.beans.factory.DisposableBean;
 import org.springframework.integration.util.ErrorHandler;
 import org.springframework.util.Assert;
 
@@ -34,7 +35,7 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  */
-public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler {
+public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler implements DisposableBean {
 
 	private final Log logger = LogFactory.getLog(this.getClass());
 
@@ -91,17 +92,26 @@ public class SimpleMessagingTaskScheduler extends AbstractMessagingTaskScheduler
 	public void stop() {
 		synchronized (this.lifecycleMonitor) {
 			if (this.running) {
-				if (this.waitForTasksToCompleteOnShutdown) {
-					this.executor.shutdown();
-				}
-				else {
-					this.executor.shutdownNow();
-				}
 				this.running = false;
 			}
 		}
 	}
 
+	public void destroy() {
+		synchronized (this.lifecycleMonitor) {
+			this.stop();
+			if (this.executor.isShutdown()) {
+				return;
+			}
+			if (this.waitForTasksToCompleteOnShutdown) {
+				this.executor.shutdown();
+			}
+			else {
+				this.executor.shutdownNow();
+			}
+		}
+	}
+
 	@Override
 	public ScheduledFuture<?> schedule(Runnable task) {
 		if (!this.running) {