diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java index f1919d181f..d331560ad5 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandler.java @@ -753,13 +753,20 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP protected static class SequenceAwareMessageGroup extends SimpleMessageGroup { public SequenceAwareMessageGroup(MessageGroup messageGroup) { - super(messageGroup); + /* + * Since this group is temporary, and never added to, we simply use the + * supplied group's message collection for the lookup rather than creating a + * new group. + */ + super(messageGroup.getMessages(), null, messageGroup.getGroupId(), messageGroup.getTimestamp(), + messageGroup.isComplete(), true); } /** - * This method determines whether messages have been added to this group that supersede the given message based on - * its sequence id. This can be helpful to avoid ending up with sequences larger than their required sequence size - * or sequences that are missing certain sequence numbers. + * This method determines whether messages have been added to this group that + * supersede the given message based on its sequence id. This can be helpful to + * avoid ending up with sequences larger than their required sequence size or + * sequences that are missing certain sequence numbers. */ @Override public boolean canAdd(Message message) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java index 29ca08f94e..f5a5b4b4ad 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroup.java @@ -67,20 +67,22 @@ public class SimpleMessageGroup implements MessageGroup { public SimpleMessageGroup(Collection> messages, Object groupId, long timestamp, boolean complete) { - this(new LinkedHashSet>(), messages, groupId, timestamp, complete); + this(new LinkedHashSet>(), messages, groupId, timestamp, complete, false); } - SimpleMessageGroup(Collection> internalStore, Collection> messages, Object groupId, - long timestamp, boolean complete) { + protected SimpleMessageGroup(Collection> internalStore, Collection> messages, + Object groupId, long timestamp, boolean complete, boolean storePreLoaded) { Assert.notNull(internalStore, "'internalStore' must not be null"); - Assert.notNull(messages, "'messages' must not be null"); this.messages = internalStore; this.groupId = groupId; this.timestamp = timestamp; this.complete = complete; - for (Message message : messages) { - if (message != null) { //see INT-2666 - addMessage(message); + if (!storePreLoaded) { + Assert.notNull(messages, "'messages' must not be null"); + for (Message message : messages) { + if (message != null) { //see INT-2666 + addMessage(message); + } } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroupFactory.java b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroupFactory.java index 37d04b32d4..ba45b31126 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroupFactory.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/store/SimpleMessageGroupFactory.java @@ -56,7 +56,7 @@ public class SimpleMessageGroupFactory implements MessageGroupFactory { @Override public MessageGroup create(Collection> messages, Object groupId, long timestamp, boolean complete) { - return new SimpleMessageGroup(this.type.get(), messages, groupId, timestamp, complete); + return new SimpleMessageGroup(this.type.get(), messages, groupId, timestamp, complete, false); } @Override diff --git a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageGroupTests.java b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageGroupTests.java index 4aaace12b0..8af222874a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageGroupTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/store/SimpleMessageGroupTests.java @@ -26,10 +26,12 @@ import java.lang.reflect.Constructor; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; +import java.util.HashSet; import java.util.List; import org.junit.Test; +import org.springframework.beans.DirectFieldAccessor; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; @@ -40,6 +42,7 @@ import org.springframework.util.StopWatch; * @author Oleg Zhurakousky * @author Dave Syer * @author Artem Bilan + * @author Gary Russell */ public class SimpleMessageGroupTests { @@ -54,11 +57,12 @@ public class SimpleMessageGroupTests { Constructor ctr = clazz.getDeclaredConstructor(MessageGroup.class); ctr.setAccessible(true); group = ctr.newInstance(group); + new DirectFieldAccessor(group).setPropertyValue("messages", new HashSet>()); } @Test public void shouldFindSupersedingMessagesIfSequenceAware() throws Exception { - this.prepareForSequenceAwareMessageGroup(); + prepareForSequenceAwareMessageGroup(); final Message message1 = MessageBuilder.withPayload("test").setSequenceNumber(1).build(); final Message message2 = MessageBuilder.fromMessage(message1).setSequenceNumber(1).build(); assertThat(group.canAdd(message1), is(true)); @@ -69,7 +73,7 @@ public class SimpleMessageGroupTests { @Test public void shouldIgnoreMessagesWithZeroSequenceNumberIfSequenceAware() throws Exception { - this.prepareForSequenceAwareMessageGroup(); + prepareForSequenceAwareMessageGroup(); final Message message1 = MessageBuilder.withPayload("test").build(); final Message message2 = MessageBuilder.fromMessage(message1).build(); assertThat(group.canAdd(message1), is(true)); diff --git a/src/reference/asciidoc/aggregator.adoc b/src/reference/asciidoc/aggregator.adoc index b152740373..c73ed7c7b6 100644 --- a/src/reference/asciidoc/aggregator.adoc +++ b/src/reference/asciidoc/aggregator.adoc @@ -221,6 +221,35 @@ As shown above, it is also the default strategy. NOTE: Before _version 5.0_, the default release strategy was `SequenceSizeReleaseStrategy` which does not perform well with large groups. With that strategy, duplicate sequence numbers are detected and rejected; this operation can be expensive. +If you are aggregating large groups, you don't need to release partial groups, and you don't need to detect/reject duplicate sequences, consider using the `SimpleSequenceSizeReleaseStrategy` instead - it is much more efficient for these use cases, and is the default since _version 5.0_ when partial group release is not specified. + +===== Aggregating Large Groups + +The 4.3 release changed the default `Collection` for messages in a `SimpleMessageGroup` to `HashSet` (it was previously a `BlockingQueue`). +This was expensive when removing individual messages from large groups (an O(n) linear scan was required). +Although the hash set is generally much faster for removing, it can be expensive for large messages because the hash has to be calculated (on both inserts and removes). +If you have messages that are expensive to hash, consider using some other collection type. +As discussed in <>, a `SimpleMessageGroupFactory` is provided so you can select the `Collection` that best suits your needs. +You can also provide your own factory implementation to create some other `Collection>`. + +Here is an example of how to configure an aggregator with the previous implementation and a `SimpleSequenceSizeReleaseStrategy`. + +[source, xml] +---- + + + + + + + + + + + +---- + ===== CorrelationStrategy The `CorrelationStrategy` interface is defined as follows: @@ -635,9 +664,9 @@ For this purpose the `groupTimeout` option allows scheduling the `MessageGroup` [source,xml] ---- + send-partial-result-on-expiry="true" + group-timeout-expression="size() ge 2 ? 10000 : -1" + release-strategy-expression="[0].headers.sequenceNumber == [0].headers.sequenceSize"/> ---- With this example, the normal _release_ will be possible if the aggregator receives the last message in sequence as defined by the `release-strategy-expression`.