diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
index e75393a723..bd744a8da3 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
@@ -41,347 +41,375 @@ import java.util.concurrent.*;
import java.util.concurrent.locks.ReentrantLock;
/**
- * MessageHandler that holds a buffer of correlated messages in a MessageStore. This class takes care of correlated
- * groups of messages that can be completed in batches. It is useful for aggregating, resequencing, or custom
+ * MessageHandler that holds a buffer of correlated messages in a MessageStore.
+ * This class takes care of correlated groups of messages that can be completed
+ * in batches. It is useful for aggregating, resequencing, or custom
* implementations requiring correlation.
*
- * To customize this handler inject {@link org.springframework.integration.aggregator.CorrelationStrategy},
+ * To customize this handler inject
+ * {@link org.springframework.integration.aggregator.CorrelationStrategy},
* {@link org.springframework.integration.aggregator.CompletionStrategy}, and
- * {@link org.springframework.integration.aggregator.MessageGroupProcessor} implementations as you require.
+ * {@link org.springframework.integration.aggregator.MessageGroupProcessor}
+ * implementations as you require.
*
- * By default the CorrelationStrategy will be a HeaderAttributeCorrelationStrategy and the CompletionStrategy will be a
- * SequenceSizeCompletionStrategy.
- *
+ * By default the CorrelationStrategy will be a HeaderAttributeCorrelationStrategy
+ * and the CompletionStrategy will be a SequenceSizeCompletionStrategy.
+ *
* @author Iwein Fuld
* @since 2.0
*/
public class CorrelatingMessageHandler extends AbstractMessageHandler implements MessageProducer, Lifecycle {
- private static final Log logger = LogFactory.getLog(CorrelatingMessageHandler.class);
+ private static final Log logger = LogFactory.getLog(CorrelatingMessageHandler.class);
- private static final long DEFAULT_SEND_TIMEOUT = 1000l;
- private static final long DEFAULT_REAPER_INTERVAL = 1000l;
- private static final long DEFAULT_TIMEOUT = 60000l;
+ private static final long DEFAULT_SEND_TIMEOUT = 1000L;
- private final MessageStore store;
- private final MessageGroupProcessor outputProcessor;
+ private static final long DEFAULT_REAPER_INTERVAL = 1000L;
- private volatile CorrelationStrategy correlationStrategy = new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID);
- private volatile CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy();
+ private static final long DEFAULT_TIMEOUT = 60000L;
- private MessageChannel outputChannel;
- private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate();
- private volatile MessageChannel discardChannel = new NullChannel();
- private ChannelResolver channelResolver;
+ private final MessageStore store;
- private final IdTracker tracker = new IdTracker();
- private final BlockingQueue keysInBuffer = new DelayQueue();
+ private final MessageGroupProcessor outputProcessor;
- private volatile ScheduledFuture reaperFutureTask;
- private volatile long reaperInterval = DEFAULT_REAPER_INTERVAL;
- private volatile long timeout = DEFAULT_TIMEOUT;
- private volatile boolean sendPartialResultOnTimeout;
+ private volatile CorrelationStrategy correlationStrategy =
+ new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID);
- private Object lifecycleMonitor = new Object();
+ private volatile CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy();
- public CorrelatingMessageHandler(MessageStore store,
- CorrelationStrategy correlationStrategy,
- CompletionStrategy completionStrategy,
- MessageGroupProcessor processor) {
- Assert.notNull(store);
- Assert.notNull(processor);
- Assert.notNull(correlationStrategy);
- Assert.notNull(completionStrategy);
- this.store = store;
- this.outputProcessor = processor;
- this.correlationStrategy = correlationStrategy;
- this.completionStrategy = completionStrategy;
- this.channelTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT);
- }
+ private MessageChannel outputChannel;
- public CorrelatingMessageHandler(MessageStore store,
- MessageGroupProcessor processor) {
- this(store, new HeaderAttributeCorrelationStrategy(
- MessageHeaders.CORRELATION_ID),
- new SequenceSizeCompletionStrategy(), processor);
- }
+ private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate();
- public CorrelatingMessageHandler(
- MessageGroupProcessor processor) {
- this(new SimpleMessageStore(100),
- new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID),
- new SequenceSizeCompletionStrategy(), processor);
- }
+ private volatile MessageChannel discardChannel = new NullChannel();
- public void setCorrelationStrategy(CorrelationStrategy correlationStrategy) {
- Assert.notNull(correlationStrategy);
- this.correlationStrategy = correlationStrategy;
- }
+ private volatile ChannelResolver channelResolver;
- public void setCompletionStrategy(CompletionStrategy completionStrategy) {
- Assert.notNull(completionStrategy);
- this.completionStrategy = completionStrategy;
- }
+ private final IdTracker tracker = new IdTracker();
- public void setTaskScheduler(TaskScheduler taskScheduler) {
- super.setTaskScheduler(taskScheduler);
- }
+ private final BlockingQueue keysInBuffer = new DelayQueue();
- public void setTimeout(long timeout) {
- this.timeout = timeout;
- }
+ private volatile ScheduledFuture reaperFutureTask;
- public void setReaperInterval(long reaperInterval) {
- this.reaperInterval = reaperInterval;
- }
+ private volatile long reaperInterval = DEFAULT_REAPER_INTERVAL;
+
+ private volatile long timeout = DEFAULT_TIMEOUT;
+
+ private volatile boolean sendPartialResultOnTimeout;
- public void setOutputChannel(MessageChannel outputChannel) {
- Assert.notNull(outputChannel, "'outputChannel' must not be null");
- this.outputChannel = outputChannel;
- }
+ private final Object lifecycleMonitor = new Object();
- public void setChannelResolver(ChannelResolver channelResolver) {
- this.channelResolver = channelResolver;
- }
- public void setDiscardChannel(MessageChannel discardChannel) {
- this.discardChannel = discardChannel;
- }
+ public CorrelatingMessageHandler(MessageStore store, CorrelationStrategy correlationStrategy,
+ CompletionStrategy completionStrategy, MessageGroupProcessor processor) {
+ Assert.notNull(store);
+ Assert.notNull(processor);
+ Assert.notNull(correlationStrategy);
+ Assert.notNull(completionStrategy);
+ this.store = store;
+ this.outputProcessor = processor;
+ this.correlationStrategy = correlationStrategy;
+ this.completionStrategy = completionStrategy;
+ this.channelTemplate.setSendTimeout(DEFAULT_SEND_TIMEOUT);
+ }
- public void setSendTimeout(long sendTimeout) {
- this.channelTemplate.setSendTimeout(sendTimeout);
- }
+ public CorrelatingMessageHandler(MessageStore store, MessageGroupProcessor processor) {
+ this(store, new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID),
+ new SequenceSizeCompletionStrategy(), processor);
+ }
- public void setSendPartialResultOnTimeout(boolean sendPartialResultOnTimeout) {
- this.sendPartialResultOnTimeout = sendPartialResultOnTimeout;
- }
+ public CorrelatingMessageHandler(MessageGroupProcessor processor) {
+ this(new SimpleMessageStore(100), new HeaderAttributeCorrelationStrategy(MessageHeaders.CORRELATION_ID),
+ new SequenceSizeCompletionStrategy(), processor);
+ }
- @Override
- protected void populateComponentMetadata(ComponentMetadata metadata) {
- metadata.setComponentType("aggregator");
+
+ public void setCorrelationStrategy(CorrelationStrategy correlationStrategy) {
+ Assert.notNull(correlationStrategy);
+ this.correlationStrategy = correlationStrategy;
+ }
+
+ public void setCompletionStrategy(CompletionStrategy completionStrategy) {
+ Assert.notNull(completionStrategy);
+ this.completionStrategy = completionStrategy;
+ }
+
+ public void setTaskScheduler(TaskScheduler taskScheduler) {
+ super.setTaskScheduler(taskScheduler);
+ }
+
+ public void setTimeout(long timeout) {
+ this.timeout = timeout;
+ }
+
+ public void setReaperInterval(long reaperInterval) {
+ this.reaperInterval = reaperInterval;
+ }
+
+ public void setOutputChannel(MessageChannel outputChannel) {
+ Assert.notNull(outputChannel, "'outputChannel' must not be null");
+ this.outputChannel = outputChannel;
+ }
+
+ public void setChannelResolver(ChannelResolver channelResolver) {
+ this.channelResolver = channelResolver;
+ }
+
+ public void setDiscardChannel(MessageChannel discardChannel) {
+ this.discardChannel = discardChannel;
+ }
+
+ public void setSendTimeout(long sendTimeout) {
+ this.channelTemplate.setSendTimeout(sendTimeout);
+ }
+
+ public void setSendPartialResultOnTimeout(boolean sendPartialResultOnTimeout) {
+ this.sendPartialResultOnTimeout = sendPartialResultOnTimeout;
}
@Override
- protected void handleMessageInternal(Message message) throws Exception {
- Object correlationKey = correlationStrategy.getCorrelationKey(message);
- if (logger.isDebugEnabled()) {
- logger.debug("Handling message with correllationKey [" + correlationKey + "]: " + message);
- }
- try {
- if (tracker.waitForLockIfNotTracked(correlationKey)) {
- MessageGroup group =
- new MessageGroup(store.list(correlationKey),
- completionStrategy, correlationKey, deleteOrTrackCallback());
- if (group.hasNoMessageSuperseding(message)) {
- store(message, correlationKey);
- group.add(message);
+ protected void populateComponentMetadata(ComponentMetadata metadata) {
+ metadata.setComponentType("aggregator");
+ }
- if (group.isComplete()) {
- if (logger.isDebugEnabled()) {
- logger.debug("Completing group with correllationKey [" + correlationKey + "]");
- }
- outputProcessor.processAndSend(group,
- channelTemplate, this.resolveReplyChannel(message, this.outputChannel, this.channelResolver));
- }
- } else {
- discardChannel.send(message);
- }
- } else {
- discardChannel.send(message);
- }
- } finally {
- tracker.unlock(correlationKey);
- }
- }
+ @Override
+ protected void handleMessageInternal(Message message) throws Exception {
+ Object correlationKey = correlationStrategy.getCorrelationKey(message);
+ if (logger.isDebugEnabled()) {
+ logger.debug("Handling message with correlationKey [" + correlationKey + "]: " + message);
+ }
+ try {
+ if (tracker.waitForLockIfNotTracked(correlationKey)) {
+ MessageGroup group = new MessageGroup(store.list(correlationKey),
+ completionStrategy, correlationKey, deleteOrTrackCallback());
- private MessageGroupListener deleteOrTrackCallback() {
- return new MessageGroupListener() {
- public void onProcessingOf(Message... processedMessage) {
- for (Message message : processedMessage) {
- store.delete(message.getHeaders().getId());
- }
- }
+ if (group.hasNoMessageSuperseding(message)) {
+ store(message, correlationKey);
+ group.add(message);
+ if (group.isComplete()) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Completing group with correlationKey [" + correlationKey + "]");
+ }
+ outputProcessor.processAndSend(group, channelTemplate,
+ this.resolveReplyChannel(message, this.outputChannel, this.channelResolver));
+ }
+ }
+ else {
+ discardChannel.send(message);
+ }
+ }
+ else {
+ discardChannel.send(message);
+ }
+ }
+ finally {
+ tracker.unlock(correlationKey);
+ }
+ }
- public void onCompletionOf(Object correlationKey) {
- tracker.pushCorrelationId(correlationKey);
- }
- };
- }
+ private MessageGroupListener deleteOrTrackCallback() {
+ return new MessageGroupListener() {
- @SuppressWarnings("unchecked")
- private void store(Message message, Object correlationKey) {
- Message toStore = message;
- if (!correlationKey.equals(message.getHeaders().getCorrelationId())) {
- toStore = MessageBuilder.fromMessage(message)
- .setCorrelationId(correlationKey).build();
- }
- store.put(toStore);
- if (!keysInBuffer.contains(correlationKey)) {
- keysInBuffer.add(new DelayedKey(correlationKey, timeout));
- }
- }
+ public void onProcessingOf(Message... processedMessage) {
+ for (Message message : processedMessage) {
+ store.delete(message.getHeaders().getId());
+ }
+ }
- public boolean isRunning() {
- synchronized (this.lifecycleMonitor) {
- return this.reaperFutureTask != null;
- }
- }
+ public void onCompletionOf(Object correlationKey) {
+ tracker.pushCorrelationId(correlationKey);
+ }
+ };
+ }
- public void start() {
- synchronized (this.lifecycleMonitor) {
- if (this.isRunning()) {
- return;
- }
- Assert.state(this.getTaskScheduler() != null, "'taskScheduler' must not be null");
- this.reaperFutureTask = this.getTaskScheduler().scheduleWithFixedDelay(
- new PrunerTask(), this.reaperInterval);
- }
- }
+ @SuppressWarnings("unchecked")
+ private void store(Message message, Object correlationKey) {
+ Message toStore = message;
+ if (!correlationKey.equals(message.getHeaders().getCorrelationId())) {
+ toStore = MessageBuilder.fromMessage(message).setCorrelationId(correlationKey).build();
+ }
+ store.put(toStore);
+ if (!keysInBuffer.contains(correlationKey)) {
+ keysInBuffer.add(new DelayedKey(correlationKey, timeout));
+ }
+ }
- public void stop() {
- synchronized (this.lifecycleMonitor) {
- if (this.isRunning()) {
- this.reaperFutureTask.cancel(true);
- }
- }
- }
+ public boolean isRunning() {
+ synchronized (this.lifecycleMonitor) {
+ return this.reaperFutureTask != null;
+ }
+ }
- private class PrunerTask implements Runnable {
- public void run() {
- if (logger.isTraceEnabled()) {
- logger.trace("PrunerTask is running");
- }
- DelayedKey delayedKey;
- try {
- while ((delayedKey = keysInBuffer.poll(reaperInterval, TimeUnit.MILLISECONDS)) != null) {
- Object key = delayedKey.getKey();
- if (logger.isDebugEnabled()) {
- logger.debug(this + "'s PrunerTask is processing " + key);
- }
- if (!forceComplete(key)) {
- keysInBuffer.offer(delayedKey);
- }
- }
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
+ public void start() {
+ synchronized (this.lifecycleMonitor) {
+ if (this.isRunning()) {
+ return;
+ }
+ Assert.state(this.getTaskScheduler() != null, "'taskScheduler' must not be null");
+ this.reaperFutureTask = this.getTaskScheduler().scheduleWithFixedDelay(
+ new PrunerTask(), this.reaperInterval);
+ }
+ }
- }
- }
-
- protected final boolean forceComplete(Object key) {
- try {
- if (tracker.tryLockFor(key)) {
- List> all = store.list(key);
- MessageGroup group = new MessageGroup(all, completionStrategy, key, deleteOrTrackCallback());
- if (all.size() > 0) {
- //last chance for normal completion
- MessageChannel outputChannel = resolveReplyChannel(all.get(0), this.outputChannel, this.channelResolver);
- boolean processed = false;
- if (group.isComplete()) {
- outputProcessor.processAndSend(group, channelTemplate, outputChannel);
- processed = true;
- }
- if (!processed) {
- if (sendPartialResultOnTimeout) {
- if (logger.isInfoEnabled()) {
- logger.info("Processing partially complete messages for key [" + key + "] to: " + outputChannel);
- }
- outputProcessor.processAndSend(group, channelTemplate, outputChannel);
- } else {
- if (logger.isInfoEnabled()) {
- logger.info("Discarding partially complete messages for key [" + key + "] to: " + discardChannel);
- }
- for (Message message : all) {
- discardChannel.send(message);
- store.delete(message.getHeaders().getId());
- }
- }
- }
- }
- return true;
- } else {
- return false;
- }
- } finally {
- tracker.unlock(key);
- }
- }
+ public void stop() {
+ synchronized (this.lifecycleMonitor) {
+ if (this.isRunning()) {
+ this.reaperFutureTask.cancel(true);
+ }
+ }
+ }
- private final class DelayedKey implements Delayed {
- private Object key;
- private Long releaseTime;
- private TimeUnit unit = TimeUnit.MILLISECONDS;
+ private class PrunerTask implements Runnable {
+ public void run() {
+ if (logger.isTraceEnabled()) {
+ logger.trace("PrunerTask is running");
+ }
+ DelayedKey delayedKey;
+ try {
+ while ((delayedKey = keysInBuffer.poll(reaperInterval, TimeUnit.MILLISECONDS)) != null) {
+ Object key = delayedKey.getKey();
+ if (logger.isDebugEnabled()) {
+ logger.debug(this + "'s PrunerTask is processing " + key);
+ }
+ if (!forceComplete(key)) {
+ keysInBuffer.offer(delayedKey);
+ }
+ }
+ }
+ catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
- public DelayedKey(Object correlationKey, long delay) {
- Assert.notNull(correlationKey, "'correlationKey' must not be null");
- this.key = correlationKey;
- this.releaseTime = System.currentTimeMillis() + delay;
- }
- public long getDelay(TimeUnit unit) {
- return unit.convert(this.releaseTime - System.currentTimeMillis(), this.unit);
- }
+ protected final boolean forceComplete(Object key) {
+ try {
+ if (tracker.tryLockFor(key)) {
+ List> all = store.list(key);
+ MessageGroup group = new MessageGroup(all, completionStrategy, key, deleteOrTrackCallback());
+ if (all.size() > 0) {
+ // last chance for normal completion
+ MessageChannel outputChannel = resolveReplyChannel(
+ all.get(0), this.outputChannel, this.channelResolver);
+ boolean processed = false;
+ if (group.isComplete()) {
+ outputProcessor.processAndSend(group, channelTemplate, outputChannel);
+ processed = true;
+ }
+ if (!processed) {
+ if (sendPartialResultOnTimeout) {
+ if (logger.isInfoEnabled()) {
+ logger.info("Processing partially complete messages for key [" +
+ key + "] to: " + outputChannel);
+ }
+ outputProcessor.processAndSend(group, channelTemplate, outputChannel);
+ }
+ else {
+ if (logger.isInfoEnabled()) {
+ logger.info("Discarding partially complete messages for key [" +
+ key + "] to: " + discardChannel);
+ }
+ for (Message message : all) {
+ discardChannel.send(message);
+ store.delete(message.getHeaders().getId());
+ }
+ }
+ }
+ }
+ return true;
+ }
+ else {
+ return false;
+ }
+ }
+ finally {
+ tracker.unlock(key);
+ }
+ }
- public int compareTo(Delayed o) {
- return ((Long) this.getDelay(this.unit)).compareTo(o.getDelay(this.unit));
- }
- public Object getKey() {
- return key;
- }
- }
+ private final class DelayedKey implements Delayed {
- private final class IdTracker {
- private ConcurrentMap trackerLocks = new ConcurrentHashMap();
- private final Queue