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

Moved awareness of SynchronousChannel out of the MessageBus and into the SubscriptionManager.

Browse Source
This commit is contained in:
Mark Fisher
2008-04-23 13:37:27 +00:00
parent 94fa8675ca
commit 9fd30d1ed4
2 changed files with 27 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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;
}

23
spring-integration-core/src/main/java/org/springframework/integration/bus/SubscriptionManager.java
View File

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