Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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.

Browse Source
This commit is contained in:
Mark Fisher
2008-01-06 15:35:21 +00:00
parent a9f7b7dbdf
commit 680fe066b9
1 changed files with 15 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageBus.java
View File

@@ -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<String, Lifecycle> entry : this.lifecycleComponents.entrySet()) {
entry.getValue().stop();
if (logger.isInfoEnabled()) {
logger.info("stopped lifecycle component '" + entry.getKey() + "'");
}
}
this.dispatcherExecutor.shutdownNow();
}
}
}