From 680fe066b90a4eb06f296a0fb19400e1744cdaa0 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Sun, 6 Jan 2008 15:35:21 +0000 Subject: [PATCH] Fixed bug with receiver tasks not shutting down. MessageReceiverExecutor is now added to the lifecycleComponents. Therefore each executor has start() and stop() calls cascaded from the MessageBus itself. --- .../integration/bus/MessageBus.java | 24 ++++++++++++------- 1 file changed, 15 insertions(+), 9 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java index 1db01c914d..ea11c3ec2d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java @@ -187,13 +187,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif this.addDispatcherTask(dispatcherTask); } if (adapter instanceof Lifecycle) { - this.lifecycleComponents.put(name, (Lifecycle) adapter); - if (this.isRunning()) { - ((Lifecycle) adapter).start(); - if (logger.isInfoEnabled()) { - logger.info("started source adapter '" + name + "'"); - } - } + this.addLifecycleComponent(name, (Lifecycle) adapter); } if (logger.isInfoEnabled()) { logger.info("registered source adapter '" + name + "'"); @@ -209,7 +203,9 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif ConsumerPolicy policy = adapter.getConsumerPolicy(); MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); UnicastMessageDispatcher dispatcher = new UnicastMessageDispatcher(retriever, policy); - dispatcher.addExecutor(new MessageReceivingExecutor(adapter, policy.getConcurrency(), policy.getMaxConcurrency())); + MessageReceivingExecutor executor = new MessageReceivingExecutor(adapter, policy.getConcurrency(), policy.getMaxConcurrency()); + dispatcher.addExecutor(executor); + this.addLifecycleComponent(name + "-executor", executor); this.addDispatcherTask(new DispatcherTask(dispatcher, policy)); if (logger.isInfoEnabled()) { logger.info("registered target adapter '" + name + "'"); @@ -258,6 +254,16 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } } + private void addLifecycleComponent(String name, Lifecycle component) { + this.lifecycleComponents.put(name, component); + if (this.isRunning()) { + component.start(); + if (logger.isInfoEnabled()) { + logger.info("started lifecycle component '" + name + "'"); + } + } + } + private void addDispatcherTask(DispatcherTask dispatcherTask) { this.dispatcherTasks.add(dispatcherTask); if (this.isRunning()) { @@ -331,13 +337,13 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif synchronized (this.lifecycleMonitor) { if (this.isRunning()) { this.running = false; + this.dispatcherExecutor.shutdownNow(); for (Map.Entry entry : this.lifecycleComponents.entrySet()) { entry.getValue().stop(); if (logger.isInfoEnabled()) { logger.info("stopped lifecycle component '" + entry.getKey() + "'"); } } - this.dispatcherExecutor.shutdownNow(); } } }