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

Added initial version of registration code for refactored source adapter.

Browse Source
This commit is contained in:
Mark Fisher
2007-12-28 03:41:14 +00:00
parent a1ffb9f6c4
commit d109d80bd1
1 changed files with 18 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -30,6 +30,7 @@ import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.Lifecycle;
import org.springframework.integration.MessagingException;
import org.springframework.integration.adapter.SourceAdapter;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.channel.PointToPointChannel;
@@ -150,6 +151,14 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
}
}
public void registerSourceAdapter(SourceAdapter adapter) {
if (adapter instanceof MessageDispatcher) {
ConsumerPolicy policy = adapter.getConsumerPolicy();
DispatcherTask dispatcherTask = new DispatcherTask((MessageDispatcher) adapter, policy);
this.addDispatcherTask(dispatcherTask);
}
}
public void activateSubscription(Subscription subscription) {
String channelName = subscription.getChannel();
String endpointName = subscription.getEndpoint();
@@ -180,17 +189,22 @@ public class MessageBus implements ChannelRegistry, ApplicationContextAware, Lif
UnicastMessageDispatcher dispatcher = new UnicastMessageDispatcher(retriever, policy);
dispatcher.addEndpointExecutor(endpointExecutor);
DispatcherTask dispatcherTask = new DispatcherTask(dispatcher, policy);
this.dispatcherTasks.add(dispatcherTask);
if (this.isRunning()) {
endpointExecutor.start();
}
this.addDispatcherTask(dispatcherTask);
if (this.logger.isInfoEnabled()) {
logger.info("registered dispatcher task: channel='" +
channelName + "' endpoint='" + endpointName + "'");
}
}
private void addDispatcherTask(DispatcherTask dispatcherTask) {
this.dispatcherTasks.add(dispatcherTask);
if (this.isRunning()) {
endpointExecutor.start();
scheduleDispatcherTask(dispatcherTask);
if (this.logger.isInfoEnabled()) {
logger.info("scheduled dispatcher task: channel='" +
channelName + "' endpoint='" + endpointName + "'");
logger.info("scheduled dispatcher task");
}
}
}