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

INT-238, INT-371 Clean up aggregator code, solve some robustness issues

Browse Source
This commit is contained in:
Marius Bogoevici
2008-11-23 07:44:18 +00:00
parent f69d774e09
commit 3ee1e4921e
10 changed files with 375 additions and 395 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageAggregator.java
View File

@@ -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
* <em>complete</em> 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}.
*
* <p>The default strategy for determining whether a group is complete is based
* on the '<code>sequenceSize</code>' property of the header. Alternatively, a
* <p>
* The default strategy for determining whether a group is complete is based on
* the '<code>sequenceSize</code>' property of the header. Alternatively, a
* custom implementation of the {@link CompletionStrategy} may be provided.
*
* <p>All considerations regarding <code>timeout</code> and grouping by
* <code>correlationId</code> from {@link AbstractMessageBarrierHandler}
* apply here as well.
* <p>
* All considerations regarding <code>timeout</code> and grouping by
* <code>correlationId</code> 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<Map<Object, Message<?>>, 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<Map<Object, Message<?>>, Object> createMessageBarrier() {
return new MessageBarrier<Map<Object, Message<?>>, Object>(new LinkedHashMap<Object, Message<?>>());
}
protected boolean isBarrierRemovable(Object correlationId, List<Message<?>> releasedMessages) {
return releasedMessages != null && releasedMessages.size() > 0;
@Override
protected void processBarrier(MessageBarrier<Map<Object, Message<?>>, Object> barrier) {
ArrayList<Message<?>> messageList = new ArrayList<Message<?>>(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<Message<?>> messages) {
Message<?> result = this.aggregateMessages(messages);
if (result == null) {
return new Message<?>[0];
@Override
protected boolean canAddMessage(Message<?> message, MessageBarrier<Map<Object, Message<?>>, 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<Map<Object, Message<?>>, Object> barrier) {
barrier.getMessages().put(message.getHeaders().getId(), message);
}
protected abstract Message<?> aggregateMessages(List<Message<?>> messages);

106
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrier.java
View File

@@ -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<Message<?>> messages = new ArrayList<Message<?>>();
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 <em>if available</em>.
* Otherwise, the return value will be <code>null</code>.
*/
public List<Message<?>> 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<Message<?>> 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<Message<?>> releaseAvailableMessages();
}

171
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AbstractMessageBarrierHandler.java
View File

@@ -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.
* <p>
* Subclasses must decide what kind of a Map they want to use, what is the logic
* for adding messages to the barrier through the '<code>doAddMessage</code>'
* method.
*
* @author Mark Fisher
* @author Marius Bogoevici
*/
public abstract class AbstractMessageBarrierHandler extends AbstractMessageHandler implements TaskSchedulerAware, InitializingBean {
public abstract class AbstractMessageBarrierHandler<T extends Map<K, Message<?>>, 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<Object, MessageBarrier> barriers = new ConcurrentHashMap<Object, MessageBarrier>();
protected final ConcurrentMap<Object, MessageBarrier<T,K>> barriers = new ConcurrentHashMap<Object, MessageBarrier<T,K>>();
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<T,K> barrier = barriers.putIfAbsent(correlationId, createMessageBarrier());
if (barrier == null) {
barrier = barriers.get(correlationId);
barrier = barriers.get(message.getHeaders().getCorrelationId());
}
List<Message<?>> 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<Message<?>> releasedMessages) {
Message<?>[] processedMessages = this.processReleasedMessages(correlationId, releasedMessages);
if (ObjectUtils.isEmpty(processedMessages)) {
protected final void sendReplies(Collection<Message<?>> 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<T, K> 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<T, K> 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<T, K> 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<T, K> 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<Object, MessageBarrier> entry : barriers.entrySet()) {
for (Map.Entry<Object, MessageBarrier<T,K>> entry : barriers.entrySet()) {
if (currentTime - entry.getValue().getTimestamp() >= timeout) {
Object correlationId = entry.getKey();
List<Message<?>> messages = entry.getValue().getMessages();
removeBarrier(correlationId);
if (sendPartialResultOnTimeout) {
afterRelease(correlationId, messages);
}
else {
for (Message<?> message : messages) {
discardMessage(message, correlationId);
MessageBarrier<T,K> 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<Message<?>> releasedMessages);
/**
* Implements the logic for transforming the released Messages.
*/
protected abstract Message<?>[] processReleasedMessages(Object correlationId, List<Message<?>> messages);
}

48
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/AggregationBarrier.java
View File

@@ -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<Message<?>> releaseAvailableMessages() {
return (this.isComplete()) ? this.getMessages() : null;
}
protected boolean hasReceivedAllMessages() {
return completionStrategy.isComplete(this.messages);
}
}

64
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/MessageBarrier.java
View File

@@ -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<T extends Map<K, Message<?>>, K> {
List<Message<?>> addAndRelease(Message<?> message);
protected final T messages;
long getTimestamp();
private volatile boolean complete = false;
List<Message<?>> 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;
}
}

99
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/Resequencer.java
View File

@@ -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<SortedMap<Integer, Message<?>>, 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<SortedMap<Integer, Message<?>>, Integer> createMessageBarrier() {
MessageBarrier<SortedMap<Integer, Message<?>>, Integer> messageBarrier
= new MessageBarrier<SortedMap<Integer, Message<?>>, Integer>(new TreeMap<Integer, Message<?>>());
messageBarrier.getMessages().put(0, createFlagMessage(0));
return messageBarrier;
}
@Override
protected void processBarrier(MessageBarrier<SortedMap<Integer, Message<?>>, Integer> barrier) {
if (hasReceivedAllMessages(barrier.getMessages())) {
barrier.setComplete();
}
List<Message<?>> 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<Message<?>> messages) {
return messages.toArray(new Message<?>[messages.size()]);
private boolean hasReceivedAllMessages(SortedMap <Integer, Message<?>> 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<Message<?>> releasedMessages) {
MessageHeaders lastMessageHeaders = releasedMessages.get(releasedMessages.size() - 1).getHeaders();
return (lastMessageHeaders.getSequenceNumber() == lastMessageHeaders.getSequenceSize());
private List<Message<?>> releaseAvailableMessages(MessageBarrier<SortedMap<Integer, Message<?>>, Integer> barrier) {
if (this.releasePartialSequences || barrier.isComplete()) {
ArrayList<Message<?>> releasedMessages = new ArrayList<Message<?>>();
Iterator<Message<?>> 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<Message<?>>();
}
}
@Override
protected boolean canAddMessage(Message<?> message,
MessageBarrier<SortedMap<Integer, Message<?>>, 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<SortedMap<Integer, Message<?>>, Integer> barrier) {
//add the message to the barrier, indexing it by its sequence number
barrier.getMessages().put(message.getHeaders().getSequenceNumber(), message);
}
private static Message<Integer> createFlagMessage(int sequenceNumber) {
return MessageBuilder.withPayload(sequenceNumber).setSequenceNumber(sequenceNumber).build();
}
}

99
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/ResequencingMessageBarrier.java
View File

@@ -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<Message<?>> 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<Message<?>>() {
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<Message<?>> releaseAvailableMessages() {
if (this.releasePartialSequences || hasReceivedAllMessages()) {
ArrayList<Message<?>> releasedMessages = new ArrayList<Message<?>>();
Iterator<Message<?>> 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<Message<?>>();
}
}
}

46
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorEndpointTests.java
View File

@@ -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();

36
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregationBarrierTests.java → org.springframework.integration/src/test/java/org/springframework/integration/aggregator/MessageBarrierTests.java
View File

@@ -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<Message<?>> messages) {
return (messages.size() == 2);
}
}
}

24
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/ResequencerTests.java
View File

@@ -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);