From 9fd30d1ed40f86474ab458815672d52e3d8d1f1b Mon Sep 17 00:00:00 2001
From: Mark Fisher <markfisher@vmware.com>
Date: Wed, 23 Apr 2008 13:37:27 +0000
Subject: [PATCH] Moved awareness of SynchronousChannel out of the MessageBus
 and into the SubscriptionManager.

---
 .../integration/bus/MessageBus.java           | 28 +++----------------
 .../integration/bus/SubscriptionManager.java  | 23 +++++++++++++++
 2 files changed, 27 insertions(+), 24 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 207f6f35c9..3ae43bfc49 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
@@ -42,7 +42,6 @@ import org.springframework.integration.channel.ChannelRegistryAware;
 import org.springframework.integration.channel.DefaultChannelRegistry;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.channel.QueueChannel;
-import org.springframework.integration.dispatcher.SynchronousChannel;
 import org.springframework.integration.endpoint.ConcurrencyPolicy;
 import org.springframework.integration.endpoint.DefaultEndpointRegistry;
 import org.springframework.integration.endpoint.EndpointRegistry;
@@ -50,7 +49,6 @@ import org.springframework.integration.endpoint.HandlerEndpoint;
 import org.springframework.integration.endpoint.MessageEndpoint;
 import org.springframework.integration.endpoint.TargetEndpoint;
 import org.springframework.integration.handler.MessageHandler;
-import org.springframework.integration.message.MessagingException;
 import org.springframework.integration.message.Target;
 import org.springframework.integration.scheduling.MessagePublishingErrorHandler;
 import org.springframework.integration.scheduling.MessagingTask;
@@ -58,7 +56,6 @@ import org.springframework.integration.scheduling.MessagingTaskScheduler;
 import org.springframework.integration.scheduling.Schedule;
 import org.springframework.integration.scheduling.SimpleMessagingTaskScheduler;
 import org.springframework.integration.scheduling.Subscription;
-import org.springframework.integration.util.ErrorHandler;
 import org.springframework.util.Assert;
 
 /**
@@ -356,7 +353,7 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 				targetEndpoint.setErrorHandler(new MessagePublishingErrorHandler(this.getErrorChannel()));
 			}
 		}
-		this.registerWithDispatcher(channel, endpoint, subscription.getSchedule());
+		this.activateSubscription(channel, endpoint, subscription.getSchedule());
 		if (logger.isInfoEnabled()) {
 			logger.info("activated subscription to channel '" + channel.getName() + 
 					"' for endpoint '" + endpoint + "'");
@@ -381,28 +378,11 @@ public class MessageBus implements ChannelRegistry, EndpointRegistry, Applicatio
 		}
 	}
 
-	private void registerWithDispatcher(MessageChannel channel, Target target, Schedule schedule) {
-		if (schedule == null && (channel instanceof SynchronousChannel)) {
-			((SynchronousChannel) channel).subscribe(target);
-			if (target instanceof Lifecycle) {
-				((Lifecycle) target).start();
-			}
-			if (target instanceof TargetEndpoint) {
-				((TargetEndpoint) target).setErrorHandler(new ErrorHandler() {
-					public void handle(Throwable t) {
-						if (t instanceof MessagingException) {
-							throw (MessagingException) t;
-						}
-						throw new MessagingException("error occurred in handler", t);
-					}
-				});
-			}
-			return;
-		}
-		SubscriptionManager manager = subscriptionManagers.get(channel);
+	private void activateSubscription(MessageChannel channel, Target target, Schedule schedule) {
+		SubscriptionManager manager = this.subscriptionManagers.get(channel);
 		if (manager == null) {
 			if (logger.isWarnEnabled()) {
-				logger.warn("no subscription manager available for channel '" + channel.getName() + "', be sure to register the channel");
+				logger.warn("no subscription manager available for channel '" + channel + "', be sure to register the channel");
 			}
 			return;
 		}
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/bus/SubscriptionManager.java b/spring-integration-core/src/main/java/org/springframework/integration/bus/SubscriptionManager.java
index ff3949f066..c6ef2cb4f8 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/bus/SubscriptionManager.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/bus/SubscriptionManager.java
@@ -29,10 +29,14 @@ import org.springframework.context.Lifecycle;
 import org.springframework.integration.ConfigurationException;
 import org.springframework.integration.channel.MessageChannel;
 import org.springframework.integration.dispatcher.PollingDispatcher;
+import org.springframework.integration.dispatcher.SynchronousChannel;
+import org.springframework.integration.endpoint.TargetEndpoint;
+import org.springframework.integration.message.MessagingException;
 import org.springframework.integration.message.Target;
 import org.springframework.integration.scheduling.MessagingTaskScheduler;
 import org.springframework.integration.scheduling.PollingSchedule;
 import org.springframework.integration.scheduling.Schedule;
+import org.springframework.integration.util.ErrorHandler;
 import org.springframework.util.Assert;
 
 /**
@@ -82,6 +86,11 @@ public class SubscriptionManager {
 		if (schedule == null) {
 			schedule = this.defaultSchedule;
 		}
+		else if (this.channel instanceof SynchronousChannel) {
+			if (logger.isInfoEnabled()) {
+				logger.info("Subscribing to a SynchronousChannel. The provided schedule will be ignored.");
+			}
+		}
 		else if (this.channel.getDispatcherPolicy().isPublishSubscribe()) {
 			if (logger.isInfoEnabled()) {
 				logger.info("This dispatcher broadcasts messages for a publish-subscribe channel. " +
@@ -96,6 +105,20 @@ public class SubscriptionManager {
 				((Lifecycle) target).start();
 			}
 		}
+		if (this.channel instanceof SynchronousChannel) {
+			((SynchronousChannel) this.channel).subscribe(target);
+			if (target instanceof TargetEndpoint) {
+				((TargetEndpoint) target).setErrorHandler(new ErrorHandler() {
+					public void handle(Throwable t) {
+						if (t instanceof MessagingException) {
+							throw (MessagingException) t;
+						}
+						throw new MessagingException("error occurred in handler", t);
+					}
+				});
+			}
+			return;
+		}
 		PollingDispatcher dispatcher = this.dispatchers.get(schedule);
 		if (dispatcher == null) {
 			dispatcher = this.dispatchers.putIfAbsent(schedule, new PollingDispatcher(this.channel, schedule));