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

Dispatchers implement Lifecycle - no longer a need for managing generic "lifecycleComponents" (first step in unification of MessageReceiver, MessageEndpoint, TargetAdapter, and MessageHandler).

Browse Source
This commit is contained in:
Mark Fisher
2008-01-10 17:12:58 +00:00
parent a68cffba62
commit d9d2fa4e9e
6 changed files with 72 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java
View File

@@ -38,6 +38,8 @@ public class PollingSourceAdapter<T> extends AbstractSourceAdapter<T> implements
private PollableSource<T> source;
private volatile boolean running;
public PollingSourceAdapter(PollableSource<T> source) {
Assert.notNull(source, "'source' must not be null");
@@ -45,6 +47,18 @@ public class PollingSourceAdapter<T> extends AbstractSourceAdapter<T> 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<T> extends AbstractSourceAdapter<T> implements
}
public int dispatch() {
if (!this.isRunning()) {
return 0;
}
int messagesProcessed = 0;
int limit = this.getConsumerPolicy().getMaxMessagesPerTask();
Collection<T> results = this.source.poll(limit);

28
spring-integration-core/src/main/java/org/springframework/integration/bus/AbstractMessageDispatcher.java
View File

@@ -40,6 +40,10 @@ public abstract class AbstractMessageDispatcher implements MessageDispatcher {
private List<MessageReceivingExecutor> executors = new CopyOnWriteArrayList<MessageReceivingExecutor>();
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.
*

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

@@ -58,12 +58,10 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
private Map<String, MessageReceiver<?>> receivers = new ConcurrentHashMap<String, MessageReceiver<?>>();
private Map<String, Lifecycle> lifecycleComponents = new ConcurrentHashMap<String, Lifecycle>();
private List<MessageDispatcher> dispatchers = new CopyOnWriteArrayList<MessageDispatcher>();
private List<DispatcherTask> dispatcherTasks = new CopyOnWriteArrayList<DispatcherTask>();
private Map<MessageReceiver<?>, MessageReceivingExecutor> receiverExecutors = new ConcurrentHashMap<MessageReceiver<?>, 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<String, Lifecycle> 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<String, Lifecycle> 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 + "'");
}
}
}

4
spring-integration-core/src/main/java/org/springframework/integration/bus/MessageDispatcher.java
View File

@@ -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();