From 6d7aebc65af33d35fc010782d8bb648113e67f98 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 9 May 2022 16:42:33 -0400 Subject: [PATCH] GH-3592: Scatter-Gather: applySeq=true by default Fixes https://github.com/spring-projects/spring-integration/issues/3592 * Configure XML parser & Java DSL for Scatter-Gather, based on the `RecipientListRouter` to set an `applySequence` to `true` by default. This will make a `gatherer` part to fully rely on the default correlation strategies --- .../integration/config/xml/ScatterGatherParser.java | 13 +++++++++---- .../dsl/BaseIntegrationFlowDefinition.java | 2 ++ .../integration/dsl/routers/RouterTests.java | 12 ++++-------- .../config/ScatterGatherParserTests-context.xml | 2 +- .../config/ScatterGatherTests-context.xml | 2 +- .../ScatterGatherHandlerIntegrationTests.java | 11 ++++------- src/reference/asciidoc/scatter-gather.adoc | 4 +++- src/reference/asciidoc/whats-new.adoc | 4 ++++ 8 files changed, 28 insertions(+), 22 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ScatterGatherParser.java b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ScatterGatherParser.java index dbf854d002..7831f87e1f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ScatterGatherParser.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/xml/ScatterGatherParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2020 the original author or authors. + * Copyright 2014-2022 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. @@ -89,7 +89,7 @@ public class ScatterGatherParser extends AbstractConsumerEndpointParser { builder.addConstructorArgReference(scatterChannel); } else { - BeanDefinition scattererDefinition = null; + BeanDefinition scattererDefinition; if (!hasScatterer) { scattererDefinition = new RootBeanDefinition(RecipientListRouter.class); } @@ -103,6 +103,11 @@ public class ScatterGatherParser extends AbstractConsumerEndpointParser { if (hasScatterer && scatterer.hasAttribute(ID_ATTRIBUTE)) { scattererId = scatterer.getAttribute(ID_ATTRIBUTE); } + + if (!scatterer.hasAttribute("apply-sequence")) { + scattererDefinition.getPropertyValues().addPropertyValue("applySequence", true); + } + parserContext.getRegistry().registerBeanDefinition(scattererId, scattererDefinition); // NOSONAR not null builder.addConstructorArgValue(new RuntimeBeanReference(scattererId)); } @@ -113,7 +118,7 @@ public class ScatterGatherParser extends AbstractConsumerEndpointParser { Element gatherer = DomUtils.getChildElementByTagName(element, "gatherer"); - BeanDefinition gathererDefinition = null; + BeanDefinition gathererDefinition; if (gatherer == null) { try { gatherer = DOCUMENT_BUILDER_FACTORY.newDocumentBuilder().newDocument().createElement("aggregator"); @@ -125,7 +130,7 @@ public class ScatterGatherParser extends AbstractConsumerEndpointParser { } gathererDefinition = GATHERER_PARSER.parse(gatherer, // NOSONAR new ParserContext(parserContext.getReaderContext(), - parserContext.getDelegate(), scatterGatherDefinition)); + parserContext.getDelegate(), scatterGatherDefinition)); String gathererId = id + ".gatherer"; if (gatherer != null && gatherer.hasAttribute(ID_ATTRIBUTE)) { gathererId = gatherer.getAttribute(ID_ATTRIBUTE); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java index 104de5ee29..121e1bcfb3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/BaseIntegrationFlowDefinition.java @@ -2749,6 +2749,7 @@ public abstract class BaseIntegrationFlowDefinition f .scatterGather(scatterer -> scatterer - .applySequence(true) .recipientFlow(m -> true, sf -> sf.handle((p, h) -> Math.random() * 10)) .recipientFlow(m -> true, sf -> sf.handle((p, h) -> Math.random() * 10)) .recipientFlow(m -> true, sf -> sf.handle((p, h) -> Math.random() * 10)), @@ -878,8 +877,7 @@ public class RouterTests { .scatterGather( scatterer -> scatterer .recipientFlow(f1 -> f1.handle((p, h) -> p + " - flow 1")) - .recipientFlow(f2 -> f2.handle((p, h) -> p + " - flow 2")) - .applySequence(true), + .recipientFlow(f2 -> f2.handle((p, h) -> p + " - flow 2")), gatherer -> gatherer .outputProcessor(mg -> mg .getMessages() @@ -900,7 +898,6 @@ public class RouterTests { return f -> f .scatterGather( scatterer -> scatterer - .applySequence(true) .recipientFlow(f1 -> f1.transform(p -> "Sub-flow#1")) .recipientFlow(f2 -> f2 .channel(c -> c.executor(taskExecutor)) @@ -923,7 +920,6 @@ public class RouterTests { public IntegrationFlow propagateErrorFromGatherer(TaskExecutor taskExecutor) { return IntegrationFlows.from(Function.class) .scatterGather(s -> s - .applySequence(true) .recipientFlow(subFlow -> subFlow .channel(c -> c.executor(taskExecutor)) .transform(p -> "foo")), @@ -943,9 +939,9 @@ public class RouterTests { @Bean public IntegrationFlow scatterGatherInSubFlow() { - return flow -> flow.scatterGather(s -> s.applySequence(true) + return flow -> flow.scatterGather(s -> s .recipientFlow(inflow -> inflow.wireTap(scatterGatherWireTapChannel()) - .scatterGather(s1 -> s1.applySequence(true) + .scatterGather(s1 -> s1 .recipientFlow(IntegrationFlowDefinition::bridge) .recipientFlow("sequencetest"::equals, IntegrationFlowDefinition::bridge), diff --git a/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherParserTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherParserTests-context.xml index a62e3b7de8..b7292545f0 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherParserTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherParserTests-context.xml @@ -14,7 +14,7 @@ - + diff --git a/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests-context.xml b/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests-context.xml index 5255b2eaf9..5754fce70f 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests-context.xml +++ b/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests-context.xml @@ -30,7 +30,7 @@ - + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ScatterGatherHandlerIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ScatterGatherHandlerIntegrationTests.java index e8e046e61c..6af6141a32 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ScatterGatherHandlerIntegrationTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ScatterGatherHandlerIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2019 the original author or authors. + * Copyright 2014-2022 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. @@ -22,8 +22,7 @@ import java.util.Arrays; import java.util.List; import java.util.concurrent.Executor; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; @@ -54,16 +53,14 @@ import org.springframework.messaging.PollableChannel; import org.springframework.messaging.SubscribableChannel; import org.springframework.messaging.support.MessageBuilder; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; /** * @author Artem Bilan * @author Gary Russell * @since 4.1 */ -@ContextConfiguration -@RunWith(SpringJUnit4ClassRunner.class) +@SpringJUnitConfig @DirtiesContext public class ScatterGatherHandlerIntegrationTests { diff --git a/src/reference/asciidoc/scatter-gather.adoc b/src/reference/asciidoc/scatter-gather.adoc index 66567b0f83..8923421efe 100644 --- a/src/reference/asciidoc/scatter-gather.adoc +++ b/src/reference/asciidoc/scatter-gather.adoc @@ -35,6 +35,9 @@ Unlike the `PublishSubscribeChannel` variant (the auction variant), having a `re With `apply-sequence="true"`, the default `sequenceSize` is supplied, and the `aggregator` can release the group correctly. The distribution option is mutually exclusive with the auction option. +NOTE: The `applySequence=true` is required only for plain Java configuration based on the `ScatterGatherHandler(MessageHandler scatterer, MessageHandler gatherer)` constructor configuration since the framework cannot mutate externally provided components. +For convenience, the XML and Java DSL for `Scatter-Gather` sets `applySequence` to true starting with version 6.0. + For both the auction and the distribution variants, the request (scatter) message is enriched with the `gatherResultChannel` header to wait for a reply message from the `aggregator`. By default, all suppliers should send their result to the `replyChannel` header (usually by omitting the `output-channel` from the ultimate endpoint). @@ -183,7 +186,6 @@ public IntegrationFlow scatterGatherAndExecutorChannelSubFlow(TaskExecutor taskE return f -> f .scatterGather( scatterer -> scatterer - .applySequence(true) .recipientFlow(f1 -> f1.transform(p -> "Sub-flow#1")) .recipientFlow(f2 -> f2 .channel(c -> c.executor(taskExecutor)) diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index e397c062e8..cc819f93b5 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -37,6 +37,10 @@ The messaging annotations don't require a `poller` attribute as an array of `@Po See <<./configuration.adoc#annotations,Annotation Support>> for more information. +For convenience, the XML and Java DSL for Scatter-Gather, based on the `RecipientListRouter`, now sets an `applySequence = true`, so the gatherer part can rely on the default correlation strategies. + +See <<./scatter-gather.adoc#scatter-gather,Scatter-Gather>> for more information. + [[x6.0-http]] === HTTP Changes