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 8413b3314d..af2952a461 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 @@ -693,13 +693,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/aggregator/SimpleSequenceSizeReleaseStrategy.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SimpleSequenceSizeReleaseStrategy.java new file mode 100644 index 0000000000..95abab264d --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/SimpleSequenceSizeReleaseStrategy.java @@ -0,0 +1,37 @@ +/* + * Copyright 2016 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 org.springframework.integration.store.MessageGroup; + +/** + * An implementation of {@link ReleaseStrategy} that simply compares the current size of + * the message list to the expected 'sequenceSize'. It does not support releasing partial + * sequences. Correlating message handlers using this strategy do not check for duplicate + * sequence numbers. + * @author Gary Russell + * @since 4.3.7 + * + */ +public class SimpleSequenceSizeReleaseStrategy implements ReleaseStrategy { + + @Override + public boolean canRelease(MessageGroup group) { + return group.getSequenceSize() == group.size(); + } + +} 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 7d86ad941b..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 @@ -66,21 +66,23 @@ public class SimpleMessageGroup implements MessageGroup { } public SimpleMessageGroup(Collection> messages, Object groupId, long timestamp, - boolean complete) { - this(new LinkedHashSet>(), messages, groupId, timestamp, complete); + boolean 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 c805c58dd7..65191ec3f6 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 94842307ec..52a5357d81 100644 --- a/src/reference/asciidoc/aggregator.adoc +++ b/src/reference/asciidoc/aggregator.adoc @@ -218,6 +218,35 @@ Spring Integration provides an out-of-the box implementation for `ReleaseStrateg This implementation consults the `SEQUENCE_NUMBER` and `SEQUENCE_SIZE` headers of each arriving message to decide when a message group is complete and ready to be aggregated. As shown above, it is also the default strategy. +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 will be the default in future releases 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: @@ -629,9 +658,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`.