diff --git a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java index 539cccfe60..ffbb6ee3b3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/aggregator/AbstractAggregatingMessageGroupProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -93,22 +93,19 @@ public abstract class AbstractAggregatingMessageGroupProcessor implements Messag builder = getMessageBuilderFactory().withPayload(payload); } - return builder.copyHeadersIfAbsent(headers) - .popSequenceDetails() - .build(); + return builder.copyHeadersIfAbsent(headers); } /** * This default implementation simply returns all headers that have no conflicts among the group. An absent header * on one or more Messages within the group is not considered a conflict. Subclasses may override this method with * more advanced conflict-resolution strategies if necessary. - * * @param group The message group. * @return The aggregated headers. */ protected Map aggregateHeaders(MessageGroup group) { - Map aggregatedHeaders = new HashMap(); - Set conflictKeys = new HashSet(); + Map aggregatedHeaders = new HashMap<>(); + Set conflictKeys = new HashSet<>(); for (Message message : group.getMessages()) { for (Entry entry : message.getHeaders().entrySet()) { String key = entry.getKey(); 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 0dd39ae595..36daf2b0b4 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 @@ -51,6 +51,8 @@ import org.springframework.integration.store.MessageGroupStore; import org.springframework.integration.store.MessageStore; import org.springframework.integration.store.SimpleMessageGroup; import org.springframework.integration.store.SimpleMessageStore; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; +import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.locks.DefaultLockRegistry; import org.springframework.integration.support.locks.LockRegistry; import org.springframework.integration.util.UUIDConverter; @@ -140,6 +142,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP private boolean expireGroupsUponTimeout = true; + private boolean popSequence = true; + private volatile boolean running; public AbstractCorrelatingMessageHandler(MessageGroupProcessor processor, MessageGroupStore store, @@ -217,6 +221,69 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP this.outputProcessor = outputProcessor; } + public void setDiscardChannel(MessageChannel discardChannel) { + Assert.notNull(discardChannel, "'discardChannel' cannot be null"); + this.discardChannel = discardChannel; + } + + public void setDiscardChannelName(String discardChannelName) { + Assert.hasText(discardChannelName, "'discardChannelName' must not be empty"); + this.discardChannelName = discardChannelName; + } + + + public void setSendPartialResultOnExpiry(boolean sendPartialResultOnExpiry) { + this.sendPartialResultOnExpiry = sendPartialResultOnExpiry; + } + + /** + * By default, when a MessageGroupStoreReaper is configured to expire partial + * groups, empty groups are also removed. Empty groups exist after a group + * is released normally. This is to enable the detection and discarding of + * late-arriving messages. If you wish to expire empty groups on a longer + * schedule than expiring partial groups, set this property. Empty groups will + * then not be removed from the MessageStore until they have not been modified + * for at least this number of milliseconds. + * @param minimumTimeoutForEmptyGroups The minimum timeout. + */ + public void setMinimumTimeoutForEmptyGroups(long minimumTimeoutForEmptyGroups) { + this.minimumTimeoutForEmptyGroups = minimumTimeoutForEmptyGroups; + } + + /** + * Set {@code releasePartialSequences} on an underlying default + * {@link SequenceSizeReleaseStrategy}. Ignored for other release strategies. + * @param releasePartialSequences true to allow release. + */ + public void setReleasePartialSequences(boolean releasePartialSequences) { + if (!this.releaseStrategySet && releasePartialSequences) { + setReleaseStrategy(new SequenceSizeReleaseStrategy()); + } + this.releasePartialSequences = releasePartialSequences; + } + + /** + * Expire (completely remove) a group if it is completed due to timeout. + * Default true + * @param expireGroupsUponTimeout the expireGroupsUponTimeout to set + * @since 4.1 + */ + public void setExpireGroupsUponTimeout(boolean expireGroupsUponTimeout) { + this.expireGroupsUponTimeout = expireGroupsUponTimeout; + } + + /** + * Perform a {@link MessageBuilder#popSequenceDetails()} for output message or not. + * Default to true. + * This option removes the sequence information added by the nearest upstream component with + * {@code applySequence=true} (for example splitter). + * @param popSequence the boolean flag to use. + * @since 5.1 + */ + public void setPopSequence(boolean popSequence) { + this.popSequence = popSequence; + } + @Override public void setTaskScheduler(TaskScheduler taskScheduler) { super.setTaskScheduler(taskScheduler); @@ -285,57 +352,6 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP return processor; } - public void setDiscardChannel(MessageChannel discardChannel) { - Assert.notNull(discardChannel, "'discardChannel' cannot be null"); - this.discardChannel = discardChannel; - } - - public void setDiscardChannelName(String discardChannelName) { - Assert.hasText(discardChannelName, "'discardChannelName' must not be empty"); - this.discardChannelName = discardChannelName; - } - - - public void setSendPartialResultOnExpiry(boolean sendPartialResultOnExpiry) { - this.sendPartialResultOnExpiry = sendPartialResultOnExpiry; - } - - /** - * By default, when a MessageGroupStoreReaper is configured to expire partial - * groups, empty groups are also removed. Empty groups exist after a group - * is released normally. This is to enable the detection and discarding of - * late-arriving messages. If you wish to expire empty groups on a longer - * schedule than expiring partial groups, set this property. Empty groups will - * then not be removed from the MessageStore until they have not been modified - * for at least this number of milliseconds. - * @param minimumTimeoutForEmptyGroups The minimum timeout. - */ - public void setMinimumTimeoutForEmptyGroups(long minimumTimeoutForEmptyGroups) { - this.minimumTimeoutForEmptyGroups = minimumTimeoutForEmptyGroups; - } - - /** - * Set {@code releasePartialSequences} on an underlying default - * {@link SequenceSizeReleaseStrategy}. Ignored for other release strategies. - * @param releasePartialSequences true to allow release. - */ - public void setReleasePartialSequences(boolean releasePartialSequences) { - if (!this.releaseStrategySet && releasePartialSequences) { - setReleaseStrategy(new SequenceSizeReleaseStrategy()); - } - this.releasePartialSequences = releasePartialSequences; - } - - /** - * Expire (completely remove) a group if it is completed due to timeout. - * Default true - * @param expireGroupsUponTimeout the expireGroupsUponTimeout to set - * @since 4.1 - */ - public void setExpireGroupsUponTimeout(boolean expireGroupsUponTimeout) { - this.expireGroupsUponTimeout = expireGroupsUponTimeout; - } - @Override public String getComponentType() { return "aggregator"; @@ -413,7 +429,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP @Override protected void handleMessageInternal(Message message) throws Exception { Object correlationKey = this.correlationStrategy.getCorrelationKey(message); - Assert.state(correlationKey != null, "Null correlation not allowed. Maybe the CorrelationStrategy is failing?"); + Assert.state(correlationKey != null, + "Null correlation not allowed. Maybe the CorrelationStrategy is failing?"); if (this.logger.isDebugEnabled()) { this.logger.debug("Handling message with correlationKey [" + correlationKey + "]: " + message); @@ -653,7 +670,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP * groups. A longer timeout for empty groups can be enabled by * setting minimumTimeoutForEmptyGroups. */ - removeGroup = lastModifiedNow <= (System.currentTimeMillis() - this.minimumTimeoutForEmptyGroups); + removeGroup = + lastModifiedNow <= (System.currentTimeMillis() - this.minimumTimeoutForEmptyGroups); if (removeGroup && this.logger.isDebugEnabled()) { this.logger.debug("Removing empty group: " + correlationKey); } @@ -753,10 +771,24 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP Object result = this.outputProcessor.processMessageGroup(group); Collection> partialSequence = null; if (result instanceof Collection) { - this.verifyResultCollectionConsistsOfMessages((Collection) result); + verifyResultCollectionConsistsOfMessages((Collection) result); partialSequence = (Collection>) result; } - this.sendOutputs(result, message); + + if (this.popSequence && partialSequence == null && !(result instanceof Message)) { + AbstractIntegrationMessageBuilder messageBuilder; + if (result instanceof AbstractIntegrationMessageBuilder) { + messageBuilder = (AbstractIntegrationMessageBuilder) result; + } + else { + messageBuilder = getMessageBuilderFactory() + .withPayload(result) + .copyHeaders(message.getHeaders()); + } + result = messageBuilder.popSequenceDetails(); + } + + sendOutputs(result, message); return partialSequence; } @@ -772,10 +804,8 @@ public abstract class AbstractCorrelatingMessageHandler extends AbstractMessageP } @Override - public void destroy() throws Exception { - for (ScheduledFuture future : this.expireGroupScheduledFutures.values()) { - future.cancel(true); - } + public void destroy() { + this.expireGroupScheduledFutures.values().forEach(future -> future.cancel(true)); } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java index 0fda31af29..f913dbadd1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/AggregatorFactoryBean.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2018 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. @@ -40,6 +40,8 @@ import org.springframework.util.StringUtils; * {@link FactoryBean} to create an {@link AggregatingMessageHandler}. * * @author Gary Russell + * @author Artem Bilan + * * @since 4.2 * */ @@ -85,6 +87,8 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe private Boolean expireGroupsUponTimeout; + private Boolean popSequence; + public void setProcessorBean(Object processorBean) { this.processorBean = processorBean; } @@ -165,6 +169,10 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe this.expireGroupsUponTimeout = expireGroupsUponTimeout; } + public void setPopSequence(Boolean popSequence) { + this.popSequence = popSequence; + } + @Override protected AggregatingMessageHandler createHandler() { MessageGroupProcessor outputProcessor; @@ -253,6 +261,10 @@ public class AggregatorFactoryBean extends AbstractSimpleMessageHandlerFactoryBe aggregator.setExpireGroupsUponTimeout(this.expireGroupsUponTimeout); } + if (this.popSequence != null) { + aggregator.setPopSequence(this.popSequence); + } + return aggregator; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java index babeeb42d0..0d0b792111 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/AbstractCorrelatingMessageHandlerParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -83,6 +83,8 @@ public abstract class AbstractCorrelatingMessageHandlerParser extends AbstractCo IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "empty-group-min-timeout", "minimumTimeoutForEmptyGroups"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "pop-sequence"); + BeanDefinition expressionDef = IntegrationNamespaceUtils.createExpressionDefinitionFromValueOrExpression("group-timeout", "group-timeout-expression", parserContext, element, false); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java index a050abfe68..c4a4010290 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/CorrelationHandlerSpec.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2018 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. @@ -34,6 +34,7 @@ import org.springframework.integration.expression.FunctionExpression; import org.springframework.integration.expression.ValueExpression; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; +import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.locks.LockRegistry; import org.springframework.messaging.MessageChannel; import org.springframework.scheduling.TaskScheduler; @@ -53,7 +54,7 @@ public abstract class CorrelationHandlerSpec extends ConsumerEndpointSpec { - private final List forceReleaseAdviceChain = new LinkedList(); + private final List forceReleaseAdviceChain = new LinkedList<>(); protected CorrelationHandlerSpec(H messageHandler) { super(messageHandler); @@ -314,4 +315,16 @@ public abstract class CorrelationHandlerSpec + + + + Boolean flag specifying, if the 'MessageBuilder.popSequenceDetails()' should be called + for the output message. Plays the opposite role to the + 'AbstractMessageSplitter.setApplySequence()'. + Defaults to 'true'. + + + + + + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java index a9224dd601..87ed57f7fa 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AbstractCorrelatingMessageHandlerTests.java @@ -38,6 +38,7 @@ import java.util.concurrent.TimeUnit; import org.junit.Test; import org.springframework.beans.DirectFieldAccessor; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupStore; @@ -150,7 +151,7 @@ public class AbstractCorrelatingMessageHandlerTests { } @Test // INT-2833 - public void testReaperReapsAnEmptyGroup() throws Exception { + public void testReaperReapsAnEmptyGroup() { final MessageGroupStore groupStore = new SimpleMessageStore(); AggregatingMessageHandler handler = new AggregatingMessageHandler(group -> group, groupStore); @@ -447,4 +448,32 @@ public class AbstractCorrelatingMessageHandlerTests { assertEquals(1, handler2DiscardChannel.getQueueSize()); } + @Test + public void testNoPopSequenceDetails() { + MessageGroupProcessor mgp = new DefaultAggregatingMessageGroupProcessor(); + AggregatingMessageHandler handler = new AggregatingMessageHandler(mgp); + handler.setReleaseStrategy(group -> true); + handler.setPopSequence(false); + QueueChannel outputChannel = new QueueChannel(); + handler.setOutputChannel(outputChannel); + + Message testMessage = MessageBuilder.withPayload("foo") + .setCorrelationId(1) + .setSequenceNumber(1) + .setSequenceSize(1) + .pushSequenceDetails(2, 2, 2) + .build(); + + handler.handleMessage(testMessage); + + Message receive = outputChannel.receive(10_000); + + assertNotNull(receive); + + assertEquals(2, receive.getHeaders().get(IntegrationMessageHeaderAccessor.CORRELATION_ID)); + assertEquals(2, receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER)); + assertEquals(2, receive.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE)); + assertTrue(receive.getHeaders().containsKey(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS)); + } + } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java index 471f415d93..0460340ee7 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/AggregatingMessageGroupProcessorHeaderTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -35,6 +35,7 @@ import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.SimpleMessageGroup; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.MessageBuilder; import org.springframework.integration.support.MutableMessageBuilder; import org.springframework.integration.support.MutableMessageBuilderFactory; @@ -43,11 +44,13 @@ import org.springframework.messaging.Message; /** * @author Mark Fisher * @author Artem Bilan + * * @since 2.0 */ public class AggregatingMessageGroupProcessorHeaderTests { - private final DefaultAggregatingMessageGroupProcessor defaultProcessor = new DefaultAggregatingMessageGroupProcessor(); + private final DefaultAggregatingMessageGroupProcessor defaultProcessor = + new DefaultAggregatingMessageGroupProcessor(); private final MethodInvokingMessageGroupProcessor methodInvokingProcessor = new MethodInvokingMessageGroupProcessor(new TestAggregatorBean(), "aggregate"); @@ -61,52 +64,52 @@ public class AggregatingMessageGroupProcessorHeaderTests { @Test public void singleMessageUsingDefaultProcessor() { - this.singleMessage(defaultProcessor); + singleMessage(this.defaultProcessor); } @Test public void singleMessageUsingMethodInvokingProcessor() { - this.singleMessage(methodInvokingProcessor); + singleMessage(this.methodInvokingProcessor); } @Test public void twoMessagesWithoutConflictsUsingDefaultProcessor() { - this.twoMessagesWithoutConflicts(defaultProcessor); + twoMessagesWithoutConflicts(this.defaultProcessor); } @Test public void twoMessagesWithoutConflictsUsingMethodInvokingProcessor() { - this.twoMessagesWithoutConflicts(methodInvokingProcessor); + twoMessagesWithoutConflicts(this.methodInvokingProcessor); } @Test public void twoMessagesWithConflictsUsingDefaultProcessor() { - this.twoMessagesWithConflicts(defaultProcessor); + twoMessagesWithConflicts(this.defaultProcessor); } @Test public void twoMessagesWithConflictsUsingMethodInvokingProcessor() { - this.twoMessagesWithConflicts(methodInvokingProcessor); + twoMessagesWithConflicts(this.methodInvokingProcessor); } @Test public void missingValuesDoNotConflictUsingDefaultProcessor() { - this.missingValuesDoNotConflict(defaultProcessor); + missingValuesDoNotConflict(this.defaultProcessor); } @Test public void missingValuesDoNotConflictUsingMethodInvokingProcessor() { - this.missingValuesDoNotConflict(methodInvokingProcessor); + missingValuesDoNotConflict(this.methodInvokingProcessor); } @Test public void multipleValuesConflictUsingDefaultProcessor() { - this.multipleValuesConflict(defaultProcessor); + multipleValuesConflict(this.defaultProcessor); } @Test public void multipleValuesConflictUsingMethodInvokingProcessor() { - this.multipleValuesConflict(methodInvokingProcessor); + multipleValuesConflict(this.methodInvokingProcessor); } @Test @@ -114,7 +117,7 @@ public class AggregatingMessageGroupProcessorHeaderTests { DefaultAggregatingMessageGroupProcessor processor = new DefaultAggregatingMessageGroupProcessor(); DirectFieldAccessor dfa = new DirectFieldAccessor(processor); dfa.setPropertyValue("messageBuilderFactory", new MutableMessageBuilderFactory()); - Map headers1 = new HashMap(); + Map headers1 = new HashMap<>(); headers1.put("k1", "foo"); headers1.put("k2", null); Message message1 = MutableMessageBuilder.withPayload("test") @@ -123,24 +126,24 @@ public class AggregatingMessageGroupProcessorHeaderTests { .setSequenceSize(2) .copyHeadersIfAbsent(headers1) .build(); - Map headers2 = new HashMap(); + Map headers2 = new HashMap<>(); headers2.put("k1", "bar"); headers2.put("k2", 123); Message message2 = correlatedMessage(1, 2, 2, headers2); - List> messages = Arrays.>asList(message1, message2); + List> messages = Arrays.asList(message1, message2); MessageGroup group = new SimpleMessageGroup(messages, 1); Object result = processor.processMessageGroup(group); assertNotNull(result); - assertTrue(result instanceof Message); - Message resultMessage = (Message) result; + assertTrue(result instanceof AbstractIntegrationMessageBuilder); + Message resultMessage = ((AbstractIntegrationMessageBuilder) result).build(); assertNull(resultMessage.getHeaders().get("k1")); assertNull(resultMessage.getHeaders().get("k2")); - headers1 = new HashMap(); + headers1 = new HashMap<>(); headers1.put("k1", "foo"); headers1.put("k2", 123); message1 = correlatedMessage(1, 2, 1, headers1); - headers2 = new HashMap(); + headers2 = new HashMap<>(); headers2.put("k1", "bar"); headers2.put("k2", null); message2 = MutableMessageBuilder.withPayload("test") @@ -149,92 +152,92 @@ public class AggregatingMessageGroupProcessorHeaderTests { .setSequenceSize(2) .copyHeadersIfAbsent(headers2) .build(); - messages = Arrays.>asList(message1, message2); + messages = Arrays.asList(message1, message2); group = new SimpleMessageGroup(messages, 1); result = processor.processMessageGroup(group); - resultMessage = (Message) result; + resultMessage = ((AbstractIntegrationMessageBuilder) result).build(); assertNull(resultMessage.getHeaders().get("k1")); assertNull(resultMessage.getHeaders().get("k2")); } private void singleMessage(MessageGroupProcessor processor) { - Map headers = new HashMap(); + Map headers = new HashMap<>(); headers.put("k1", "value1"); - headers.put("k2", new Integer(2)); + headers.put("k2", 2); Message message = correlatedMessage(1, 1, 1, headers); - List> messages = Collections.>singletonList(message); + List> messages = Collections.singletonList(message); MessageGroup group = new SimpleMessageGroup(messages, 1); Object result = processor.processMessageGroup(group); assertNotNull(result); - assertTrue(result instanceof Message); - Message resultMessage = (Message) result; + assertTrue(result instanceof AbstractIntegrationMessageBuilder); + Message resultMessage = ((AbstractIntegrationMessageBuilder) result).build(); assertEquals("value1", resultMessage.getHeaders().get("k1")); assertEquals(2, resultMessage.getHeaders().get("k2")); } private void twoMessagesWithoutConflicts(MessageGroupProcessor processor) { - Map headers = new HashMap(); + Map headers = new HashMap<>(); headers.put("k1", "value1"); - headers.put("k2", new Integer(2)); + headers.put("k2", 2); Message message1 = correlatedMessage(1, 2, 1, headers); Message message2 = correlatedMessage(1, 2, 2, headers); - List> messages = Arrays.>asList(message1, message2); + List> messages = Arrays.asList(message1, message2); MessageGroup group = new SimpleMessageGroup(messages, 1); Object result = processor.processMessageGroup(group); assertNotNull(result); - assertTrue(result instanceof Message); - Message resultMessage = (Message) result; + assertTrue(result instanceof AbstractIntegrationMessageBuilder); + Message resultMessage = ((AbstractIntegrationMessageBuilder) result).build(); assertEquals("value1", resultMessage.getHeaders().get("k1")); assertEquals(2, resultMessage.getHeaders().get("k2")); } private void twoMessagesWithConflicts(MessageGroupProcessor processor) { - Map headers1 = new HashMap(); + Map headers1 = new HashMap<>(); headers1.put("k1", "foo"); - headers1.put("k2", new Integer(123)); + headers1.put("k2", 123); Message message1 = correlatedMessage(1, 2, 1, headers1); - Map headers2 = new HashMap(); + Map headers2 = new HashMap<>(); headers2.put("k1", "bar"); - headers2.put("k2", new Integer(123)); + headers2.put("k2", 123); Message message2 = correlatedMessage(1, 2, 2, headers2); - List> messages = Arrays.>asList(message1, message2); + List> messages = Arrays.asList(message1, message2); MessageGroup group = new SimpleMessageGroup(messages, 1); Object result = processor.processMessageGroup(group); assertNotNull(result); - assertTrue(result instanceof Message); - Message resultMessage = (Message) result; + assertTrue(result instanceof AbstractIntegrationMessageBuilder); + Message resultMessage = ((AbstractIntegrationMessageBuilder) result).build(); assertNull(resultMessage.getHeaders().get("k1")); assertEquals(123, resultMessage.getHeaders().get("k2")); } private void missingValuesDoNotConflict(MessageGroupProcessor processor) { - Map headers1 = new HashMap(); + Map headers1 = new HashMap<>(); headers1.put("only1", "value1"); headers1.put("commonTo1And2", "foo"); - headers1.put("commonToAll", new Integer(123)); + headers1.put("commonToAll", 123); headers1.put("conflictBetween1And2", "valueFor1"); Message message1 = correlatedMessage(1, 3, 1, headers1); - Map headers2 = new HashMap(); + Map headers2 = new HashMap<>(); headers2.put("only2", "value2"); headers2.put("commonTo1And2", "foo"); headers2.put("commonTo2And3", "bar"); headers2.put("conflictBetween1And2", "valueFor2"); headers2.put("conflictBetween2And3", "valueFor2"); - headers2.put("commonToAll", new Integer(123)); + headers2.put("commonToAll", 123); Message message2 = correlatedMessage(1, 3, 2, headers2); - Map headers3 = new HashMap(); + Map headers3 = new HashMap<>(); headers3.put("only3", "value3"); headers3.put("commonTo2And3", "bar"); - headers3.put("commonToAll", new Integer(123)); + headers3.put("commonToAll", 123); headers3.put("conflictBetween2And3", "valueFor3"); Message message3 = correlatedMessage(1, 3, 3, headers3); - List> messages = Arrays.>asList(message1, message2, message3); + List> messages = Arrays.asList(message1, message2, message3); MessageGroup group = new SimpleMessageGroup(messages, 1); Object result = processor.processMessageGroup(group); assertNotNull(result); - assertTrue(result instanceof Message); - Message resultMessage = (Message) result; + assertTrue(result instanceof AbstractIntegrationMessageBuilder); + Message resultMessage = ((AbstractIntegrationMessageBuilder) result).build(); assertEquals("value1", resultMessage.getHeaders().get("only1")); assertEquals("value2", resultMessage.getHeaders().get("only2")); assertEquals("value3", resultMessage.getHeaders().get("only3")); @@ -246,30 +249,31 @@ public class AggregatingMessageGroupProcessorHeaderTests { } private void multipleValuesConflict(MessageGroupProcessor processor) { - Map headers1 = new HashMap(); + Map headers1 = new HashMap<>(); headers1.put("common", "valueForAll"); headers1.put("conflict", "valueFor1"); Message message1 = correlatedMessage(1, 3, 1, headers1); - Map headers2 = new HashMap(); + Map headers2 = new HashMap<>(); headers2.put("common", "valueForAll"); headers2.put("conflict", "valueFor2"); Message message2 = correlatedMessage(1, 3, 2, headers2); - Map headers3 = new HashMap(); + Map headers3 = new HashMap<>(); headers3.put("conflict", "valueFor3"); headers3.put("common", "valueForAll"); Message message3 = correlatedMessage(1, 3, 3, headers3); - List> messages = Arrays.>asList(message1, message2, message3); + List> messages = Arrays.asList(message1, message2, message3); MessageGroup group = new SimpleMessageGroup(messages, 1); Object result = processor.processMessageGroup(group); assertNotNull(result); - assertTrue(result instanceof Message); - Message resultMessage = (Message) result; + assertTrue(result instanceof AbstractIntegrationMessageBuilder); + Message resultMessage = ((AbstractIntegrationMessageBuilder) result).build(); assertEquals("valueForAll", resultMessage.getHeaders().get("common")); assertNull(resultMessage.getHeaders().get("conflict")); } private static Message correlatedMessage(Object correlationId, Integer sequenceSize, Integer sequenceNumber, Map headers) { + return MessageBuilder.withPayload("test") .setCorrelationId(correlationId) .setSequenceNumber(sequenceNumber) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java index b9637ed30d..2443118ee5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/ExpressionEvaluatingMessageGroupProcessorTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2017 the original author or authors. + * Copyright 2002-2018 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. @@ -33,6 +33,7 @@ import org.mockito.junit.MockitoJUnitRunner; import org.springframework.beans.factory.BeanFactory; import org.springframework.integration.store.MessageGroup; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; @@ -40,6 +41,7 @@ import org.springframework.messaging.Message; * @author Alex Peters * @author Mark Fisher * @author Gary Russell + * @author Artem Bilan */ @RunWith(MockitoJUnitRunner.class) public class ExpressionEvaluatingMessageGroupProcessorTests { @@ -49,7 +51,7 @@ public class ExpressionEvaluatingMessageGroupProcessorTests { @Mock private MessageGroup group; - private final List> messages = new ArrayList>(); + private final List> messages = new ArrayList<>(); @Before @@ -62,36 +64,36 @@ public class ExpressionEvaluatingMessageGroupProcessorTests { @Test - public void testProcessAndSendWithSizeExpressionEvaluated() throws Exception { + public void testProcessAndSendWithSizeExpressionEvaluated() { when(group.getMessages()).thenReturn(messages); processor = new ExpressionEvaluatingMessageGroupProcessor("#root.size()"); processor.setBeanFactory(mock(BeanFactory.class)); Object result = processor.processMessageGroup(group); - assertTrue(result instanceof Message); - Message resultMessage = (Message) result; + assertTrue(result instanceof AbstractIntegrationMessageBuilder); + Message resultMessage = ((AbstractIntegrationMessageBuilder) result).build(); assertEquals(5, resultMessage.getPayload()); } @Test - public void testProcessAndCheckHeaders() throws Exception { + public void testProcessAndCheckHeaders() { when(group.getMessages()).thenReturn(messages); processor = new ExpressionEvaluatingMessageGroupProcessor("#root"); processor.setBeanFactory(mock(BeanFactory.class)); Object result = processor.processMessageGroup(group); processor.setBeanFactory(mock(BeanFactory.class)); - assertTrue(result instanceof Message); - Message resultMessage = (Message) result; + assertTrue(result instanceof AbstractIntegrationMessageBuilder); + Message resultMessage = ((AbstractIntegrationMessageBuilder) result).build(); assertEquals("bar", resultMessage.getHeaders().get("foo")); } @Test - public void testProcessAndSendWithProjectionExpressionEvaluated() throws Exception { + public void testProcessAndSendWithProjectionExpressionEvaluated() { when(group.getMessages()).thenReturn(messages); processor = new ExpressionEvaluatingMessageGroupProcessor("![payload]"); processor.setBeanFactory(mock(BeanFactory.class)); Object result = processor.processMessageGroup(group); - assertTrue(result instanceof Message); - Message resultMessage = (Message) result; + assertTrue(result instanceof AbstractIntegrationMessageBuilder); + Message resultMessage = ((AbstractIntegrationMessageBuilder) result).build(); assertTrue(resultMessage.getPayload() instanceof Collection); Collection list = (Collection) resultMessage.getPayload(); assertEquals(5, list.size()); @@ -103,13 +105,13 @@ public class ExpressionEvaluatingMessageGroupProcessorTests { } @Test - public void testProcessAndSendWithFilterAndProjectionExpressionEvaluated() throws Exception { + public void testProcessAndSendWithFilterAndProjectionExpressionEvaluated() { when(group.getMessages()).thenReturn(messages); processor = new ExpressionEvaluatingMessageGroupProcessor("?[payload>2].![payload]"); processor.setBeanFactory(mock(BeanFactory.class)); Object result = processor.processMessageGroup(group); - assertTrue(result instanceof Message); - Message resultMessage = (Message) result; + assertTrue(result instanceof AbstractIntegrationMessageBuilder); + Message resultMessage = ((AbstractIntegrationMessageBuilder) result).build(); assertTrue(resultMessage.getPayload() instanceof Collection); Collection list = (Collection) resultMessage.getPayload(); assertEquals(3, list.size()); @@ -119,14 +121,14 @@ public class ExpressionEvaluatingMessageGroupProcessorTests { } @Test - public void testProcessAndSendWithFilterAndProjectionAndMethodInvokingExpressionEvaluated() throws Exception { + public void testProcessAndSendWithFilterAndProjectionAndMethodInvokingExpressionEvaluated() { when(group.getMessages()).thenReturn(messages); processor = new ExpressionEvaluatingMessageGroupProcessor(String.format("T(%s).sum(?[payload>2].![payload])", getClass().getName())); processor.setBeanFactory(mock(BeanFactory.class)); Object result = processor.processMessageGroup(group); - assertTrue(result instanceof Message); - Message resultMessage = (Message) result; + assertTrue(result instanceof AbstractIntegrationMessageBuilder); + Message resultMessage = ((AbstractIntegrationMessageBuilder) result).build(); assertEquals(3 + 4 + 5, resultMessage.getPayload()); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java index 271c006722..a04e636137 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/aggregator/MethodInvokingMessageGroupProcessorTests.java @@ -47,6 +47,7 @@ import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.SimpleMessageGroup; +import org.springframework.integration.support.AbstractIntegrationMessageBuilder; import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.handler.annotation.Header; @@ -80,7 +81,7 @@ public class MethodInvokingMessageGroupProcessorTests { @Test - public void shouldFindAnnotatedAggregatorMethod() throws Exception { + public void shouldFindAnnotatedAggregatorMethod() { @SuppressWarnings("unused") class AnnotatedAggregatorMethod { @@ -102,7 +103,7 @@ public class MethodInvokingMessageGroupProcessorTests { MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new AnnotatedAggregatorMethod()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat(((Message) result).getPayload(), is(7)); + assertThat(((AbstractIntegrationMessageBuilder) result).build().getPayload(), is(7)); } @Test @@ -123,7 +124,7 @@ public class MethodInvokingMessageGroupProcessorTests { MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat(((Message) result).getPayload(), is(7)); + assertThat(((AbstractIntegrationMessageBuilder) result).build().getPayload(), is(7)); } @Test @@ -144,7 +145,7 @@ public class MethodInvokingMessageGroupProcessorTests { MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat(((Message) result).getPayload(), is(7)); + assertThat(((AbstractIntegrationMessageBuilder) result).build().getPayload(), is(7)); } @Test @@ -169,7 +170,7 @@ public class MethodInvokingMessageGroupProcessorTests { messagesUpForProcessing.add(MessageBuilder.withPayload(3).setHeader("foo", Arrays.asList(101, 102)).build()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat(((Message) result).getPayload(), is("[1, 2, 4, 3, 101, 102]")); + assertThat(((AbstractIntegrationMessageBuilder) result).build().getPayload(), is("[1, 2, 4, 3, 101, 102]")); } @Test @@ -196,7 +197,7 @@ public class MethodInvokingMessageGroupProcessorTests { messagesUpForProcessing.add(MessageBuilder.withPayload(3).setHeader("foo", Arrays.asList(101, 102)).build()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat(((Message) result).getPayload(), is("[1, 2, 4, 3, 101, 102]")); + assertThat(((AbstractIntegrationMessageBuilder) result).build().getPayload(), is("[1, 2, 4, 3, 101, 102]")); } @Test @@ -222,7 +223,7 @@ public class MethodInvokingMessageGroupProcessorTests { MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat(((Message) result).getPayload(), is("[1, 2, 4]")); + assertThat(((AbstractIntegrationMessageBuilder) result).build().getPayload(), is("[1, 2, 4]")); } @Test @@ -243,7 +244,7 @@ public class MethodInvokingMessageGroupProcessorTests { MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat(((Message) result).getPayload(), is(7)); + assertThat(((AbstractIntegrationMessageBuilder) result).build().getPayload(), is(7)); } @Test @@ -264,7 +265,7 @@ public class MethodInvokingMessageGroupProcessorTests { MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new SimpleAggregator()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat(((Message) result).getPayload(), is(7)); + assertThat(((AbstractIntegrationMessageBuilder) result).build().getPayload(), is(7)); } @@ -296,7 +297,7 @@ public class MethodInvokingMessageGroupProcessorTests { processor.setConversionService(conversionService); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat(((Message) result).getPayload(), is(7)); + assertThat(((AbstractIntegrationMessageBuilder) result).build().getPayload(), is(7)); } @Test @@ -326,7 +327,7 @@ public class MethodInvokingMessageGroupProcessorTests { MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new UnannotatedAggregator()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertThat(((Message) result).getPayload(), is(7)); + assertThat(((AbstractIntegrationMessageBuilder) result).build().getPayload(), is(7)); } @Test @@ -353,7 +354,7 @@ public class MethodInvokingMessageGroupProcessorTests { MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new UnannotatedAggregator()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - assertTrue(((Message) result).getPayload() instanceof Iterator); + assertTrue(((AbstractIntegrationMessageBuilder) result).build().getPayload() instanceof Iterator); } @Test @@ -380,7 +381,7 @@ public class MethodInvokingMessageGroupProcessorTests { MessageGroupProcessor processor = new MethodInvokingMessageGroupProcessor(new AnnotatedParametersAggregator()); when(messageGroupMock.getMessages()).thenReturn(messagesUpForProcessing); Object result = processor.processMessageGroup(messageGroupMock); - Object payload = ((Message) result).getPayload(); + Object payload = ((AbstractIntegrationMessageBuilder) result).build().getPayload(); assertTrue(payload instanceof Integer); assertEquals(7, payload); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests-context.xml index 0f4a3d717a..1d988602c6 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests-context.xml @@ -46,6 +46,7 @@ lock-registry="lockRegistry" scheduler="scheduler" message-store="store" + pop-sequence="false" order="5"> diff --git a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java index 6bb8288879..160057c17a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/config/AggregatorParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2016 the original author or authors. + * Copyright 2002-2018 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. @@ -85,7 +85,7 @@ public class AggregatorParserTests { public void testAggregation() { MessageChannel input = (MessageChannel) context.getBean("aggregatorWithReferenceInput"); TestAggregatorBean aggregatorBean = (TestAggregatorBean) context.getBean("aggregatorBean"); - List> outboundMessages = new ArrayList>(); + List> outboundMessages = new ArrayList<>(); outboundMessages.add(createMessage("123", "id1", 3, 1, null)); outboundMessages.add(createMessage("789", "id1", 3, 3, null)); outboundMessages.add(createMessage("456", "id1", 3, 2, null)); @@ -106,7 +106,7 @@ public class AggregatorParserTests { QueueChannel output = this.context.getBean("outputChannel", QueueChannel.class); output.purge(null); MessageChannel input = (MessageChannel) context.getBean("aggregatorWithMGPReferenceInput"); - List> outboundMessages = new ArrayList>(); + List> outboundMessages = new ArrayList<>(); outboundMessages.add(createMessage("123", "id1", 3, 1, null)); outboundMessages.add(createMessage("789", "id1", 3, 3, null)); outboundMessages.add(createMessage("456", "id1", 3, 2, null)); @@ -122,7 +122,7 @@ public class AggregatorParserTests { QueueChannel output = this.context.getBean("outputChannel", QueueChannel.class); output.purge(null); MessageChannel input = (MessageChannel) context.getBean("aggregatorWithCustomMGPReferenceInput"); - List> outboundMessages = new ArrayList>(); + List> outboundMessages = new ArrayList<>(); outboundMessages.add(createMessage("123", "id1", 3, 1, null)); outboundMessages.add(createMessage("789", "id1", 3, 3, null)); outboundMessages.add(createMessage("456", "id1", 3, 2, null)); @@ -137,9 +137,9 @@ public class AggregatorParserTests { public void testAggregationByExpression() { MessageChannel input = (MessageChannel) context.getBean("aggregatorWithExpressionsInput"); SubscribableChannel outputChannel = (SubscribableChannel) context.getBean("aggregatorWithExpressionsOutput"); - final AtomicReference> aggregatedMessage = new AtomicReference>(); + final AtomicReference> aggregatedMessage = new AtomicReference<>(); outputChannel.subscribe(aggregatedMessage::set); - List> outboundMessages = new ArrayList>(); + List> outboundMessages = new ArrayList<>(); outboundMessages.add(MessageBuilder.withPayload("123").setHeader("foo", "1").build()); outboundMessages.add(MessageBuilder.withPayload("456").setHeader("foo", "1").build()); outboundMessages.add(MessageBuilder.withPayload("789").setHeader("foo", "1").build()); @@ -155,7 +155,7 @@ public class AggregatorParserTests { } @Test - public void testPropertyAssignment() throws Exception { + public void testPropertyAssignment() { EventDrivenConsumer endpoint = (EventDrivenConsumer) context.getBean("completelyDefinedAggregator"); ReleaseStrategy releaseStrategy = (ReleaseStrategy) context.getBean("releaseStrategy"); CorrelationStrategy correlationStrategy = (CorrelationStrategy) context.getBean("correlationStrategy"); @@ -195,12 +195,12 @@ public class AggregatorParserTests { assertSame(this.context.getBean("store"), TestUtils.getPropertyValue(consumer, "messageStore")); assertEquals(5, TestUtils.getPropertyValue(consumer, "order")); assertNotNull(TestUtils.getPropertyValue(consumer, "forceReleaseAdviceChain")); - + assertFalse(TestUtils.getPropertyValue(consumer, "popSequence", Boolean.class)); } @Test public void testSimpleJavaBeanAggregator() { - List> outboundMessages = new ArrayList>(); + List> outboundMessages = new ArrayList<>(); MessageChannel input = (MessageChannel) context.getBean("aggregatorWithReferenceAndMethodInput"); outboundMessages.add(createMessage(1L, "id1", 3, 1, null)); outboundMessages.add(createMessage(2L, "id1", 3, 3, null)); diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java index e0da803d39..2c290d3c79 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java @@ -29,6 +29,7 @@ import static org.junit.Assert.fail; import java.util.Arrays; import java.util.List; import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; import org.junit.Test; import org.junit.runner.RunWith; @@ -38,6 +39,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.annotation.Router; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; @@ -531,6 +533,40 @@ public class RouterTests { assertNull(this.messageHandlingExceptionChannel.receive(0)); } + @Autowired + @Qualifier("nestedScatterGatherFlow.input") + private MessageChannel nestedScatterGatherFlowInput; + + @Test + @SuppressWarnings("unchecked") + public void testNestedScatterGather() { + QueueChannel replyChannel = new QueueChannel(); + Message request = MessageBuilder.withPayload("this is a test") + .setReplyChannel(replyChannel) + .build(); + this.nestedScatterGatherFlowInput.send(request); + Message bestQuoteMessage = replyChannel.receive(10000); + assertNotNull(bestQuoteMessage); + Object payload = bestQuoteMessage.getPayload(); + assertThat(payload, instanceOf(String.class)); + List topSequenceDetails = + (List) bestQuoteMessage.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS, List.class) + .get(0); + + assertEquals(request.getHeaders().getId(), + bestQuoteMessage.getHeaders().get(IntegrationMessageHeaderAccessor.CORRELATION_ID)); + + assertEquals(bestQuoteMessage.getHeaders().get(IntegrationMessageHeaderAccessor.CORRELATION_ID), + topSequenceDetails.get(0)); + + assertEquals(bestQuoteMessage.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_NUMBER), + topSequenceDetails.get(1)); + + assertEquals(bestQuoteMessage.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_SIZE), + topSequenceDetails.get(2)); + } + + @Configuration @EnableIntegration @EnableMessageHistory({ "recipientListOrder*", "recipient1*", "recipient2*" }) @@ -773,6 +809,29 @@ public class RouterTests { .gatherTimeout(10_000)); } + @Bean + public IntegrationFlow nestedScatterGatherFlow() { + return f -> f + .split(s -> s.delimiters(" ")) + .scatterGather( + scatterer -> scatterer + .recipientFlow(f1 -> f1.handle((p, h) -> p + " - flow 1")) + .recipientFlow(f2 -> f2.handle((p, h) -> p + " - flow 2")) + .applySequence(true), + gatherer -> gatherer + .outputProcessor(mg -> mg + .getMessages() + .stream() + .map(m -> m.getPayload().toString()) + .collect(Collectors.joining(", "))), + scatterGather -> scatterGather.gatherTimeout(10_000)) + .aggregate() + ., String>transform(source -> + source.stream() + .map(s -> "- " + s) + .collect(Collectors.joining("\n"))); + } + } private static class RoutingTestBean { diff --git a/src/reference/asciidoc/aggregator.adoc b/src/reference/asciidoc/aggregator.adoc index 0921a97f5f..a406863642 100644 --- a/src/reference/asciidoc/aggregator.adoc +++ b/src/reference/asciidoc/aggregator.adoc @@ -103,6 +103,13 @@ This method is invoked for aggregating messages as follows: NOTE: In the interest of code simplicity and promoting best practices such as low coupling, testability, and others, the preferred way of implementing the aggregation logic is through a POJO and using the XML or annotation support for configuring it in the application. +Starting with version 5.1, after processing message group, an `AbstractCorrelatingMessageHandler` performs a `MessageBuilder.popSequenceDetails()` message headers modification for the proper splitter-aggregator scenario with several nested levels. +It is done only if the message group release result is not a message or collection of messages. +In that case a target `MessageGroupProcessor` is responsible for the `MessageBuilder.popSequenceDetails()` call while building those messages. +This functionality can be controlled by a new `popSequence` `boolean` property, so the `MessageBuilder.popSequenceDetails()` can be disabled in some scenarios when correlation details have not been populated by the standard splitter. +This property, essentially, undoes what has been done by the nearest upstream `applySequence = true` in the `AbstractMessageSplitter`. +See <> for more information. + [[agg-message-collection]] IMPORTANT: The `SimpleMessageGroup.getMessages()` method returns an `unmodifiableCollection`. Therefore, if your aggregating POJO method has a `Collection` parameter, the argument passed in is exactly that `Collection` instance and, when you use a `SimpleMessageStore` for the aggregator, that original `Collection` is cleared after releasing the group. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 3d7625b46a..90a404971f 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -107,6 +107,9 @@ See <> for more information. If the `groupTimeout` is evaluated to a negative value, an aggregator now expires the group immediately. Only `null` is considered as a signal to do nothing for the current message. +A new `popSequence` property has been introduced to allow (by default) to call a `MessageBuilder.popSequenceDetails()` for the output message. +Also an `AbstractAggregatingMessageGroupProcessor` returns now an `AbstractIntegrationMessageBuilder` instead of the whole `Message` for optimization. + See <> for more information. [[x5.1-publisher]]