b7e928b846
Fixes https://github.com/spring-projects/spring-integration/issues/3024 To avoid confusing about `errorChannel` header behavior and make it more clear how to handle errors in Spring Integration flows, it would be better to present an `Error Handling` chapter on the top level. So, now it is a first chapter of the `Appendices` section
216 lines
12 KiB
Plaintext
216 lines
12 KiB
Plaintext
[[scatter-gather]]
|
|
=== Scatter-Gather
|
|
|
|
Starting with version 4.1, Spring Integration provides an implementation of the https://www.enterpriseintegrationpatterns.com/BroadcastAggregate.html[scatter-gather] enterprise integration pattern.
|
|
It is a compound endpoint for which the goal is to send a message to the recipients and aggregate the results.
|
|
As noted in https://www.enterpriseintegrationpatterns.com/[_Enterprise Integration Patterns_], it is a component for scenarios such as "`best quote`", where 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 by using discrete components.
|
|
This enhancement brings more convenient configuration.
|
|
|
|
The `ScatterGatherHandler` is a request-reply endpoint that combines a `PublishSubscribeChannel` (or a `RecipientListRouter`) and an `AggregatingMessageHandler`.
|
|
The request message is sent to the `scatter` channel, and the `ScatterGatherHandler` waits for the reply that the aggregator sends to the `outputChannel`.
|
|
|
|
[[scatter-gather-functionality]]
|
|
==== Functionality
|
|
|
|
The `Scatter-Gather` pattern suggests two scenarios: "`auction`" and "`distribution`".
|
|
In both cases, the `aggregation` function is the same and provides all the options available for the `AggregatingMessageHandler`.
|
|
(Actually, the `ScatterGatherHandler` requires only an `AggregatingMessageHandler` as a constructor argument.)
|
|
See <<./aggregator.adoc#aggregator,Aggregator>> 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 <<./content-enrichment.adoc#content-enricher,Content Enricher>>).
|
|
However, in this case, you should create your own custom `correlationStrategy` for the `aggregation` function.
|
|
|
|
===== Distribution
|
|
|
|
The distribution `Scatter-Gather` variant is based on the `RecipientListRouter` (see <<./router.adoc#router-implementations-recipientlistrouter,`RecipientListRouter`>>) with all available options for the `RecipientListRouter`.
|
|
This is the second `ScatterGatherHandler` constructor argument.
|
|
If you want to rely on only the default `correlationStrategy` for the `recipient-list-router` and the `aggregator`, you should specify `apply-sequence="true"`.
|
|
Otherwise, you should supply a custom `correlationStrategy` for the `aggregator`.
|
|
Unlike the `PublishSubscribeChannel` variant (the auction variant), having a `recipient-list-router` `selector` option lets filter target suppliers based on the message.
|
|
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.
|
|
|
|
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).
|
|
However, the `gatherChannel` option is also provided, letting suppliers send their reply to that channel for the aggregation.
|
|
|
|
[[scatter-gather-namespace]]
|
|
==== Configuring a Scatter-Gather Endpoint
|
|
|
|
The following example shows Java configuration for the bean definition for `Scatter-Gather`:
|
|
|
|
====
|
|
[source,java]
|
|
----
|
|
@Bean
|
|
public MessageHandler distributor() {
|
|
RecipientListRouter router = new RecipientListRouter();
|
|
router.setApplySequence(true);
|
|
router.setChannels(Arrays.asList(distributionChannel1(), distributionChannel2(),
|
|
distributionChannel3()));
|
|
return router;
|
|
}
|
|
|
|
@Bean
|
|
public MessageHandler gatherer() {
|
|
return new AggregatingMessageHandler(
|
|
new ExpressionEvaluatingMessageGroupProcessor("^[payload gt 5] ?: -1D"),
|
|
new SimpleMessageStore(),
|
|
new HeaderAttributeCorrelationStrategy(
|
|
IntegrationMessageHeaderAccessor.CORRELATION_ID),
|
|
new ExpressionEvaluatingReleaseStrategy("size() == 2"));
|
|
}
|
|
|
|
@Bean
|
|
@ServiceActivator(inputChannel = "distributionChannel")
|
|
public MessageHandler scatterGatherDistribution() {
|
|
ScatterGatherHandler handler = new ScatterGatherHandler(distributor(), gatherer());
|
|
handler.setOutputChannel(output());
|
|
return handler;
|
|
}
|
|
----
|
|
====
|
|
|
|
In the preceding example, 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.
|
|
|
|
The following example shows how to configure the `<scatter-gather>` endpoint by using the XML namespace:
|
|
|
|
====
|
|
[source,xml]
|
|
----
|
|
<scatter-gather
|
|
id="" <1>
|
|
auto-startup="" <2>
|
|
input-channel="" <3>
|
|
output-channel="" <4>
|
|
scatter-channel="" <5>
|
|
gather-channel="" <6>
|
|
order="" <7>
|
|
phase="" <8>
|
|
send-timeout="" <9>
|
|
gather-timeout="" <10>
|
|
requires-reply="" > <11>
|
|
<scatterer/> <12>
|
|
<gatherer/> <13>
|
|
</scatter-gather>
|
|
----
|
|
|
|
<1> The id of the endpoint.
|
|
The `ScatterGatherHandler` bean is registered with an alias of `id + '.handler'`.
|
|
The `RecipientListRouter` bean is registered with an alias of `id + '.scatterer'`.
|
|
The `AggregatingMessageHandler`bean is registered with an alias of `id + '.gatherer'`.
|
|
Optional.
|
|
(The `BeanFactory` generates a default `id` value.)
|
|
<2> Lifecycle attribute signaling whether the endpoint should be started during application context initialization.
|
|
In addition, the `ScatterGatherHandler` also implements `Lifecycle` and starts and stops `gatherEndpoint`, which is created internally if a `gather-channel` is provided.
|
|
Optional.
|
|
(The default is `true`.)
|
|
<3> The channel on which to receive request messages to handle them in the `ScatterGatherHandler`.
|
|
Required.
|
|
<4> The channel to which the `ScatterGatherHandler` sends the aggregation results.
|
|
Optional.
|
|
(Incoming messages can specify a reply channel themselves in the `replyChannel` message header).
|
|
<5> The channel to which to send the scatter message for the auction scenario.
|
|
Optional.
|
|
Mutually exclusive with the `<scatterer>` sub-element.
|
|
<6> The channel on which to receive replies from each supplier for the aggregation.
|
|
It is used as the `replyChannel` header in the scatter message.
|
|
Optional.
|
|
By default, the `FixedSubscriberChannel` is created.
|
|
<7> The order of this component when more than one handler is subscribed to the same `DirectChannel` (use for load balancing purposes).
|
|
Optional.
|
|
<8> Specifies the phase in which the endpoint should be started and stopped.
|
|
The startup order proceeds from lowest to highest, and the shutdown order is from highest to lowest.
|
|
By default, this value is `Integer.MAX_VALUE`, meaning that this container starts as late as possible and stops as soon as possible.
|
|
Optional.
|
|
<9> The timeout interval to wait when sending a reply `Message` to the `output-channel`.
|
|
By default, the send blocks for one second.
|
|
It applies only if the output channel has some 'sending' limitations -- for example, a `QueueChannel` with a fixed 'capacity' that is full.
|
|
In this case, a `MessageDeliveryException` is thrown.
|
|
The `send-timeout` is ignored for `AbstractSubscribableChannel` implementations.
|
|
For `group-timeout(-expression)`, the `MessageDeliveryException` from the scheduled expire task leads this task to be rescheduled.
|
|
Optional.
|
|
<10> Lets you specify how long the scatter-gather waits for the reply message before returning.
|
|
By default, it waits indefinitely.
|
|
'null' is returned if the reply times out.
|
|
Optional.
|
|
It defaults to `-1`, meaning to wait indefinitely.
|
|
<11> Specifies whether the scatter-gather must return a non-null value.
|
|
This value is `true` by default.
|
|
Consequently, a `ReplyRequiredException` is 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.
|
|
<12> The `<recipient-list-router>` options.
|
|
Optional.
|
|
Mutually exclusive with `scatter-channel` attribute.
|
|
<13> The `<aggregator>` options.
|
|
Required.
|
|
====
|
|
|
|
[[scatter-gather-error-handling]]
|
|
==== Error Handling
|
|
|
|
Since Scatter-Gather is a multi request-reply component, error handling has some extra complexity.
|
|
In some cases, it is better to just catch and ignore downstream exceptions if the `ReleaseStrategy` allows the process to finish with fewer replies than requests.
|
|
In other cases something like a "`compensation message`" should be considered for returning from sub-flow, when an error happens.
|
|
|
|
Every async sub-flow should be configured with a `errorChannel` header for the proper error message sending from the `MessagePublishingErrorHandler`.
|
|
Otherwise, an error will be sent to the global `errorChannel` with the common error handling logic.
|
|
See <<./error-handling.adoc#error-handling,Error Handling>> for more information about async error processing.
|
|
|
|
Synchronous flows may use an `ExpressionEvaluatingRequestHandlerAdvice` for ignoring the exception or returning a compensation message.
|
|
When an exception is thrown from one of the sub-flows to the `ScatterGatherHandler`, it is just re-thrown to upstream.
|
|
This way all other sub-flows will work for nothing and their replies are going to be ignored in the `ScatterGatherHandler`.
|
|
This might be an expected behavior sometimes, but in most cases it would be better to handle the error in the particular sub-flow without impacting all others and the expectations in the gatherer.
|
|
|
|
Starting with version 5.1.3, the `ScatterGatherHandler` is supplied with the `errorChannelName` option.
|
|
It is populated to the `errorChannel` header of the scatter message and is used in the when async error happens or can be used in the regular synchronous sub-flow for directly sending an error message.
|
|
|
|
The sample configuration below demonstrates async error handling by returning a compensation message:
|
|
|
|
====
|
|
[source,java]
|
|
----
|
|
@Bean
|
|
public IntegrationFlow scatterGatherAndExecutorChannelSubFlow(TaskExecutor taskExecutor) {
|
|
return f -> f
|
|
.scatterGather(
|
|
scatterer -> scatterer
|
|
.applySequence(true)
|
|
.recipientFlow(f1 -> f1.transform(p -> "Sub-flow#1"))
|
|
.recipientFlow(f2 -> f2
|
|
.channel(c -> c.executor(taskExecutor))
|
|
.transform(p -> {
|
|
throw new RuntimeException("Sub-flow#2");
|
|
})),
|
|
null,
|
|
s -> s.errorChannel("scatterGatherErrorChannel"));
|
|
}
|
|
|
|
@ServiceActivator(inputChannel = "scatterGatherErrorChannel")
|
|
public Message<?> processAsyncScatterError(MessagingException payload) {
|
|
return MessageBuilder.withPayload(payload.getCause().getCause())
|
|
.copyHeaders(payload.getFailedMessage().getHeaders())
|
|
.build();
|
|
}
|
|
----
|
|
====
|
|
|
|
To produce a proper reply, we have to copy headers (including `replyChannel` and `errorChannel`) from the `failedMessage` of the `MessagingException` that has been sent to the `scatterGatherErrorChannel` by the `MessagePublishingErrorHandler`.
|
|
This way the target exception is returned to the gatherer of the `ScatterGatherHandler` for reply messages group completion.
|
|
Such an exception `payload` can be filtered out in the `MessageGroupProcessor` of the gatherer or processed other way downstream, after the scatter-gather endpoint.
|
|
|
|
NOTE: Before sending scattering results to the gatherer, `ScatterGatherHandler` reinstates the request message headers, including reply and error channels if any.
|
|
This way errors from the `AggregatingMessageHandler` are going to be propagated to the caller, even if an async hand off is applied in scatter recipient subflows.
|
|
For successful operation, a `gatherResultChannel`, `originalReplyChannel` and `originalErrorChannel` headers must be transferred back to replies from scatter recipient subflows.
|
|
In this case a reasonable, finite `gatherTimeout` must be configured for the `ScatterGatherHandler`.
|
|
Otherwise it is going to be blocked waiting for a reply from the gatherer forever, by default.
|
|
|