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 a1e05f201e..b9d8e56594 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 @@ -111,6 +111,7 @@ public class ScatterGatherParser extends AbstractConsumerEndpointParser { IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "gather-channel"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "gather-timeout"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "requires-reply"); return builder; } diff --git a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd index 578cb85d84..6e77bd235a 100644 --- a/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd +++ b/spring-integration-core/src/main/resources/org/springframework/integration/config/xml/spring-integration-4.1.xsd @@ -4296,8 +4296,16 @@ The list of component name patterns you want to track (e.g., tracked-components Allows to specify how long the Scatter-Gather will wait for reply Messages for gathering. By default it will wait indefinitely. Value is specified in milliseconds. - It will be applied only if the 'gather-channel' is specified and it is some blocking channel, - e.g. 'QueueChannel'. + + + + + + + Specify whether the Scatter-Gather must return a non-null value. This value is + 'true' by default, hence a ReplyRequiredException will be thrown when + the underlying aggregator returns a null value after 'gather-timeout'. + Note, if 'null' is a possibility, the 'gather-timeout' should be specified to avoid an indefinite wait. diff --git a/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherParserTests.java b/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherParserTests.java index ad5f876bcc..f9a7973c0b 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherParserTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherParserTests.java @@ -64,6 +64,7 @@ public class ScatterGatherParserTests { Object reaper = this.beanFactory.getBean("reaper"); assertSame(gatherer.getMessageStore(), TestUtils.getPropertyValue(reaper, "messageGroupStore")); + assertTrue(TestUtils.getPropertyValue(scatterGather, "requiresReply", Boolean.class)); } @Test diff --git a/src/reference/docbook/scatter-gather.xml b/src/reference/docbook/scatter-gather.xml index 52b603d3d5..1eee2cad04 100644 --- a/src/reference/docbook/scatter-gather.xml +++ b/src/reference/docbook/scatter-gather.xml @@ -5,22 +5,235 @@
Introduction - TBD + Starting with version 4.1, Spring Integration provides an implementation + of the Scatter-Gather + Enterprise Integration Pattern. It is a compound endpoint, where the goal is to send a message + to the recipients and aggregate the results. Quoting the EIP Book, it is a component for scenarios like + best quote, when we need to request information from several suppliers + and decide which one provides us with the best term for the requested item. + + + Previously, the pattern could be configured using discrete components, this enhancement brings + more convenient configuration. + + + The ScatterGatherHandler is a request-reply endpoint + that combines a + PublishSubscribeChannel (or RecipientListRouter) + and an AggregatingMessageHandler. The request message is sent to the + scatter channel and the ScatterGatherHandler waits for the reply + from the aggregator to sends to the outputChannel.
Functionality - TBD + The Scatter-Gather pattern suggests two scenarios - Auction and + Distribution. In both cases, the aggregation function is the same and + provides all options available for the AggregatingMessageHandler. Actually the + ScatterGatherHandler just requires an AggregatingMessageHandler + as a constructor argument. See for more information. + + Auction + + The Auction Scatter-Gather variant uses + publish-subscribe logic for the request message, where the + scatter channel is a PublishSubscribeChannel with + apply-sequence="true". However, this channel can be any + MessageChannel implementation as is the case with the request-channel + in the ContentEnricher (see ) but, in this case, the end-user + should support his own custom correlationStrategy for the aggregation function. + + Distribution + + The Distribution Scatter-Gather variant is based on the + RecipientListRouter (see ) + with all available options for the RecipientListRouter. This is the second + ScatterGatherHandler constructor argument. If you want to rely just on the default + correlationStrategy for the recipient-list-router and the + aggregator, you should specify apply-sequence="true". Otherwise, a custom + correlationStrategy should be supplied for the aggregator. + Unlike the PublishSubscribeChannel (Auction) variant, having a + recipient-list-router selector option, we can filter + target suppliers based on the message. With apply-sequence="true" the default + sequenceSize will be supplied and the aggregator will be able to release the group + correctly. The Distribution option is mutually exclusive with the + Auction option. + + + In both cases, the request (scatter) message is enriched with the + gatherResultChannel QueueChannel 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). + However, the gatherChannel option is also provided, allowing suppliers to send their + reply to that channel for the aggregation.
- Configuring a Scatter-Gather + Configuring a Scatter-Gather Endpoint - TBD + For Java and Annotation configuration, the bean definition for the Scatter-Gather + is: + + + Here, we configure the RecipientListRouter distributor bean, with + applySequence="true" and the list of recipient channels. The next bean is for an + AggregatingMessageHandler. Finally, we inject both those beans into the + ScatterGatherHandler bean definition and mark it as a + @ServiceActivator to wire the Scatter-Gather component into the integration flow. + + + Configuring the <scatter-gather> endpoint using the XML namespace: + + ]]> ]]> ]]>]]> + + + + + The id of the Endpoint. + The ScatterGatherHandler bean is registered with id + '.handler' + alias. The RecipientListRouter - with id + '.scatterer'. + And the AggregatingMessageHandler with id + '.gatherer'. + Optional (a default id is generated value by BeanFactory). + + + + + Lifecycle attribute signaling if the Endpoint should be started during Application Context + initialization. In addition, the ScatterGatherHandler also implements + Lifecycle and starts/stops the gatherEndpoint, which + is created internally if a gather-channel is provided. + Optional (default is true). + + + + The channel to receive request messages to handle them in the ScatterGatherHandler. + Required. + + + + The channel to which the Scatter-Gather will send the aggregation + results. Optional (because incoming messages can specify a + reply channel themselves via replyChannel Message Header). + + + + The channel to send the scatter message for the Auction scenario. + Optional. Mutually exclusive with <scatterer> sub + -element. + + + + + The channel to receive replies from each supplier for the aggregation. is used as the + replyChannel header in the scatter message. + Optional. By default the FixedSubscriberChannel is + created. + + + + + Order of this component when more than one handler is subscribed to the same DirectChannel + (use for load balancing purposes). + Optional. + + + + Specify the phase in which the endpoint + should be started and stopped. The startup order proceeds + from lowest to highest, and the shutdown order is the + reverse of that. By default this value is Integer.MAX_VALUE + meaning that this container starts as late as possible and + stops as soon as possible. + Optional. + + + + The timeout interval to wait when sending a reply + Message to the output-channel. + By default the send will block for one second. + It applies only if the output channel has some 'sending' limitations, e.g. a QueueChannel + with a fixed 'capacity' and is full. In this case, a MessageDeliveryException is thrown. + The send-timeout is ignored in case of AbstractSubscribableChannel implementations. + In case of group-timeout(-expression) the MessageDeliveryException + from the scheduled expire task leads this task to be rescheduled. + Optional. + + + + Allows you to specify how long the Scatter-Gather will wait for the reply message + before returning. By default it will wait indefinitely. 'null' is returned + if the reply times out. + Optional. Defaults to -1 - indefinitely. + + + + + Specify whether the Scatter-Gather must return a non-null value. This value is + true by default, hence a ReplyRequiredException will be thrown + when the underlying aggregator returns a null value after gather-timeout. + Note, if null is a possibility, the gather-timeout should be specified + to avoid an indefinite wait. + + + + + The <recipient-list-router> options. + Optional. Mutually exclusive with scatter-channel + attribute. + + + + The <aggregator> options. + Required. + + +