diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageAggregator.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageAggregator.java index bcb887ee2f..f9938a3ae2 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageAggregator.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageAggregator.java @@ -16,60 +16,95 @@ package org.springframework.integration.aggregator; +import java.util.ArrayList; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageBuilder; import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; /** - * A base class for aggregating a group of Messages into a single Message. + * A base class for aggregating a group of Messages into a single Message. * Extends {@link AbstractMessageBarrierHandler} and waits for a * complete group of {@link Message Messages} to arrive. Subclasses * must provide the implementation of the {@link #aggregateMessages(List)} * method to combine the group of Messages into a single {@link Message}. * - *

The default strategy for determining whether a group is complete is based - * on the 'sequenceSize' property of the header. Alternatively, a + *

+ * 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. * - *

All considerations regarding timeout and grouping by - * correlationId from {@link AbstractMessageBarrierHandler} - * apply here as well. + *

+ * All considerations regarding timeout and grouping by + * correlationId from {@link AbstractMessageBarrierHandler} apply + * here as well. * * @author Mark Fisher * @author Marius Bogoevici */ -public abstract class AbstractMessageAggregator extends AbstractMessageBarrierHandler { +public abstract class AbstractMessageAggregator extends + AbstractMessageBarrierHandler>, Object> { private volatile CompletionStrategy completionStrategy = new SequenceSizeCompletionStrategy(); - + /** * Strategy to determine whether the group of messages is complete. */ public void setCompletionStrategy(CompletionStrategy completionStrategy) { - Assert.notNull(completionStrategy, "'completionStrategy' must not be null"); + Assert.notNull(completionStrategy, + "'completionStrategy' must not be null"); this.completionStrategy = completionStrategy; } - protected MessageBarrier createMessageBarrier() { - return new AggregationBarrier(this.completionStrategy); + @Override + protected MessageBarrier>, Object> createMessageBarrier() { + return new MessageBarrier>, Object>(new LinkedHashMap>()); } - protected boolean isBarrierRemovable(Object correlationId, List> releasedMessages) { - return releasedMessages != null && releasedMessages.size() > 0; + @Override + protected void processBarrier(MessageBarrier>, Object> barrier) { + ArrayList> messageList = new ArrayList>(barrier.getMessages().values()); + if (!barrier.isComplete() && !CollectionUtils.isEmpty(messageList)) { + if (this.completionStrategy.isComplete(messageList)) { + barrier.setComplete(); + } + } + if (barrier.isComplete()) { + this.removeBarrier(barrier.getCorrelationId()); + Message result = this.aggregateMessages(messageList); + if (result != null) { + if (result.getHeaders().getCorrelationId() == null) { + result = MessageBuilder.fromMessage(result) + .setCorrelationId(barrier.getCorrelationId()) + .build(); + } + this.sendReply(result, this.resolveReplyChannelFromMessage(messageList.get(0))); + } + } } - - protected Message[] processReleasedMessages(Object correlationId, List> messages) { - Message result = this.aggregateMessages(messages); - if (result == null) { - return new Message[0]; + + @Override + protected boolean canAddMessage(Message message, MessageBarrier>, Object> barrier) { + if (!super.canAddMessage(message, barrier)) { + return false; } - if (result.getHeaders().getCorrelationId() == null) { - result = MessageBuilder.fromMessage(result).setCorrelationId(correlationId).build(); + if (barrier.messages.containsKey(message.getHeaders().getId())) { + logger.debug("The barrier has received message: " + message + + ", but it already contains a similar message: " + + barrier.getMessages().get(message.getHeaders().getId())); + return false; } - return new Message[] { result }; + return true; + } + + @Override + protected void doAddMessage(Message message, MessageBarrier>, Object> barrier) { + barrier.getMessages().put(message.getHeaders().getId(), message); } protected abstract Message aggregateMessages(List> messages); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrier.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrier.java deleted file mode 100644 index 20f41bd868..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrier.java +++ /dev/null @@ -1,106 +0,0 @@ -/* - * 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.aggregator; - -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.core.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; - } - this.addMessage(message); - this.complete = this.hasReceivedAllMessages(); - return this.releaseAvailableMessages(); - } - finally { - this.lock.unlock(); - } - } - - protected void addMessage(Message message) { - this.messages.add(message); - } - - public List> getMessages() { - return Collections.unmodifiableList(this.messages); - } - - /** - * Subclasses must 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 must 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/aggregator/AbstractMessageBarrierHandler.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java index e53ba83a49..c335bfa80e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java @@ -16,7 +16,7 @@ package org.springframework.integration.aggregator; -import java.util.List; +import java.util.Collection; import java.util.Map; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; @@ -39,8 +39,6 @@ import org.springframework.integration.scheduling.IntervalTrigger; import org.springframework.integration.scheduling.TaskScheduler; import org.springframework.integration.scheduling.TaskSchedulerAware; import org.springframework.util.Assert; -import org.springframework.util.CollectionUtils; -import org.springframework.util.ObjectUtils; /** * Base class for {@link MessageBarrier}-based Message Handlers. A @@ -60,11 +58,16 @@ import org.springframework.util.ObjectUtils; * Messages with that timed-out 'correlationId' will be sent to the * 'discardChannel' if provided unless 'sendPartialResultsOnTimeout' is set to * true in which case the incomplete group will be sent to the output channel. + *

+ * Subclasses must decide what kind of a Map they want to use, what is the logic + * for adding messages to the barrier through the 'doAddMessage' + * method. * * @author Mark Fisher * @author Marius Bogoevici */ -public abstract class AbstractMessageBarrierHandler extends AbstractMessageHandler implements TaskSchedulerAware, InitializingBean { +public abstract class AbstractMessageBarrierHandler>, K> + extends AbstractMessageHandler implements TaskSchedulerAware, InitializingBean { public final static long DEFAULT_SEND_TIMEOUT = 1000; @@ -76,13 +79,13 @@ public abstract class AbstractMessageBarrierHandler extends AbstractMessageHandl protected final Log logger = LogFactory.getLog(this.getClass()); - private MessageChannel outputChannel; + private volatile MessageChannel outputChannel; private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); private volatile MessageChannel discardChannel; - protected final ConcurrentMap barriers = new ConcurrentHashMap(); + protected final ConcurrentMap> barriers = new ConcurrentHashMap>(); private volatile long timeout = DEFAULT_TIMEOUT; @@ -96,9 +99,9 @@ public abstract class AbstractMessageBarrierHandler extends AbstractMessageHandl private volatile boolean initialized; - private TaskScheduler taskScheduler; + private volatile TaskScheduler taskScheduler; - private ScheduledFuture reaperFutureTask; + private volatile ScheduledFuture reaperFutureTask; public AbstractMessageBarrierHandler() { @@ -173,7 +176,7 @@ public abstract class AbstractMessageBarrierHandler extends AbstractMessageHandl return; } Assert.state(this.taskScheduler != null, "TaskScheduler must not be null"); - this.reaperFutureTask = this.taskScheduler.schedule(new ReaperTask(), new IntervalTrigger(this.reaperInterval, + this.reaperFutureTask = this.taskScheduler.schedule(new PrunerTask(), new IntervalTrigger(this.reaperInterval, TimeUnit.MILLISECONDS)); } @@ -194,18 +197,18 @@ public abstract class AbstractMessageBarrierHandler extends AbstractMessageHandl + " requires the 'correlationId' property"); } if (this.trackedCorrelationIds.contains(correlationId)) { - this.discardMessage(message, correlationId); + if (logger.isDebugEnabled()) { + logger.debug("Handling of Message group with correlationId '" + correlationId + + "' has already completed or timed out."); + } + this.discardMessage(message); } else { this.processMessage(message, correlationId); } } - private void discardMessage(Message message, Object correlationId) { - if (logger.isDebugEnabled()) { - logger.debug("Handling of Message group with correlationId '" + correlationId - + "' has already completed or timed out."); - } + private void discardMessage(Message message) { if (this.discardChannel != null) { boolean sent = this.channelTemplate.send(message, this.discardChannel); if (!sent && logger.isWarnEnabled()) { @@ -215,42 +218,44 @@ public abstract class AbstractMessageBarrierHandler extends AbstractMessageHandl } private void processMessage(Message message, Object correlationId) { - MessageBarrier barrier = barriers.putIfAbsent(correlationId, createMessageBarrier()); + MessageBarrier barrier = barriers.putIfAbsent(correlationId, createMessageBarrier()); if (barrier == null) { - barrier = barriers.get(correlationId); + barrier = barriers.get(message.getHeaders().getCorrelationId()); } - List> releasedMessages = barrier.addAndRelease(message); - if (!CollectionUtils.isEmpty(releasedMessages)) { - if (isBarrierRemovable(correlationId, releasedMessages)) { - this.removeBarrier(correlationId); + synchronized (barrier) { + if (canAddMessage(message, barrier)) { + doAddMessage(message, barrier); } - this.afterRelease(correlationId, releasedMessages); + processBarrier(barrier); } } - private void afterRelease(Object correlationId, List> releasedMessages) { - Message[] processedMessages = this.processReleasedMessages(correlationId, releasedMessages); - if (ObjectUtils.isEmpty(processedMessages)) { + protected final void sendReplies(Collection> messages, MessageChannel defaultReplyChannel) { + if (messages.isEmpty()) { return; } - for (Message result : processedMessages) { - MessageChannel replyChannel = this.outputChannel; - if (replyChannel == null) { - replyChannel = this.resolveReplyChannelFromMessage(result); - if (replyChannel == null) { - replyChannel = this.resolveReplyChannelFromMessage(releasedMessages.get(0)); - } - } - if (replyChannel != null) { - this.channelTemplate.send(result, replyChannel); - } - else if (logger.isWarnEnabled()) { - logger.warn("unable to determine reply target for aggregation result: " + result); - } + for (Message result : messages) { + sendReply(result, defaultReplyChannel); } } - protected MessageChannel resolveReplyChannelFromMessage(Message message) { + protected final void sendReply(Message message, MessageChannel defaultReplyChannel) { + MessageChannel replyChannel = this.outputChannel; + if (replyChannel == null) { + replyChannel = this.resolveReplyChannelFromMessage(message); + if (replyChannel == null) { + replyChannel = defaultReplyChannel; + } + } + if (replyChannel != null) { + this.channelTemplate.send(message, replyChannel); + } + else if (logger.isWarnEnabled()) { + logger.warn("unable to determine reply target for aggregation result: " + message); + } + } + + protected final MessageChannel resolveReplyChannelFromMessage(Message message) { Object replyChannel = message.getHeaders().getReplyChannel(); if (replyChannel != null) { if (replyChannel instanceof MessageChannel) { @@ -263,7 +268,7 @@ public abstract class AbstractMessageBarrierHandler extends AbstractMessageHandl return null; } - private void removeBarrier(Object correlationId) { + protected final void removeBarrier(Object correlationId) { if (this.barriers.remove(correlationId) != null) { synchronized (this.trackedCorrelationIds) { boolean added = this.trackedCorrelationIds.offer(correlationId); @@ -275,43 +280,71 @@ public abstract class AbstractMessageBarrierHandler extends AbstractMessageHandl } } - private class ReaperTask implements Runnable { + /** + * Verifies that a message can be added to the barrier. To be overridden by subclasses, which may add + * their own verifications. Subclasses overriding this method must call the method from the superclass. + */ + protected boolean canAddMessage(Message message, MessageBarrier barrier) { + if (barrier.isComplete()) { + if (logger.isDebugEnabled()) { + logger.debug("Message received after aggregation has already completed: " + message); + } + return false; + } + return true; + } + + + /** + * Factory method for creating a MessageBarrier implementation. + */ + protected abstract MessageBarrier createMessageBarrier(); + + /** + * A method for processing the information in the message barrier after a message has been added or on pruning. + * The decision as to whether the messages from the {@link MessageBarrier} + * can be released normally belongs here, although calling code may forcibly set the MessageBarrier's 'complete' + * flag to true before invoking the method. + * @param barrier the {@link MessageBarrier} to be processed + */ + protected abstract void processBarrier(MessageBarrier barrier); + + /** + * A method implemented by subclasses to add the incoming message to the message barrier. This is deferred to subclasses, + * as they should have full control over how the messages are indexed in the MessageBarrier. + */ + protected abstract void doAddMessage(Message message, MessageBarrier barrier); + + /** + * A task that runs periodically, pruning the timed-out message barriers. + */ + private class PrunerTask implements Runnable { public void run() { long currentTime = System.currentTimeMillis(); - for (Map.Entry entry : barriers.entrySet()) { + 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) { - discardMessage(message, correlationId); + MessageBarrier barrier = entry.getValue(); + synchronized (barrier) { + removeBarrier(entry.getKey()); + if (sendPartialResultOnTimeout) { + barrier.setComplete(); + processBarrier(barrier); + } + else { + for (Object message : barrier.getMessages().values()) { + if (logger.isDebugEnabled()) { + logger.debug("Handling of Message group with correlationId '" + entry.getKey() + + "' has timed out."); + } + discardMessage((Message) 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 complete 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/aggregator/AggregationBarrier.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregationBarrier.java deleted file mode 100644 index 4893e635b1..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregationBarrier.java +++ /dev/null @@ -1,48 +0,0 @@ -/* - * 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.aggregator; - -import java.util.List; - -import org.springframework.integration.core.Message; - -/** - * MessageBarrier implementation for message aggregation. Delegates to a - * {@link CompletionStrategy} to determine when the group of messages is ready - * for aggregation. - * - * @author Marius Bogoevici - * @author Mark Fisher - */ -public class AggregationBarrier extends AbstractMessageBarrier { - - protected final CompletionStrategy completionStrategy; - - - public AggregationBarrier(CompletionStrategy completionStrategy) { - this.completionStrategy = completionStrategy; - } - - - protected List> releaseAvailableMessages() { - return (this.isComplete()) ? this.getMessages() : null; - } - - protected boolean hasReceivedAllMessages() { - return completionStrategy.isComplete(this.messages); - } -} diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageBarrier.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageBarrier.java index 28dcfef163..a497f53d6e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageBarrier.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageBarrier.java @@ -16,24 +16,70 @@ package org.springframework.integration.aggregator; -import java.util.List; +import java.util.Map; import org.springframework.integration.core.Message; /** - * Common interface for routing components that release a list of - * {@link Message Messages} based upon a condition that is met when a - * {@link Message} arrives. + * Utility class for AbstractMessageBarrierHandler and its subclasses for + * storing objects while in transit. It is a wrapper around a {@link Map}, + * providing special properties for recording the complete status, the creation + * time (for determining if a group of messages has timed out), and the + * correlation id for a group of messages (available after the first message has + * been added to it). This is a parameterized type, allowing different different + * client classes to use different types of Maps and their respective features. + * + * This class is not thread-safe and will be synchronized by the calling code. * - * @author Mark Fisher * @author Marius Bogoevici */ -public interface MessageBarrier { +public class MessageBarrier>, K> { - List> addAndRelease(Message message); + protected final T messages; - long getTimestamp(); + private volatile boolean complete = false; - List> getMessages(); + private final long timestamp = System.currentTimeMillis(); + + public MessageBarrier(T messages) { + this.messages = messages; + } + + public Object getCorrelationId() { + if (!messages.isEmpty()) { + return messages.values().iterator().next().getHeaders() + .getCorrelationId(); + } + return null; + } + + /** + * 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; + } + + /** + * Marks the barrier as complete. + */ + public void setComplete() { + this.complete = true; + } + + /** + * True if the barrier has received all the messages and can proceed to + * release them. + */ + public boolean isComplete() { + return this.complete; + } + + public T getMessages() { + return this.messages; + } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java index bf28871b44..72b8e93679 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java @@ -16,10 +16,15 @@ package org.springframework.integration.aggregator; +import java.util.ArrayList; +import java.util.Iterator; import java.util.List; +import java.util.SortedMap; +import java.util.TreeMap; import org.springframework.integration.core.Message; -import org.springframework.integration.core.MessageHeaders; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.util.CollectionUtils; /** * An {@link AbstractMessageBarrierHandler} that waits for a group of @@ -35,7 +40,7 @@ import org.springframework.integration.core.MessageHeaders; * * @author Marius Bogoevici */ -public class Resequencer extends AbstractMessageBarrierHandler { +public class Resequencer extends AbstractMessageBarrierHandler>, Integer> { private volatile boolean releasePartialSequences = true; @@ -44,17 +49,93 @@ public class Resequencer extends AbstractMessageBarrierHandler { this.releasePartialSequences = releasePartialSequences; } - protected MessageBarrier createMessageBarrier() { - return new ResequencingMessageBarrier(this.releasePartialSequences); + @Override + protected MessageBarrier>, Integer> createMessageBarrier() { + MessageBarrier>, Integer> messageBarrier + = new MessageBarrier>, Integer>(new TreeMap>()); + messageBarrier.getMessages().put(0, createFlagMessage(0)); + return messageBarrier; + } + + @Override + protected void processBarrier(MessageBarrier>, Integer> barrier) { + if (hasReceivedAllMessages(barrier.getMessages())) { + barrier.setComplete(); + } + List> releasedMessages = releaseAvailableMessages(barrier); + if (!CollectionUtils.isEmpty(releasedMessages)) { + Message lastMessage = releasedMessages.get(releasedMessages.size()-1); + if (lastMessage.getHeaders().getSequenceNumber().equals(lastMessage.getHeaders().getSequenceSize() - 1)) { + this.removeBarrier(barrier.getCorrelationId()); + } + this.sendReplies(releasedMessages, this.resolveReplyChannelFromMessage(releasedMessages.get(0))); + } } - protected Message[] processReleasedMessages(Object correlationId, List> messages) { - return messages.toArray(new Message[messages.size()]); + private boolean hasReceivedAllMessages(SortedMap > messages) { + Message firstMessage = messages.get(messages.firstKey()); + Message lastMessage = messages.get(messages.lastKey()); + return (lastMessage.getHeaders().getSequenceNumber() == lastMessage.getHeaders().getSequenceSize() + && (lastMessage.getHeaders().getSequenceNumber() - firstMessage.getHeaders().getSequenceNumber() == messages.size() - 1)); } - protected boolean isBarrierRemovable(Object correlationId, List> releasedMessages) { - MessageHeaders lastMessageHeaders = releasedMessages.get(releasedMessages.size() - 1).getHeaders(); - return (lastMessageHeaders.getSequenceNumber() == lastMessageHeaders.getSequenceSize()); + private List> releaseAvailableMessages(MessageBarrier>, Integer> barrier) { + if (this.releasePartialSequences || barrier.isComplete()) { + ArrayList> releasedMessages = new ArrayList>(); + Iterator> it = barrier.getMessages().values().iterator(); + //remove the initial flag from the list + Message flag = it.next(); + it.remove(); + int lastReleasedSequenceNumber = flag.getHeaders().getSequenceNumber(); + while (it.hasNext()) { + Message currentMessage = it.next(); + if (lastReleasedSequenceNumber == currentMessage.getHeaders().getSequenceNumber() - 1) { + releasedMessages.add(currentMessage); + lastReleasedSequenceNumber = currentMessage.getHeaders().getSequenceNumber(); + it.remove(); + } + else { + break; + } + } + //re-insert the flag so that we know where to start releasing next + barrier.getMessages().put(lastReleasedSequenceNumber, createFlagMessage(lastReleasedSequenceNumber)); + return releasedMessages; + } + else { + return new ArrayList>(); + } + } + + @Override + protected boolean canAddMessage(Message message, + MessageBarrier>, Integer> barrier) { + if (!super.canAddMessage(message, barrier)) { + return false; + } + Message flagMessage = barrier.getMessages().get(barrier.getMessages().firstKey()); + if (barrier.messages.containsKey(message.getHeaders().getSequenceNumber()) + || flagMessage.getHeaders().getSequenceNumber() >= message.getHeaders().getSequenceNumber()) { + logger.debug("A message with the same sequence number has been already received: " + message); + return false; + } + Message lastMessage = barrier.getMessages().get(barrier.getMessages().lastKey()); + if (lastMessage != flagMessage + && lastMessage.getHeaders().getSequenceSize() < message.getHeaders().getSequenceNumber()) { + logger.debug("The message has a sequence number which is larger than the sequence size: "+ message); + return false; + } + return true; + } + + @Override + protected void doAddMessage(Message message, MessageBarrier>, Integer> barrier) { + //add the message to the barrier, indexing it by its sequence number + barrier.getMessages().put(message.getHeaders().getSequenceNumber(), message); + } + + private static Message createFlagMessage(int sequenceNumber) { + return MessageBuilder.withPayload(sequenceNumber).setSequenceNumber(sequenceNumber).build(); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ResequencingMessageBarrier.java b/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ResequencingMessageBarrier.java deleted file mode 100644 index 5e568b8dd9..0000000000 --- a/org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ResequencingMessageBarrier.java +++ /dev/null @@ -1,99 +0,0 @@ -/* - * 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.aggregator; - -import java.util.ArrayList; -import java.util.Collections; -import java.util.Comparator; -import java.util.Iterator; -import java.util.List; - -import org.springframework.integration.core.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.getHeaders().getSequenceNumber() - m2.getHeaders().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.getHeaders().getSequenceNumber() == lastMessage.getHeaders().getSequenceSize() - && (lastMessage.getHeaders().getSequenceNumber() - firstMessage.getHeaders().getSequenceNumber() - == this.messages.size() - 1 - && this.lastReleasedSequenceNumber == firstMessage.getHeaders().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.getHeaders().getSequenceNumber() - 1) { - releasedMessages.add(currentMessage); - this.lastReleasedSequenceNumber = currentMessage.getHeaders().getSequenceNumber(); - it.remove(); - } - else { - break; - } - } - return releasedMessages; - } - else { - return new ArrayList>(); - } - } - -} diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorEndpointTests.java index e3c2e27753..624b49729b 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorEndpointTests.java @@ -16,11 +16,6 @@ package org.springframework.integration.aggregator; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; - import java.util.ArrayList; import java.util.Collections; import java.util.List; @@ -28,9 +23,12 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import org.junit.After; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; import org.junit.Before; import org.junit.Test; - import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.core.task.TaskExecutor; import org.springframework.integration.channel.QueueChannel; @@ -44,6 +42,7 @@ import org.springframework.integration.scheduling.TaskScheduler; /** * @author Mark Fisher + * @author Marius Bogoevici */ public class AggregatorEndpointTests { @@ -80,6 +79,41 @@ public class AggregatorEndpointTests { assertEquals("123456789", reply.getPayload()); } + @Test + public void testCompleteGroupWithinTimeoutWithDuplicates() throws InterruptedException { + QueueChannel replyChannel = new QueueChannel(); + Message message1 = createMessage("123", "ABC", 3, 1, replyChannel); + Message message2 = createMessage("456", "ABC", 3, 2, replyChannel); + Message message3 = createMessage("789", "ABC", 3, 3, replyChannel); + CountDownLatch latch = new CountDownLatch(3); + //for testing the duplication scenario, the messages must be processed synchronously + new AggregatorTestTask(this.aggregator, message1, latch).run(); + new AggregatorTestTask(this.aggregator, message2, latch).run(); + new AggregatorTestTask(this.aggregator, message2, latch).run(); + new AggregatorTestTask(this.aggregator, message3, latch).run(); + Message reply = replyChannel.receive(500); + assertNotNull(reply); + assertEquals("123456789", reply.getPayload()); + } + + @Test + public void testCompleteGroupWithinTimeoutWithInconsistentStructure() throws InterruptedException { + QueueChannel replyChannel = new QueueChannel(); + Message message1 = createMessage("123", "ABC", 3, 1, replyChannel); + Message message2 = createMessage("456", "ABC", 3, 2, replyChannel); + Message message4 = createMessage("xyz", "ABC", 4, 3, replyChannel); + Message message3 = createMessage("789", "ABC", 3, 3, replyChannel); + CountDownLatch latch = new CountDownLatch(3); + //for testing the duplication scenario, the messages must be processed synchronously + new AggregatorTestTask(this.aggregator, message1, latch).run(); + new AggregatorTestTask(this.aggregator, message2, latch).run(); + new AggregatorTestTask(this.aggregator, message2, latch).run(); + new AggregatorTestTask(this.aggregator, message3, latch).run(); + Message reply = replyChannel.receive(500); + assertNotNull(reply); + assertEquals("123456789", reply.getPayload()); + } + @Test public void testShouldNotSendPartialResultOnTimeoutByDefault() throws InterruptedException { QueueChannel discardChannel = new QueueChannel(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregationBarrierTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MessageBarrierTests.java similarity index 53% rename from org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregationBarrierTests.java rename to org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MessageBarrierTests.java index 09715da719..e1105551d0 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregationBarrierTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MessageBarrierTests.java @@ -16,45 +16,33 @@ package org.springframework.integration.aggregator; +import java.util.LinkedHashMap; + import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; - -import java.util.List; - import org.junit.Test; -import org.springframework.integration.aggregator.AggregationBarrier; -import org.springframework.integration.aggregator.CompletionStrategy; -import org.springframework.integration.core.Message; import org.springframework.integration.message.StringMessage; /** * @author Mark Fisher */ -public class AggregationBarrierTests { +public class MessageBarrierTests { - @Test - public void testBasicCompletionCheck() { - AggregationBarrier barrier = new AggregationBarrier(new TwoMessageCompletionStrategy()); - assertNull(barrier.addAndRelease(new StringMessage("test1"))); - assertNotNull(barrier.addAndRelease(new StringMessage("test2"))); - } @Test public void testMessageRetrieval() { - AggregationBarrier barrier = new AggregationBarrier(new TwoMessageCompletionStrategy()); - barrier.addAndRelease(new StringMessage("test1")); + MessageBarrier barrier = new MessageBarrier(new LinkedHashMap()); + barrier.getMessages().put("1", new StringMessage("test1")); assertEquals(1, barrier.getMessages().size()); - barrier.addAndRelease(new StringMessage("test2")); + barrier.getMessages().put("2", new StringMessage("test2")); assertEquals(2, barrier.getMessages().size()); } @Test public void testTimestamp() { long before = System.currentTimeMillis(); - AggregationBarrier barrier = new AggregationBarrier(new TwoMessageCompletionStrategy()); + MessageBarrier barrier = new MessageBarrier(new LinkedHashMap()); long timestamp = barrier.getTimestamp(); assertTrue(before <= timestamp); long after = System.currentTimeMillis(); @@ -63,16 +51,8 @@ public class AggregationBarrierTests { @Test public void testEmptyMessageList() { - AggregationBarrier barrier = new AggregationBarrier(new TwoMessageCompletionStrategy()); + MessageBarrier barrier = new MessageBarrier(new LinkedHashMap()); assertEquals(0, barrier.getMessages().size()); } - - private static class TwoMessageCompletionStrategy implements CompletionStrategy { - - public boolean isComplete(List> messages) { - return (messages.size() == 2); - } - } - } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java index 539c57dfab..b32617b99c 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java @@ -71,6 +71,30 @@ public class ResequencerTests { assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber()); } + @Test + public void testResequencingWithDuplicateMessages() { + this.resequencer.setReleasePartialSequences(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); + this.resequencer.handleMessage(message1); + this.resequencer.handleMessage(message3); + this.resequencer.handleMessage(message3); + this.resequencer.handleMessage(message2); + Message reply1 = replyChannel.receive(0); + Message reply2 = replyChannel.receive(0); + Message reply3 = replyChannel.receive(0); + assertNotNull(reply1); + assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber()); + assertNotNull(reply2); + assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber()); + assertNotNull(reply3); + assertEquals(new Integer(3), reply3.getHeaders().getSequenceNumber()); + } + + + @Test public void testResequencingWithIncompleteSequenceRelease() throws InterruptedException { this.resequencer.setReleasePartialSequences(true);