diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractMessageBarrier.java b/org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractMessageBarrier.java new file mode 100644 index 0000000000..485d387536 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractMessageBarrier.java @@ -0,0 +1,104 @@ +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.router; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.locks.ReentrantLock; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.integration.message.Message; + +/** + * Default implementation for a {@link MessageBarrier}. + * + * @author Marius Bogoevici + */ +public abstract class AbstractMessageBarrier implements MessageBarrier { + + private final Log logger = LogFactory.getLog(this.getClass()); + + protected final List> messages = new ArrayList>(); + + private volatile boolean complete = false; + + private final ReentrantLock lock = new ReentrantLock(); + + private final long timestamp = System.currentTimeMillis(); + + + /** + * Returns the creation time of this barrier as the number of milliseconds + * since January 1, 1970. + * @see System#currentTimeMillis() + */ + public long getTimestamp() { + return this.timestamp; + } + + protected boolean isComplete(){ + return this.complete; + } + + /** + * Adds a message to the aggregation group and releases if available. + * Otherwise, the return value will be null. + */ + public List> addAndRelease(Message message) { + try { + this.lock.lock(); + if (this.complete) { + if (logger.isDebugEnabled()) { + logger.debug("Message received after aggregation has already completed: " + message); + } + return null; + } + addMessage(message); + this.complete = hasReceivedAllMessages(); + return releaseAvailableMessages(); + } + finally { + this.lock.unlock(); + } + } + + protected void addMessage(Message message) { + this.messages.add(message); + } + + public List> getMessages() { + return Collections.unmodifiableList(this.messages); + } + + /** + * Subclasses will implement this method to indicate if all possible messages that could be received by + * a given barrier have already been received (e.g. all messages from a given sequence) + */ + protected abstract boolean hasReceivedAllMessages(); + + + /** + * Subclasses will implement this method to return the messages that can be released by this barrier, after + * the receipt of a given message. It might be possible that a number of messages are released before the barrier + * has ended its work (partial release) and this depends completely on the implementation of the barrier. + * However, once hasReceivedAllMessages() is deemed true, only one call to releaseAvailableMessages() shall + * yield results. + */ + protected abstract List> releaseAvailableMessages(); + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractMessageBarrierHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractMessageBarrierHandler.java new file mode 100644 index 0000000000..d9dbf0d2f2 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractMessageBarrierHandler.java @@ -0,0 +1,293 @@ +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.router; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.handler.MessageHandler; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageHandlingException; +import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; + +/** + * Base class for {@link MessageBarrier}-based MessageHandlers. + * A {@link MessageHandler} implementation that waits for a group of + * {@link Message Messages} to arrive and process them together. + * Uses a {@link MessageBarrier} to store messages and to decide on how + * the messages can be released. + *

+ * Each {@link Message} that is received by this handler will be associated with + * a group based upon the 'correlationId' property of its + * header. If no such property is available, a {@link MessageHandlingException} + * will be thrown. + *

+ * The 'timeout' value determines how long to wait for the + * complete group after the arrival of the first {@link Message} of the group. + * The default value is 1 minute. If the timeout elapses prior to completion, + * then Messages with that timed-out 'correlationId' will be sent to the + * 'discardChannel' if provided. + * + * @author Mark Fisher + * @author Marius Bogoevici + */ +public abstract class AbstractMessageBarrierHandler implements MessageHandler, InitializingBean { + + public final static long DEFAULT_SEND_TIMEOUT = 1000; + + public final static long DEFAULT_TIMEOUT = 60000; + + public final static long DEFAULT_REAPER_INTERVAL = 1000; + + public final static int DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY = 1000; + + protected final Log logger = LogFactory.getLog(this.getClass()); + + protected volatile MessageChannel defaultReplyChannel; + + private volatile MessageChannel discardChannel; + + protected volatile long sendTimeout = DEFAULT_SEND_TIMEOUT; + + protected final ConcurrentMap barriers + = new ConcurrentHashMap(); + + private volatile long timeout = DEFAULT_TIMEOUT; + + private volatile boolean sendPartialResultOnTimeout = false; + + private volatile long reaperInterval = DEFAULT_REAPER_INTERVAL; + + private volatile int trackedCorrelationIdCapacity = DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY; + + protected volatile BlockingQueue trackedCorrelationIds; + + protected final ScheduledExecutorService executor; + + private volatile boolean initialized; + + + public AbstractMessageBarrierHandler(ScheduledExecutorService executor) { + this.executor = (executor != null) ? executor : Executors.newSingleThreadScheduledExecutor(); + } + + + /** + * Set the default channel for sending aggregated Messages. Note that + * precedence will be given to the 'returnAddress' of the aggregated + * message itself, then to the 'returnAddress' of the original message. + */ + public void setDefaultReplyChannel(MessageChannel defaultReplyChannel) { + this.defaultReplyChannel = defaultReplyChannel; + } + + /** + * Specify a channel for sending Messages that arrive after their aggregation + * group has either completed or timed-out. + */ + public void setDiscardChannel(MessageChannel discardChannel) { + this.discardChannel = discardChannel; + } + + /** + * Set the timeout for sending aggregation results and discarded Messages. + */ + public void setSendTimeout(long sendTimeout) { + this.sendTimeout = sendTimeout; + } + + /** + * Specify whether to aggregate and send the resulting Message when the + * timeout elapses prior to the CompletionStrategy. + */ + public void setSendPartialResultOnTimeout(boolean sendPartialResultOnTimeout) { + this.sendPartialResultOnTimeout = sendPartialResultOnTimeout; + } + + /** + * Set the interval in milliseconds for the reaper thread. Default is 1000. + */ + public void setReaperInterval(long reaperInterval) { + Assert.isTrue(reaperInterval > 0, "'reaperInterval' must be a positive value"); + this.reaperInterval = reaperInterval; + } + + /** + * Set the number of completed correlationIds to track. Default is 1000. + */ + public void setTrackedCorrelationIdCapacity(int trackedCorrelationIdCapacity) { + Assert.isTrue(trackedCorrelationIdCapacity > 0, "'trackedCorrelationIdCapacity' must be a positive value"); + this.trackedCorrelationIdCapacity = trackedCorrelationIdCapacity; + } + + /** + * Initialize this handler. + */ + public void afterPropertiesSet() { + this.trackedCorrelationIds = new ArrayBlockingQueue(this.trackedCorrelationIdCapacity); + this.executor.scheduleWithFixedDelay(new ReaperTask(), + this.reaperInterval, this.reaperInterval, TimeUnit.MILLISECONDS); + this.initialized = true; + } + + /** + * Maximum time to wait (in milliseconds) for the completion strategy to + * become true. The default is 60000 (1 minute). + */ + public void setTimeout(long timeout) { + Assert.isTrue(timeout >= 0, "'timeout' must not be negative"); + this.timeout = timeout; + } + + public Message handle(Message message) { + if (!this.initialized) { + this.afterPropertiesSet(); + } + Object correlationId = message.getHeader().getCorrelationId(); + if (correlationId == null) { + throw new MessageHandlingException(message, + this.getClass().getSimpleName() + " requires the 'correlationId' property"); + } + if (this.trackedCorrelationIds.contains(correlationId)) { + if (logger.isDebugEnabled()) { + logger.debug("Handling for correlationId '" + correlationId + + "' has already completed or timed out."); + } + this.sendToDiscardChannelIfAvailable(message); + return null; + } + MessageBarrier barrier = barriers.putIfAbsent(correlationId, createMessageBarrier()); + if (barrier == null) { + barrier = barriers.get(correlationId); + } + List> releasedMessages = barrier.addAndRelease(message); + if (CollectionUtils.isEmpty(releasedMessages)) { + return null; + } + if (isBarrierRemovable(correlationId, releasedMessages)) { + this.removeBarrier(correlationId); + } + afterRelease(correlationId, releasedMessages); + return null; + } + + private void afterRelease(Object correlationId, List> releasedMessages) { + Message[] processedMessages = this.processReleasedMessages(correlationId, releasedMessages); + for (Message result : processedMessages) { + MessageChannel replyChannel = this.resolveReplyChannelFromMessage(result); + if (replyChannel == null) { + replyChannel = this.resolveReplyChannelFromMessage(releasedMessages.get(0)); + if (replyChannel == null) { + replyChannel = this.defaultReplyChannel; + } + } + if (replyChannel != null) { + replyChannel.send(result, this.sendTimeout); + } + else if (logger.isWarnEnabled()) { + logger.warn("unable to determine reply channel for aggregation result: " + result); + } + } + } + + private void sendToDiscardChannelIfAvailable(Message message) { + if (this.discardChannel != null) { + if (!this.discardChannel.send(message, this.sendTimeout)) { + if (logger.isWarnEnabled()) { + logger.warn("unable to send to 'discardChannel', message: " + message); + } + } + } + } + + protected MessageChannel resolveReplyChannelFromMessage(Message message) { + Object returnAddress = message.getHeader().getReturnAddress(); + if (returnAddress != null) { + if (returnAddress instanceof MessageChannel) { + return (MessageChannel) returnAddress; + } + if (logger.isWarnEnabled()) { + logger.warn("Aggregator can only reply to a 'returnAddress' of type MessageChannel."); + } + } + return null; + } + + private void removeBarrier(Object correlationId) { + if (this.barriers.remove(correlationId) != null) { + synchronized (this.trackedCorrelationIds) { + boolean added = this.trackedCorrelationIds.offer(correlationId); + if (!added) { + this.trackedCorrelationIds.poll(); + this.trackedCorrelationIds.offer(correlationId); + } + } + } + } + + private class ReaperTask implements Runnable { + + public void run() { + long currentTime = System.currentTimeMillis(); + for (Map.Entry entry : barriers.entrySet()) { + if (currentTime - entry.getValue().getTimestamp() >= timeout) { + Object correlationId = entry.getKey(); + List> messages = entry.getValue().getMessages(); + removeBarrier(correlationId); + if (sendPartialResultOnTimeout) { + afterRelease(correlationId, messages); + } + else { + for (Message message : messages) { + sendToDiscardChannelIfAvailable(message); + } + } + } + } + } + } + + /** + * Factory method for creating a suitable MessageBarrier + * implementation. + */ + protected abstract MessageBarrier createMessageBarrier(); + + /** + * Implements the logic for deciding whether, based on what the + * MessageBarrier has released so far, work for the correlationId + * can be considered done and the barrier can be released. + */ + protected abstract boolean isBarrierRemovable(Object correlationId, List> releasedMessages); + + /** + * Implements the logic for transforming the released Messages. + */ + protected abstract Message[] processReleasedMessages(Object correlationId, List> messages); + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java index 0d37420f1e..2036fcc231 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java @@ -17,88 +17,34 @@ package org.springframework.integration.router; import java.util.List; -import java.util.Map; -import java.util.concurrent.ArrayBlockingQueue; -import java.util.concurrent.BlockingQueue; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.beans.factory.InitializingBean; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.handler.MessageHandler; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageHandlingException; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; /** - * A {@link MessageHandler} implementation that waits for a complete + * An {@link AbstractMessageBarrierHandler} that waits for a complete * group of {@link Message Messages} to arrive and then delegates to an * {@link Aggregator} to combine them into a single {@link Message}. *

- * Each {@link Message} that is received by this handler will be associated with - * a group based upon the 'correlationId' property of its - * header. If no such property is available, a {@link MessageHandlingException} - * will be thrown. - *

* The default strategy for determining whether a group is complete is based on * the 'sequenceSize' property of the header. Alternatively, a * custom implementation of the {@link CompletionStrategy} may be provided. *

- * The 'timeout' value determines how long to wait for the - * complete group after the arrival of the first {@link Message} of the group. - * The default value is 1 minute. If the timeout elapses prior to completion, - * then Messages with that timed-out 'correlationId' will be sent to the - * 'discardChannel' if provided. + * All considerations regarding timeout and grouping by + * 'correlationId' from {@link AbstractMessageBarrierHandler} apply + * here as well. * * @author Mark Fisher * @author Marius Bogoevici */ -public class AggregatingMessageHandler implements MessageHandler, InitializingBean { - - public final static long DEFAULT_SEND_TIMEOUT = 1000; - - public final static long DEFAULT_TIMEOUT = 60000; - - public final static long DEFAULT_REAPER_INTERVAL = 1000; - - public final static int DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY = 1000; - - - private final Log logger = LogFactory.getLog(this.getClass()); +public class AggregatingMessageHandler extends AbstractMessageBarrierHandler { private final Aggregator aggregator; - private volatile MessageChannel defaultReplyChannel; - - private volatile MessageChannel discardChannel; - - private volatile long sendTimeout = DEFAULT_SEND_TIMEOUT; - private volatile CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy(); - private final ConcurrentMap barriers = new ConcurrentHashMap(); - - private volatile long timeout = DEFAULT_TIMEOUT; - - private volatile boolean sendPartialResultOnTimeout = false; - - private volatile long reaperInterval = DEFAULT_REAPER_INTERVAL; - - private volatile int trackedCorrelationIdCapacity = DEFAULT_TRACKED_CORRRELATION_ID_CAPACITY; - - private volatile BlockingQueue trackedCorrelationIds; - - private final ScheduledExecutorService executor; - - private volatile boolean initialized; - /** * Create a handler that delegates to the provided aggregator to combine a @@ -107,74 +53,16 @@ public class AggregatingMessageHandler implements MessageHandler, InitializingBe * single-threaded executor will be created. */ public AggregatingMessageHandler(Aggregator aggregator, ScheduledExecutorService executor) { + super(executor); Assert.notNull(aggregator, "'aggregator' must not be null"); this.aggregator = aggregator; - this.executor = (executor != null) ? executor : Executors.newSingleThreadScheduledExecutor(); } public AggregatingMessageHandler(Aggregator aggregator) { this(aggregator, null); } - - /** - * Set the default channel for sending aggregated Messages. Note that - * precedence will be given to the 'returnAddress' of the aggregated - * message itself, then to the 'returnAddress' of the original message. - */ - public void setDefaultReplyChannel(MessageChannel defaultReplyChannel) { - this.defaultReplyChannel = defaultReplyChannel; - } - - /** - * Specify a channel for sending Messages that arrive after their aggregation - * group has either completed or timed-out. - */ - public void setDiscardChannel(MessageChannel discardChannel) { - this.discardChannel = discardChannel; - } - - /** - * Set the timeout for sending aggregation results and discarded Messages. - */ - public void setSendTimeout(long sendTimeout) { - this.sendTimeout = sendTimeout; - } - - /** - * Specify whether to aggregate and send the resulting Message when the - * timeout elapses prior to the CompletionStrategy. - */ - public void setSendPartialResultOnTimeout(boolean sendPartialResultOnTimeout) { - this.sendPartialResultOnTimeout = sendPartialResultOnTimeout; - } - - /** - * Set the interval in milliseconds for the reaper thread. Default is 1000. - */ - public void setReaperInterval(long reaperInterval) { - Assert.isTrue(reaperInterval > 0, "'reaperInterval' must be a positive value"); - this.reaperInterval = reaperInterval; - } - - /** - * Set the number of completed correlationIds to track. Default is 1000. - */ - public void setTrackedCorrelationIdCapacity(int trackedCorrelationIdCapacity) { - Assert.isTrue(trackedCorrelationIdCapacity > 0, "'trackedCorrelationIdCapacity' must be a positive value"); - this.trackedCorrelationIdCapacity = trackedCorrelationIdCapacity; - } - - /** - * Initialize this handler. - */ - public void afterPropertiesSet() { - this.trackedCorrelationIds = new ArrayBlockingQueue(this.trackedCorrelationIdCapacity); - this.executor.scheduleWithFixedDelay(new ReaperTask(), - this.reaperInterval, this.reaperInterval, TimeUnit.MILLISECONDS); - this.initialized = true; - } - + /** * Strategy to determine whether the group of messages is complete. */ @@ -183,128 +71,24 @@ public class AggregatingMessageHandler implements MessageHandler, InitializingBe this.completionStrategy = completionStrategy; } - /** - * Maximum time to wait (in milliseconds) for the completion strategy to - * become true. The default is 60000 (1 minute). - */ - public void setTimeout(long timeout) { - Assert.isTrue(timeout >= 0, "'timeout' must not be negative"); - this.timeout = timeout; + protected MessageBarrier createMessageBarrier() { + return new AggregationBarrier(this.completionStrategy); } - public Message handle(Message message) { - if (!this.initialized) { - this.afterPropertiesSet(); - } - Object correlationId = message.getHeader().getCorrelationId(); - if (correlationId == null) { - throw new MessageHandlingException(message, - this.getClass().getSimpleName() + " requires the 'correlationId' property"); - } - if (this.trackedCorrelationIds.contains(correlationId)) { - if (logger.isDebugEnabled()) { - logger.debug("Aggregation for correlationId '" + correlationId + - "' has already completed or timed out."); - } - this.sendToDiscardChannelIfAvailable(message); - return null; - } - AggregationBarrier barrier = barriers.putIfAbsent(correlationId, - new AggregationBarrier(this.completionStrategy)); - if (barrier == null) { - barrier = barriers.get(correlationId); - } - List> releasedMessages = barrier.addAndRelease(message); - if (CollectionUtils.isEmpty(releasedMessages)) { - return null; - } - this.removeBarrier(correlationId); - this.aggregationCompleted(correlationId, releasedMessages); - return null; + protected boolean isBarrierRemovable(Object correlationId, List> releasedMessages) { + return releasedMessages != null && releasedMessages.size() > 0; } - - private void sendToDiscardChannelIfAvailable(Message message) { - if (this.discardChannel != null) { - if (!this.discardChannel.send(message, this.sendTimeout)) { - if (logger.isWarnEnabled()) { - logger.warn("unable to send to 'discardChannel', message: " + message); - } - } - } - } - - private void aggregationCompleted(Object correlationId, List> messages) { + + protected Message[] processReleasedMessages(Object correlationId, List> messages) { if (CollectionUtils.isEmpty(messages)) { if (logger.isDebugEnabled()) { logger.debug("no messages to aggregate"); } - return; + return new Message[0]; } Message result = aggregator.aggregate(messages); - if (result.getHeader().getCorrelationId() == null) { - result.getHeader().setCorrelationId(correlationId); - } - MessageChannel replyChannel = this.resolveReplyChannelFromMessage(result); - if (replyChannel == null) { - replyChannel = this.resolveReplyChannelFromMessage(messages.get(0)); - if (replyChannel == null) { - replyChannel = this.defaultReplyChannel; - } - } - if (replyChannel != null) { - replyChannel.send(result, this.sendTimeout); - } - else if (logger.isWarnEnabled()) { - logger.warn("unable to determine reply channel for aggregation result: " + result); - } - } - - private void removeBarrier(Object correlationId) { - if (this.barriers.remove(correlationId) != null) { - synchronized (this.trackedCorrelationIds) { - boolean added = this.trackedCorrelationIds.offer(correlationId); - if (!added) { - this.trackedCorrelationIds.poll(); - this.trackedCorrelationIds.offer(correlationId); - } - } - } - } - - private MessageChannel resolveReplyChannelFromMessage(Message message) { - Object returnAddress = message.getHeader().getReturnAddress(); - if (returnAddress != null) { - if (returnAddress instanceof MessageChannel) { - return (MessageChannel) returnAddress; - } - if (logger.isWarnEnabled()) { - logger.warn("Aggregator can only reply to a 'returnAddress' of type MessageChannel."); - } - } - return null; - } - - - private class ReaperTask implements Runnable { - - public void run() { - long currentTime = System.currentTimeMillis(); - for (Map.Entry entry : barriers.entrySet()) { - if (currentTime - entry.getValue().getTimestamp() >= timeout) { - Object correlationId = entry.getKey(); - List> messages = entry.getValue().getMessages(); - removeBarrier(correlationId); - if (sendPartialResultOnTimeout) { - aggregationCompleted(correlationId, messages); - } - else { - for (Message message : messages) { - sendToDiscardChannelIfAvailable(message); - } - } - } - } - } + return new Message[] { result }; + } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/router/AggregationBarrier.java b/org.springframework.integration/src/main/java/org/springframework/integration/router/AggregationBarrier.java index 432e1f9095..4d718fafbe 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/router/AggregationBarrier.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/router/AggregationBarrier.java @@ -17,14 +17,8 @@ package org.springframework.integration.router; import java.util.List; -import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.locks.ReentrantLock; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; import org.springframework.integration.message.Message; -import org.springframework.util.Assert; /** * MessageBarrier implementation for message aggregation. Delegates to a @@ -34,63 +28,21 @@ import org.springframework.util.Assert; * @author Marius Bogoevici * @author Mark Fisher */ -public class AggregationBarrier implements MessageBarrier { +public class AggregationBarrier extends AbstractMessageBarrier { - private final Log logger = LogFactory.getLog(this.getClass()); - - private final List> messages = new CopyOnWriteArrayList>(); - - private final CompletionStrategy completionStrategy; - - private volatile boolean complete = false; - - private final ReentrantLock lock = new ReentrantLock(); - - private final long timestamp = System.currentTimeMillis(); + protected final CompletionStrategy completionStrategy; - /** - * Create an AggregationBarrier with the given {@link CompletionStrategy}. - */ public AggregationBarrier(CompletionStrategy completionStrategy) { - Assert.notNull(completionStrategy, "'completionStrategy' must not be null"); this.completionStrategy = completionStrategy; } - /** - * Returns the creation time of this barrier as the number of milliseconds - * since January 1, 1970. - * @see java.lang.System#currentTimeMillis() - */ - public long getTimestamp() { - return this.timestamp; + protected List> releaseAvailableMessages() { + return (this.isComplete()) ? this.getMessages() : null; } - /** - * Adds a message to the aggregation group and releases if complete. - * Otherwise, the return value will be null. - */ - public List> addAndRelease(Message message) { - try { - this.lock.lock(); - if (this.complete) { - if (logger.isDebugEnabled()) { - logger.debug("Message received after aggregation has already completed: " + message); - } - return null; - } - this.messages.add(message); - this.complete = completionStrategy.isComplete(this.messages); - return (this.complete) ? this.messages : null; - } - finally { - this.lock.unlock(); - } + protected boolean hasReceivedAllMessages() { + return completionStrategy.isComplete(this.messages); } - - public List> getMessages() { - return this.messages; - } - } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/router/MessageBarrier.java b/org.springframework.integration/src/main/java/org/springframework/integration/router/MessageBarrier.java index aec642986c..b075c1e27a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/router/MessageBarrier.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/router/MessageBarrier.java @@ -26,9 +26,14 @@ import org.springframework.integration.message.Message; * {@link Message} arrives. * * @author Mark Fisher + * @author Marius Bogoevici */ public interface MessageBarrier { List> addAndRelease(Message message); + long getTimestamp(); + + List> getMessages(); + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/router/ResequencingMessageBarrier.java b/org.springframework.integration/src/main/java/org/springframework/integration/router/ResequencingMessageBarrier.java new file mode 100644 index 0000000000..1e71db96df --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/router/ResequencingMessageBarrier.java @@ -0,0 +1,97 @@ +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.router; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.Iterator; +import java.util.List; + +import org.springframework.integration.message.Message; + +/** + * MessageBarrier implementation for resequencing. It can either + * release partial sequences as messages arrive, or the full sequence. + * @author Marius Bogoevici + */ +public class ResequencingMessageBarrier extends AbstractMessageBarrier { + + private int lastReleasedSequenceNumber; + + private final Comparator> resequencingComparator; + + private final boolean releasePartialSequences; + + + /** + * @param releasePartialSequences specifies whether partial sequences should + * be released as they arrive, or the resequencer + */ + public ResequencingMessageBarrier(boolean releasePartialSequences) { + this.resequencingComparator = new Comparator>() { + public int compare(Message m1, Message m2) { + return m1.getHeader().getSequenceNumber() - m2.getHeader().getSequenceNumber(); + } + }; + this.releasePartialSequences = releasePartialSequences; + this.lastReleasedSequenceNumber = 0; + } + + + protected void addMessage(Message message) { + int insertionPoint = Collections.binarySearch(messages, message, resequencingComparator); + if (insertionPoint < 0) { + insertionPoint = -(insertionPoint + 1); + this.messages.add(insertionPoint, message); + } + } + + protected boolean hasReceivedAllMessages() { + //verify that there is a contiguous sequence of messages from the first to the last + //and the last message is last in sequence, and the first message is the next one to be delivered + //(aggregated, this means that the last possibile partial sequence of messages has been received + Message firstMessage = this.messages.get(0); + Message lastMessage = this.messages.get(messages.size() - 1); + return (lastMessage.getHeader().getSequenceNumber() == lastMessage.getHeader().getSequenceSize() + && (lastMessage.getHeader().getSequenceNumber() - firstMessage.getHeader().getSequenceNumber() + == this.messages.size() - 1 + && this.lastReleasedSequenceNumber == firstMessage.getHeader().getSequenceNumber() - 1)); + } + + protected List> releaseAvailableMessages() { + if (this.releasePartialSequences || hasReceivedAllMessages()) { + ArrayList> releasedMessages = new ArrayList>(); + Iterator> it = this.messages.iterator(); + while (it.hasNext()) { + Message currentMessage = it.next(); + if (this.lastReleasedSequenceNumber == currentMessage.getHeader().getSequenceNumber() - 1) { + releasedMessages.add(currentMessage); + this.lastReleasedSequenceNumber = currentMessage.getHeader().getSequenceNumber(); + it.remove(); + } + else { + break; + } + } + return releasedMessages; + } + else { + return new ArrayList>(); + } + } + +} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/router/ResequencingMessageHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/router/ResequencingMessageHandler.java new file mode 100644 index 0000000000..4e384f8bf4 --- /dev/null +++ b/org.springframework.integration/src/main/java/org/springframework/integration/router/ResequencingMessageHandler.java @@ -0,0 +1,64 @@ +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.router; + +import java.util.List; +import java.util.concurrent.ScheduledExecutorService; + +import org.springframework.integration.message.Message; + + +/** + * An {@link AbstractMessageBarrierHandler} that waits for a group of + * {@link Message Messages} to arrive and re-sends them in order, sorted + * by their sequenceNumber. + *

+ * This handler can either release partial sequences of messages or can + * wait for the whole sequence to arrive before re-sending them. + *

+ * All considerations regarding timeout and grouping by + * 'correlationId' from {@link AbstractMessageBarrierHandler} apply + * here as well. + * + * @author Marius Bogoevici + */ +public class ResequencingMessageHandler extends AbstractMessageBarrierHandler{ + + private boolean releasePartialSequences; + + public ResequencingMessageHandler(boolean releasePartialSequences) { + this(null, releasePartialSequences); + } + + public ResequencingMessageHandler(ScheduledExecutorService executor, boolean releasePartialSequences) { + super(executor); + this.releasePartialSequences = releasePartialSequences; + } + + protected MessageBarrier createMessageBarrier() { + return new ResequencingMessageBarrier(releasePartialSequences); + } + + protected Message[] processReleasedMessages(Object correlationId, List> messages) { + return messages.toArray(new Message[messages.size()]); + } + + protected boolean isBarrierRemovable(Object correlationId, List> releasedMessages) { + return (releasedMessages.get(releasedMessages.size() - 1).getHeader().getSequenceNumber() == + releasedMessages.get(releasedMessages.size() - 1).getHeader().getSequenceSize()); + } +} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/router/ResequencerMessageHandlerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/router/ResequencerMessageHandlerTests.java new file mode 100644 index 0000000000..cd945968f0 --- /dev/null +++ b/org.springframework.integration/src/test/java/org/springframework/integration/router/ResequencerMessageHandlerTests.java @@ -0,0 +1,143 @@ +/* + * Copyright 2002-2008 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.router; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import org.junit.Test; + +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.StringMessage; + +/** + * @author Marius Bogoevici + */ +public class ResequencerMessageHandlerTests { + + @Test + public void testBasicResequencing() throws InterruptedException { + ResequencingMessageHandler aggregator = new ResequencingMessageHandler(false); + QueueChannel replyChannel = new QueueChannel(); + Message message1 = createMessage("123", "ABC", 3, 3, replyChannel); + Message message2 = createMessage("456", "ABC", 3, 1, replyChannel); + Message message3 = createMessage("789", "ABC", 3, 2, replyChannel); + CountDownLatch latch = new CountDownLatch(3); + aggregator.handle(message1); + aggregator.handle(message3); + aggregator.handle(message2); + latch.await(1000, TimeUnit.MILLISECONDS); + Message reply1 = replyChannel.receive(500); + Message reply2 = replyChannel.receive(500); + Message reply3 = replyChannel.receive(500); + assertNotNull(reply1); + assertEquals(1, reply1.getHeader().getSequenceNumber()); + assertNotNull(reply2); + assertEquals(2, reply2.getHeader().getSequenceNumber()); + assertNotNull(reply3); + assertEquals(3, reply3.getHeader().getSequenceNumber()); + } + + @Test + public void testResequencingWithIncompleteSequenceRelease() throws InterruptedException { + ResequencingMessageHandler aggregator = new ResequencingMessageHandler(true); + QueueChannel replyChannel = new QueueChannel(); + Message message1 = createMessage("123", "ABC", 4, 2, replyChannel); + Message message2 = createMessage("456", "ABC", 4, 1, replyChannel); + Message message3 = createMessage("789", "ABC", 4, 4, replyChannel); + Message message4 = createMessage("XYZ", "ABC", 4, 3, replyChannel); + CountDownLatch latch = new CountDownLatch(3); + aggregator.handle(message1); + aggregator.handle(message2); + aggregator.handle(message3); + latch.await(1000, TimeUnit.MILLISECONDS); + Message reply1 = replyChannel.receive(500); + Message reply2 = replyChannel.receive(500); + Message reply3 = replyChannel.receive(500); + // only messages 1 and 2 must have been received by now + assertNotNull(reply1); + assertEquals(1, reply1.getHeader().getSequenceNumber()); + assertNotNull(reply2); + assertEquals(2, reply2.getHeader().getSequenceNumber()); + assertNull(reply3); + // when sending the last message, the whole sequence must have been sent + latch = new CountDownLatch(1); + aggregator.handle(message4); + latch.await(1000, TimeUnit.MILLISECONDS); + reply3 = replyChannel.receive(500); + Message reply4 = replyChannel.receive(500); + assertNotNull(reply3); + assertEquals(3, reply3.getHeader().getSequenceNumber()); + assertNotNull(reply4); + assertEquals(4, reply4.getHeader().getSequenceNumber()); + } + + + @Test + public void testResequencingWithCompleteSequenceRelease() throws InterruptedException { + ResequencingMessageHandler aggregator = new ResequencingMessageHandler(false); + QueueChannel replyChannel = new QueueChannel(); + Message message1 = createMessage("123", "ABC", 4, 2, replyChannel); + Message message2 = createMessage("456", "ABC", 4, 1, replyChannel); + Message message3 = createMessage("789", "ABC", 4, 4, replyChannel); + Message message4 = createMessage("XYZ", "ABC", 4, 3, replyChannel); + CountDownLatch latch = new CountDownLatch(3); + aggregator.handle(message1); + aggregator.handle(message2); + aggregator.handle(message3); + latch.await(1000, TimeUnit.MILLISECONDS); + Message reply1 = replyChannel.receive(500); + Message reply2 = replyChannel.receive(500); + Message reply3 = replyChannel.receive(500); + // no message must have been received by now + assertNull(reply1); + assertNull(reply2); + assertNull(reply3); + // when sending the last message, the whole sequence must have been sent + latch = new CountDownLatch(1); + aggregator.handle(message4); + latch.await(1000, TimeUnit.MILLISECONDS); + reply1 = replyChannel.receive(500); + reply2 = replyChannel.receive(500); + reply3 = replyChannel.receive(500); + Message reply4 = replyChannel.receive(500); + assertNotNull(reply1); + assertEquals(1, reply1.getHeader().getSequenceNumber()); + assertNotNull(reply2); + assertEquals(2, reply2.getHeader().getSequenceNumber()); + assertNotNull(reply3); + assertEquals(3, reply3.getHeader().getSequenceNumber()); + assertNotNull(reply4); + assertEquals(4, reply4.getHeader().getSequenceNumber()); + } + + private static Message createMessage(String payload, Object correlationId, + int sequenceSize, int sequenceNumber, MessageChannel replyChannel) { + StringMessage message = new StringMessage(payload); + message.getHeader().setCorrelationId(correlationId); + message.getHeader().setSequenceSize(sequenceSize); + message.getHeader().setSequenceNumber(sequenceNumber); + message.getHeader().setReturnAddress(replyChannel); + return message; + } + +}