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

AbstractMessagingGateway now provides the TaskScheduler to its replyMessageCorrelator and manages its Lifecycle rather than delegating to MessageBus.

Browse Source
This commit is contained in:
Mark Fisher
2008-10-31 21:19:09 +00:00
parent b3b346e175
commit 18bc987e26
1 changed files with 38 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
org.springframework.integration/src/main/java/org/springframework/integration/gateway/AbstractMessagingGateway.java
View File

@@ -16,8 +16,7 @@
package org.springframework.integration.gateway;
import org.springframework.integration.bus.MessageBus;
import org.springframework.integration.bus.MessageBusAware;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.channel.SubscribableChannel;
@@ -33,6 +32,8 @@ import org.springframework.integration.endpoint.SubscribingConsumerEndpoint;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.MessageConsumer;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.scheduling.TaskScheduler;
import org.springframework.integration.scheduling.TaskSchedulerAware;
import org.springframework.util.Assert;
/**
@@ -43,7 +44,7 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public abstract class AbstractMessagingGateway implements MessagingGateway, MessageEndpoint, MessageBusAware {
public abstract class AbstractMessagingGateway implements MessagingGateway, MessageEndpoint, TaskSchedulerAware, Lifecycle {
private volatile MessageChannel requestChannel;
@@ -53,12 +54,16 @@ public abstract class AbstractMessagingGateway implements MessagingGateway, Mess
private volatile boolean shouldThrowErrors = true;
private volatile TaskScheduler taskScheduler;
private volatile MessageEndpoint replyMessageCorrelator;
private volatile MessageBus messageBus;
private final Object replyMessageCorrelatorMonitor = new Object();
private volatile boolean running;
private final Object lifecycleMonitor = new Object();
/**
* Set the request channel.
@@ -109,8 +114,8 @@ public abstract class AbstractMessagingGateway implements MessagingGateway, Mess
this.shouldThrowErrors = shouldThrowErrors;
}
public void setMessageBus(MessageBus messageBus) {
this.messageBus = messageBus;
public void setTaskScheduler(TaskScheduler taskScheduler) {
this.taskScheduler = taskScheduler;
}
public void send(Object object) {
@@ -172,7 +177,6 @@ public abstract class AbstractMessagingGateway implements MessagingGateway, Mess
if (this.replyMessageCorrelator != null) {
return;
}
Assert.state(this.messageBus != null, "No MessageBus available. Cannot register reply correlator.");
MessageEndpoint correlator = null;
MessageConsumer consumer = new AbstractReplyProducingMessageConsumer() {
@Override
@@ -188,13 +192,38 @@ public abstract class AbstractMessagingGateway implements MessagingGateway, Mess
PollingConsumerEndpoint endpoint = new PollingConsumerEndpoint(
consumer, (PollableChannel) this.replyChannel);
endpoint.afterPropertiesSet();
endpoint.setTaskScheduler(this.taskScheduler);
correlator = endpoint;
}
this.messageBus.registerEndpoint(correlator);
if (this.isRunning() && correlator instanceof Lifecycle) {
((Lifecycle) correlator).start();
}
this.replyMessageCorrelator = correlator;
}
}
public boolean isRunning() {
return this.running;
}
public void start() {
synchronized (this.lifecycleMonitor) {
if (this.replyMessageCorrelator != null && this.replyMessageCorrelator instanceof Lifecycle) {
((Lifecycle) this.replyMessageCorrelator).start();
}
this.running = true;
}
}
public void stop() {
synchronized (this.lifecycleMonitor) {
if (this.replyMessageCorrelator != null && this.replyMessageCorrelator instanceof Lifecycle) {
((Lifecycle) this.replyMessageCorrelator).stop();
}
this.running = false;
}
}
/**
* Subclasses must implement this to map from an Object to a Message.
*/