diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java index b94358e88b..90156b46b2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java @@ -38,6 +38,8 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements private PollableSource source; + private volatile boolean running; + public PollingSourceAdapter(PollableSource source) { Assert.notNull(source, "'source' must not be null"); @@ -45,6 +47,18 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements this.setConsumerPolicy(ConsumerPolicy.newPollingPolicy(DEFAULT_PERIOD)); } + public boolean isRunning() { + return this.running; + } + + public void start() { + this.running = true; + } + + public void stop() { + this.running = false; + } + public void setPeriod(int period) { Assert.isTrue(period > 0, "'period' must be a positive value"); this.getConsumerPolicy().setPeriod(period); @@ -56,6 +70,9 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements } public int dispatch() { + if (!this.isRunning()) { + return 0; + } int messagesProcessed = 0; int limit = this.getConsumerPolicy().getMaxMessagesPerTask(); Collection results = this.source.poll(limit); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java index 24e192f87c..c7a89944d0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java @@ -40,6 +40,10 @@ public abstract class AbstractMessageDispatcher implements MessageDispatcher { private List executors = new CopyOnWriteArrayList(); + private volatile boolean running; + + private Object lifecycleMonitor = new Object(); + public AbstractMessageDispatcher(MessageRetriever retriever) { this.retriever = retriever; @@ -56,6 +60,30 @@ public abstract class AbstractMessageDispatcher implements MessageDispatcher { return this.executors; } + public boolean isRunning() { + return this.running; + } + + public void start() { + synchronized (this.lifecycleMonitor) { + if (!this.isRunning()) { + for (MessageReceivingExecutor executor : this.executors) { + executor.start(); + } + } + } + } + + public void stop() { + synchronized (this.lifecycleMonitor) { + if (this.isRunning()) { + for (MessageReceivingExecutor executor : this.executors) { + executor.stop(); + } + } + } + } + /** * Retrieves messages and dispatches to the executors. * 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 9211b00db4..0901520605 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 @@ -58,12 +58,10 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif private Map> receivers = new ConcurrentHashMap>(); - private Map lifecycleComponents = new ConcurrentHashMap(); + private List dispatchers = new CopyOnWriteArrayList(); private List dispatcherTasks = new CopyOnWriteArrayList(); - private Map, MessageReceivingExecutor> receiverExecutors = new ConcurrentHashMap, MessageReceivingExecutor>(); - private ScheduledThreadPoolExecutor dispatcherExecutor; private int dispatcherPoolSize = 10; @@ -147,7 +145,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif this.activateSubscription(subscription); if (logger.isInfoEnabled()) { logger.info("activated subscription to channel '" + subscription.getChannel() + - "' for endpoint '" + subscription.getReceiver() + "'"); + "' for receiver '" + subscription.getReceiver() + "'"); } } } @@ -210,12 +208,10 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif ConsumerPolicy policy = adapter.getConsumerPolicy(); if (adapter instanceof MessageDispatcher) { MessageDispatcher dispatcher = (MessageDispatcher) adapter; + this.dispatchers.add(dispatcher); DispatcherTask dispatcherTask = new DispatcherTask(dispatcher, policy); this.addDispatcherTask(dispatcherTask); } - if (adapter instanceof Lifecycle) { - this.addLifecycleComponent(name, (Lifecycle) adapter); - } if (logger.isInfoEnabled()) { logger.info("registered source adapter '" + name + "'"); } @@ -264,8 +260,6 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif private void doActivate(MessageChannel channel, MessageReceiver receiver, ConsumerPolicy policy) { MessageReceivingExecutor executor = new MessageReceivingExecutor(receiver, policy.getConcurrency(), policy.getMaxConcurrency()); - this.receiverExecutors.put(receiver, executor); - this.addLifecycleComponent(receiver.getName() + "-executor", executor); MessageRetriever retriever = new ChannelPollingMessageRetriever(channel, policy); DefaultMessageDispatcher dispatcher = new DefaultMessageDispatcher(retriever); dispatcher.setRejectionLimit(policy.getRejectionLimit()); @@ -275,6 +269,7 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif if (this.isRunning()) { executor.start(); } + this.dispatchers.add(dispatcher); this.addDispatcherTask(dispatcherTask); if (this.logger.isInfoEnabled()) { logger.info("registered dispatcher task: channel='" + @@ -282,16 +277,6 @@ 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()) { @@ -302,17 +287,6 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif } } - public int getActiveCountForReceiver(String receiverName) { - MessageReceiver receiver = this.receivers.get(receiverName); - if (receiver != null) { - MessageReceivingExecutor executor = this.receiverExecutors.get(receiver); - if (executor != null) { - return executor.getActiveCount(); - } - } - return 0; - } - private void scheduleDispatcherTask(DispatcherTask task) { ConsumerPolicy policy = task.getPolicy(); if (policy.getPeriod() <= 0) { @@ -345,10 +319,10 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif synchronized (this.lifecycleMonitor) { if (!this.isRunning()) { this.running = true; - for (Map.Entry entry : this.lifecycleComponents.entrySet()) { - entry.getValue().start(); + for (MessageDispatcher dispatcher : this.dispatchers) { + dispatcher.start(); if (logger.isInfoEnabled()) { - logger.info("started lifecycle component '" + entry.getKey() + "'"); + logger.info("started dispatcher '" + dispatcher + "'"); } } for (DispatcherTask task : this.dispatcherTasks) { @@ -363,10 +337,10 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif if (this.isRunning()) { this.running = false; this.dispatcherExecutor.shutdownNow(); - for (Map.Entry entry : this.lifecycleComponents.entrySet()) { - entry.getValue().stop(); + for (MessageDispatcher dispatcher : this.dispatchers) { + dispatcher.stop(); if (logger.isInfoEnabled()) { - logger.info("stopped lifecycle component '" + entry.getKey() + "'"); + logger.info("stopped dispatcher '" + dispatcher + "'"); } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java index 7e070d6399..d5e6656471 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java @@ -16,12 +16,14 @@ package org.springframework.integration.bus; +import org.springframework.context.Lifecycle; + /** * Strategy interface for dispatching messages. * * @author Mark Fisher */ -public interface MessageDispatcher { +public interface MessageDispatcher extends Lifecycle { int dispatch(); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java index 20cf97fbc0..c3aad89bad 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java @@ -43,6 +43,7 @@ public class PollingSourceAdapterTests { PollingSourceAdapter adapter = new PollingSourceAdapter(source); adapter.setChannel(channel); adapter.setPeriod(100); + adapter.start(); adapter.dispatch(); Message message = channel.receive(); assertNotNull("message should not be null", message); @@ -57,6 +58,7 @@ public class PollingSourceAdapterTests { adapter.setChannel(channel); adapter.setPeriod(500); adapter.setSendTimeout(10); + adapter.start(); adapter.dispatch(); adapter.dispatch(); Message message1 = channel.receive(); @@ -78,6 +80,7 @@ public class PollingSourceAdapterTests { adapter.setChannel(channel); adapter.setPeriod(1000); adapter.setMaxMessagesPerTask(5); + adapter.start(); adapter.dispatch(); Message message1 = channel.receive(0); assertNotNull("message should not be null", message1); @@ -100,6 +103,7 @@ public class PollingSourceAdapterTests { adapter.setChannel(channel); adapter.setPeriod(1000); adapter.setMaxMessagesPerTask(2); + adapter.start(); adapter.dispatch(); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java index 6f8cf950a6..284e579948 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -139,5 +139,15 @@ public class MessageBusTests { public ConsumerPolicy getConsumerPolicy() { return ConsumerPolicy.newPollingPolicy(1000); } + + public boolean isRunning() { + return true; + } + + public void start() { + } + + public void stop() { + } } }