From 38d588aeb96de0be39c9c0ed34ac53c3b33cce7a Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 20 Apr 2010 21:46:16 +0000 Subject: [PATCH] formatting --- .../aggregator/CorrelatingMessageHandler.java | 606 +++++++++--------- 1 file changed, 317 insertions(+), 289 deletions(-) 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 trackedCorrelationIds = new LinkedBlockingQueue(); + private final Object key; - private void pushCorrelationId(Object correlationKey) { - while (!trackedCorrelationIds.offer(correlationKey)) { - //make room in the queue - trackedCorrelationIds.poll(); - } - trackerLocks.remove(correlationKey); - } + private final Long releaseTime; - /** - * Call this method to check if an id is tracked and obtain a lock for it. Don't forget to finally unlock - * afterwards. - * - * @return false if the key was tracked, true after obtaining the lock otherwise. - */ - private boolean waitForLockIfNotTracked(Object correlationKey) { - ReentrantLock lock = trackerLocks.get(correlationKey); - if (lock == null) { - if (trackedCorrelationIds.contains(correlationKey)) { - //this correlation key is already processed in the near past: disallow processing - return false; - } - lock = new ReentrantLock(); - ReentrantLock original = trackerLocks.putIfAbsent(correlationKey, lock); - lock = original == null ? lock : original; - } - lock.lock(); - return true; - } + private final TimeUnit unit = TimeUnit.MILLISECONDS; - private boolean tryLockFor(Object correlationKey) { - ReentrantLock lock = trackerLocks.get(correlationKey); - if (lock == null) { - lock = new ReentrantLock(); - ReentrantLock original = trackerLocks.putIfAbsent(correlationKey, lock); - lock = original == null ? lock : original; - } - return lock.tryLock(); - } + 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); + } + + public int compareTo(Delayed o) { + return ((Long) this.getDelay(this.unit)).compareTo(o.getDelay(this.unit)); + } + + public Object getKey() { + return key; + } + } + + + private final class IdTracker { + + private final ConcurrentMap trackerLocks = new ConcurrentHashMap(); + + private final Queue trackedCorrelationIds = new LinkedBlockingQueue(); + + private void pushCorrelationId(Object correlationKey) { + while (!trackedCorrelationIds.offer(correlationKey)) { + // make room in the queue + trackedCorrelationIds.poll(); + } + trackerLocks.remove(correlationKey); + } + + /** + * Call this method to check if an id is tracked and obtain a lock for + * it. Don't forget to finally unlock afterwards. + * @return false if the key was tracked, true after obtaining the lock otherwise + */ + private boolean waitForLockIfNotTracked(Object correlationKey) { + ReentrantLock lock = trackerLocks.get(correlationKey); + if (lock == null) { + if (trackedCorrelationIds.contains(correlationKey)) { + // this correlation key is already processed + // in the near past: disallow processing + return false; + } + lock = new ReentrantLock(); + ReentrantLock original = trackerLocks.putIfAbsent(correlationKey, lock); + lock = original == null ? lock : original; + } + lock.lock(); + return true; + } + + private boolean tryLockFor(Object correlationKey) { + ReentrantLock lock = trackerLocks.get(correlationKey); + if (lock == null) { + lock = new ReentrantLock(); + ReentrantLock original = trackerLocks.putIfAbsent(correlationKey, lock); + lock = original == null ? lock : original; + } + return lock.tryLock(); + } + + private void unlock(Object correlationKey) { + ReentrantLock lock = trackerLocks.get(correlationKey); + if (lock != null && lock.isHeldByCurrentThread()) { + lock.unlock(); + } + } + } - private void unlock(Object correlationKey) { - ReentrantLock lock = trackerLocks.get(correlationKey); - if (lock != null && lock.isHeldByCurrentThread()) { - lock.unlock(); - } - } - } }