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 098ada7421..e714ff20f0 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 @@ -359,8 +359,23 @@ public abstract class AbstractMessageBarrierHandler barrier); + + /** + * A method for discarding the content of the message barrier. + * Can be overridden by subclasses. + * @param entry + * @param barrier + */ + protected void discardBarrier(MessageBarrier barrier) { + for (Message message : barrier.getMessages()) { + if (logger.isDebugEnabled()) { + logger.debug("Handling of Message group with correlation key '" + barrier.getCorrelationKey()+ "' has timed out."); + } + discardMessage(message); + } + } - /** + /** * A task that runs periodically, pruning the timed-out message barriers. */ private class PrunerTask implements Runnable { @@ -377,13 +392,7 @@ public abstract class AbstractMessageBarrierHandler> { private Object correlationKey; private final long timestamp = System.currentTimeMillis(); + + private final Map attributes = new HashMap(); public MessageBarrier(T messages, Object correlationKey) { this.messages = messages; @@ -81,5 +86,22 @@ public class MessageBarrier> { public T getMessages() { return this.messages; } + + /** + * Sets a the value of a given attribute on the MessageBarrier. + * @param attributeName + * @param value + */ + public void setAttribute(String attributeName, Object value) { + this.attributes.put(attributeName, value); + } + /** + * Gets the value of a given attribute from the MessageBarrier. + * @param attributeName + */ + public V getAttribute(String attributeName) { + return (V)this.attributes.get(attributeName); + } + } 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 163406720f..bd8bb8feb9 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 @@ -23,7 +23,6 @@ import java.util.SortedSet; import java.util.TreeSet; import org.springframework.integration.core.Message; -import org.springframework.integration.message.MessageBuilder; import org.springframework.util.CollectionUtils; /** @@ -37,10 +36,16 @@ import org.springframework.util.CollectionUtils; * All considerations regarding timeout and grouping by * 'correlationId' from {@link AbstractMessageBarrierHandler} * apply here as well. + * + * It is assumed that all messages have the same sequence_size header attribute + * and that the sequence numbers of the messages are successive, starting with + * 1 up to sequenceSize. Messages that do not satisfy this condition are + * considered out-of-sequence and thus rejected. + * * * Note: messages with the same sequence number will be treated as equivalent * by this class (i.e. after a message with a given sequence number is received, - * further messages from withing the same group, that have the same sequence number, + * further messages from within the same group, that have the same sequence number, * will be ignored. * * @author Marius Bogoevici @@ -48,7 +53,9 @@ import org.springframework.util.CollectionUtils; public class Resequencer extends AbstractMessageBarrierHandler>> { private volatile boolean releasePartialSequences = true; - + + private static final String LAST_RELEASED_SEQUENCE_NUMBER = "last.released.sequence.number"; + public void setReleasePartialSequences(boolean releasePartialSequences) { this.releasePartialSequences = releasePartialSequences; @@ -58,13 +65,13 @@ public class Resequencer extends AbstractMessageBarrierHandler>> createMessageBarrier(Object correlationKey) { MessageBarrier>> messageBarrier = new MessageBarrier>>(new TreeSet>(new MessageSequenceComparator()), correlationKey); - messageBarrier.getMessages().add(createFlagMessage(0)); + messageBarrier.setAttribute(LAST_RELEASED_SEQUENCE_NUMBER, 0); return messageBarrier; } @Override protected void processBarrier(MessageBarrier>> barrier) { - if (hasReceivedAllMessages(barrier.getMessages())) { + if (hasReceivedAllMessages(barrier)) { barrier.setComplete(); } List> releasedMessages = releaseAvailableMessages(barrier); @@ -77,21 +84,18 @@ public class Resequencer extends AbstractMessageBarrierHandler> messages) { - Message firstMessage = messages.first(); - Message lastMessage = messages.last(); - return (lastMessage.getHeaders().getSequenceNumber().equals(lastMessage.getHeaders().getSequenceSize()) - && (lastMessage.getHeaders().getSequenceNumber() - firstMessage.getHeaders().getSequenceNumber() == messages.size() - 1)); + private boolean hasReceivedAllMessages(MessageBarrier>> barrier) { + int sequenceSize = barrier.getMessages().first().getHeaders().getSequenceSize(); + int messagesCurrentlyInBarrier = barrier.getMessages().size(); + int lastReleasedSequenceNumber = barrier.getAttribute(LAST_RELEASED_SEQUENCE_NUMBER); + return (lastReleasedSequenceNumber + messagesCurrentlyInBarrier == sequenceSize); } private List> releaseAvailableMessages(MessageBarrier>> barrier) { if (this.releasePartialSequences || barrier.isComplete()) { ArrayList> releasedMessages = new ArrayList>(); Iterator> it = barrier.getMessages().iterator(); - //remove the initial flag from the list - Message flag = it.next(); - it.remove(); - int lastReleasedSequenceNumber = flag.getHeaders().getSequenceNumber(); + int lastReleasedSequenceNumber = barrier.getAttribute(LAST_RELEASED_SEQUENCE_NUMBER); while (it.hasNext()) { Message currentMessage = it.next(); if (lastReleasedSequenceNumber == currentMessage.getHeaders().getSequenceNumber() - 1) { @@ -103,8 +107,7 @@ public class Resequencer extends AbstractMessageBarrierHandler message, - MessageBarrier>> barrier) { + protected boolean canAddMessage(Message message, MessageBarrier>> barrier) { if (!super.canAddMessage(message, barrier)) { return false; } - Message flagMessage = barrier.getMessages().first(); + int lastReleasedSequenceNumber = barrier.getAttribute(LAST_RELEASED_SEQUENCE_NUMBER); if (barrier.messages.contains(message) - || flagMessage.getHeaders().getSequenceNumber() >= message.getHeaders().getSequenceNumber()) { + || lastReleasedSequenceNumber >= message.getHeaders().getSequenceNumber()) { logger.debug("A message with the same sequence number has been already received: " + message); return false; } - Message lastMessage = barrier.getMessages().last(); - if (lastMessage != flagMessage - && lastMessage.getHeaders().getSequenceSize() < message.getHeaders().getSequenceNumber()) { + // one can always add a message to the barrier if it's empty. Afterwards, assume that the complete sequence size + // + if (!barrier.getMessages().isEmpty() && + barrier.getMessages().first().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; } - private static Message createFlagMessage(int sequenceNumber) { - return MessageBuilder.withPayload(sequenceNumber).setSequenceNumber(sequenceNumber).build(); - } - } 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 b32617b99c..a4f968d876 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 @@ -124,6 +124,37 @@ public class ResequencerTests { assertNotNull(reply4); assertEquals(new Integer(4), reply4.getHeaders().getSequenceNumber()); } + + @Test + public void testResequencingWithDiscard() throws InterruptedException { + QueueChannel discardChannel = new QueueChannel(); + Message message1 = createMessage("123", "ABC", 4, 2, null); + Message message2 = createMessage("456", "ABC", 4, 1, null); + Message message3 = createMessage("789", "ABC", 4, 4, null); + Message message4 = createMessage("XYZ", "ABC", 4, 3, null); + this.resequencer.setSendPartialResultOnTimeout(false); + this.resequencer.setReleasePartialSequences(false); + this.resequencer.setDiscardChannel(discardChannel); + this.resequencer.setTimeout(90000); + this.resequencer.handleMessage(message1); + this.resequencer.handleMessage(message2); + this.resequencer.discardBarrier(this.resequencer.barriers.get("ABC")); + Message reply1 = discardChannel.receive(0); + Message reply2 = discardChannel.receive(0); + Message reply3 = discardChannel.receive(0); + // only messages 1 and 2 should have been received by now + assertNotNull(reply1); + assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber()); + assertNotNull(reply2); + assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber()); + assertNull(reply3); + // when sending the last message, the whole sequence must have been sent + this.resequencer.handleMessage(message4); + reply3 = discardChannel.receive(0); + assertNull(reply3); + Message reply4 = discardChannel.receive(0); + assertNull(reply4); + } @Test