From d109d80bd125e553bf502bc970c1ebfb32ca9bca Mon Sep 17 00:00:00 2001
From: Mark Fisher <markfisher@vmware.com>
Date: Fri, 28 Dec 2007 03:41:14 +0000
Subject: [PATCH] Added initial version of registration code for refactored
 source adapter.

---
 .../integration/bus/MessageBus.java           | 22 +++++++++++++++----
 1 file changed, 18 insertions(+), 4 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 6afe4df2fe..e33c09a12b 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
@@ -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");
 			}
 		}
 	}