From 65ba407e842cf686cf3b61843b8ef57e14be2556 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Wed, 2 Jul 2008 02:37:13 +0000 Subject: [PATCH] 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. --- .../integration/bus/MessageBus.java | 25 +++++++++++-------- .../scheduling/MessagingTaskScheduler.java | 2 +- .../SimpleMessagingTaskScheduler.java | 24 ++++++++++++------ 3 files changed, 33 insertions(+), 18 deletions(-) diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java b/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java index 41ba925e86..428ff00671 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/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(); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java index 7ac632a370..d5974daad7 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/MessagingTaskScheduler.java +++ b/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. diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java b/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java index 86273fe466..91ddc053e8 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/scheduling/SimpleMessagingTaskScheduler.java +++ b/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) {