Skip to content

Commit e8aa861

Browse files
artembilangaryrussell
artembilan
authored and
garyrussell
committed
GH-2735: Add errorChannel to ScatterGatherHandler
Fixes #2735 * Fix typos in Scatter-Gather JavaDocs * Add Scatter-Gather error handling documentation Doc polishing
1 parent 5c46efe commit e8aa861

File tree

5 files changed

+189
-46
lines changed

5 files changed

+189
-46
lines changed

‎spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -86,6 +86,7 @@
8686
import org.springframework.integration.transformer.MethodInvokingTransformer;
8787
import org.springframework.integration.transformer.Transformer;
8888
import org.springframework.integration.util.ClassUtils;
89+
import org.springframework.lang.Nullable;
8990
import org.springframework.messaging.Message;
9091
import org.springframework.messaging.MessageChannel;
9192
import org.springframework.messaging.MessageHandler;
@@ -2847,7 +2848,7 @@ public B scatterGather(Consumer<RecipientListRouterSpec> scatterer) {
28472848
* Can be {@code null}.
28482849
* @return the current {@link IntegrationFlowDefinition}.
28492850
*/
2850-
public B scatterGather(Consumer<RecipientListRouterSpec> scatterer, Consumer<AggregatorSpec> gatherer) {
2851+
public B scatterGather(Consumer<RecipientListRouterSpec> scatterer, @Nullable Consumer<AggregatorSpec> gatherer) {
28512852
return scatterGather(scatterer, gatherer, null);
28522853
}
28532854

@@ -2861,8 +2862,9 @@ public B scatterGather(Consumer<RecipientListRouterSpec> scatterer, Consumer<Agg
28612862
* {@link ScatterGatherHandler} and its endpoint. Can be {@code null}.
28622863
* @return the current {@link IntegrationFlowDefinition}.
28632864
*/
2864-
public B scatterGather(Consumer<RecipientListRouterSpec> scatterer, Consumer<AggregatorSpec> gatherer,
2865-
Consumer<ScatterGatherSpec> scatterGather) {
2865+
public B scatterGather(Consumer<RecipientListRouterSpec> scatterer, @Nullable Consumer<AggregatorSpec> gatherer,
2866+
@Nullable Consumer<ScatterGatherSpec> scatterGather) {
2867+
28662868
Assert.notNull(scatterer, "'scatterer' must not be null");
28672869
RecipientListRouterSpec recipientListRouterSpec = new RecipientListRouterSpec();
28682870
scatterer.accept(recipientListRouterSpec);

‎spring-integration-core/src/main/java/org/springframework/integration/dsl/ScatterGatherSpec.java‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2016 the original author or authors.
2+
* Copyright 2016-2019 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -16,6 +16,7 @@
1616

1717
package org.springframework.integration.dsl;
1818

19+
import org.springframework.integration.context.IntegrationContextUtils;
1920
import org.springframework.integration.scattergather.ScatterGatherHandler;
2021
import org.springframework.messaging.MessageChannel;
2122

@@ -58,4 +59,16 @@ public ScatterGatherSpec gatherTimeout(long gatherTimeout) {
5859
return this;
5960
}
6061

62+
/**
63+
* Specify a {@link MessageChannel} bean name for async error processing.
64+
* Defaults to {@link IntegrationContextUtils#ERROR_CHANNEL_BEAN_NAME}.
65+
* @param errorChannel the {@link MessageChannel} bean name for async error processing.
66+
* @return the current {@link ScatterGatherSpec} instance.
67+
* @since 5.1.3
68+
*/
69+
public ScatterGatherSpec errorChannel(String errorChannel) {
70+
this.handler.setErrorChannelName(errorChannel);
71+
return this;
72+
}
73+
6174
}

‎spring-integration-core/src/main/java/org/springframework/integration/scattergather/ScatterGatherHandler.java‎

Lines changed: 64 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2014-2018 the original author or authors.
2+
* Copyright 2014-2019 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -17,22 +17,25 @@
1717
package org.springframework.integration.scattergather;
1818

1919
import org.springframework.aop.support.AopUtils;
20+
import org.springframework.beans.factory.BeanFactory;
21+
import org.springframework.beans.factory.BeanInitializationException;
2022
import org.springframework.context.Lifecycle;
2123
import org.springframework.integration.channel.FixedSubscriberChannel;
2224
import org.springframework.integration.channel.QueueChannel;
25+
import org.springframework.integration.channel.ReactiveStreamsSubscribableChannel;
2326
import org.springframework.integration.context.IntegrationContextUtils;
2427
import org.springframework.integration.core.MessageProducer;
2528
import org.springframework.integration.endpoint.AbstractEndpoint;
2629
import org.springframework.integration.endpoint.EventDrivenConsumer;
2730
import org.springframework.integration.endpoint.PollingConsumer;
31+
import org.springframework.integration.endpoint.ReactiveStreamsConsumer;
2832
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
2933
import org.springframework.integration.support.channel.HeaderChannelRegistry;
3034
import org.springframework.messaging.Message;
3135
import org.springframework.messaging.MessageChannel;
3236
import org.springframework.messaging.MessageDeliveryException;
3337
import org.springframework.messaging.MessageHandler;
3438
import org.springframework.messaging.MessageHeaders;
35-
import org.springframework.messaging.MessagingException;
3639
import org.springframework.messaging.PollableChannel;
3740
import org.springframework.messaging.SubscribableChannel;
3841
import org.springframework.util.Assert;
@@ -57,13 +60,22 @@ public class ScatterGatherHandler extends AbstractReplyProducingMessageHandler i
5760

5861
private MessageChannel gatherChannel;
5962

63+
private String errorChannelName = IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME;
64+
6065
private long gatherTimeout = -1;
6166

6267
private AbstractEndpoint gatherEndpoint;
6368

6469
private HeaderChannelRegistry replyChannelRegistry;
6570

6671

72+
public ScatterGatherHandler(MessageHandler scatterer, MessageHandler gatherer) {
73+
this(new FixedSubscriberChannel(scatterer), gatherer);
74+
Assert.notNull(scatterer, "'scatterer' must not be null");
75+
Class<?> scattererClass = AopUtils.getTargetClass(scatterer);
76+
checkClass(scattererClass, "org.springframework.integration.router.RecipientListRouter", "scatterer");
77+
}
78+
6779
public ScatterGatherHandler(MessageChannel scatterChannel, MessageHandler gatherer) {
6880
Assert.notNull(scatterChannel, "'scatterChannel' must not be null");
6981
Assert.notNull(gatherer, "'gatherer' must not be null");
@@ -73,13 +85,6 @@ public ScatterGatherHandler(MessageChannel scatterChannel, MessageHandler gather
7385
this.gatherer = gatherer;
7486
}
7587

76-
public ScatterGatherHandler(MessageHandler scatterer, MessageHandler gatherer) {
77-
this(new FixedSubscriberChannel(scatterer), gatherer);
78-
Assert.notNull(scatterer, "'scatterer' must not be null");
79-
Class<?> scattererClass = AopUtils.getTargetClass(scatterer);
80-
checkClass(scattererClass, "org.springframework.integration.router.RecipientListRouter", "scatterer");
81-
}
82-
8388
public void setGatherChannel(MessageChannel gatherChannel) {
8489
this.gatherChannel = gatherChannel;
8590
}
@@ -88,8 +93,20 @@ public void setGatherTimeout(long gatherTimeout) {
8893
this.gatherTimeout = gatherTimeout;
8994
}
9095

96+
/**
97+
* Specify a {@link MessageChannel} bean name for async error processing.
98+
* Defaults to {@link IntegrationContextUtils#ERROR_CHANNEL_BEAN_NAME}.
99+
* @param errorChannelName the {@link MessageChannel} bean name for async error processing.
100+
* @since 5.1.3
101+
*/
102+
public void setErrorChannelName(String errorChannelName) {
103+
Assert.hasText(errorChannelName, "'errorChannelName' must not be empty.");
104+
this.errorChannelName = errorChannelName;
105+
}
106+
91107
@Override
92108
protected void doInit() {
109+
BeanFactory beanFactory = getBeanFactory();
93110
if (this.gatherChannel == null) {
94111
this.gatherChannel = new FixedSubscriberChannel(this.gatherer);
95112
}
@@ -101,33 +118,39 @@ else if (this.gatherChannel instanceof PollableChannel) {
101118
this.gatherEndpoint = new PollingConsumer((PollableChannel) this.gatherChannel, this.gatherer);
102119
((PollingConsumer) this.gatherEndpoint).setReceiveTimeout(this.gatherTimeout);
103120
}
121+
else if (this.gatherChannel instanceof ReactiveStreamsSubscribableChannel) {
122+
this.gatherEndpoint = new ReactiveStreamsConsumer(this.gatherChannel, this.gatherer);
123+
}
104124
else {
105-
throw new MessagingException("Unsupported 'replyChannel' type [" + this.gatherChannel.getClass() + "]."
106-
+ "SubscribableChannel or PollableChannel type are supported.");
125+
throw new BeanInitializationException("Unsupported 'replyChannel' type '" +
126+
this.gatherChannel.getClass() + "'. " +
127+
"'SubscribableChannel', 'PollableChannel' or 'ReactiveStreamsSubscribableChannel' " +
128+
"types are supported.");
107129
}
108-
this.gatherEndpoint.setBeanFactory(this.getBeanFactory());
130+
this.gatherEndpoint.setBeanFactory(beanFactory);
109131
this.gatherEndpoint.afterPropertiesSet();
110132
}
111133

112-
((MessageProducer) this.gatherer).setOutputChannel(new FixedSubscriberChannel(message -> {
113-
MessageHeaders headers = message.getHeaders();
114-
if (headers.containsKey(GATHER_RESULT_CHANNEL)) {
115-
Object gatherResultChannel = headers.get(GATHER_RESULT_CHANNEL);
116-
if (gatherResultChannel instanceof MessageChannel) {
117-
messagingTemplate.send((MessageChannel) gatherResultChannel, message);
118-
}
119-
else if (gatherResultChannel instanceof String) {
120-
messagingTemplate.send((String) gatherResultChannel, message);
121-
}
122-
}
123-
else {
124-
throw new MessageDeliveryException(message,
125-
"The 'gatherResultChannel' header is required to delivery gather result.");
126-
}
127-
}));
128-
129-
this.replyChannelRegistry = getBeanFactory()
130-
.getBean(IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME,
134+
((MessageProducer) this.gatherer)
135+
.setOutputChannel(new FixedSubscriberChannel(message -> {
136+
MessageHeaders headers = message.getHeaders();
137+
if (headers.containsKey(GATHER_RESULT_CHANNEL)) {
138+
Object gatherResultChannel = headers.get(GATHER_RESULT_CHANNEL);
139+
if (gatherResultChannel instanceof MessageChannel) {
140+
messagingTemplate.send((MessageChannel) gatherResultChannel, message);
141+
}
142+
else if (gatherResultChannel instanceof String) {
143+
messagingTemplate.send((String) gatherResultChannel, message);
144+
}
145+
}
146+
else {
147+
throw new MessageDeliveryException(message,
148+
"The 'gatherResultChannel' header is required to delivery gather result.");
149+
}
150+
}));
151+
152+
this.replyChannelRegistry =
153+
beanFactory.getBean(IntegrationContextUtils.INTEGRATION_HEADER_CHANNEL_REGISTRY_BEAN_NAME,
131154
HeaderChannelRegistry.class);
132155
}
133156

@@ -137,11 +160,13 @@ protected Object handleRequestMessage(Message<?> requestMessage) {
137160

138161
Object gatherResultChannelName = this.replyChannelRegistry.channelToChannelName(gatherResultChannel);
139162

140-
Message<?> scatterMessage = getMessageBuilderFactory()
141-
.fromMessage(requestMessage)
142-
.setHeader(GATHER_RESULT_CHANNEL, gatherResultChannelName)
143-
.setReplyChannel(this.gatherChannel)
144-
.build();
163+
Message<?> scatterMessage =
164+
getMessageBuilderFactory()
165+
.fromMessage(requestMessage)
166+
.setHeader(GATHER_RESULT_CHANNEL, gatherResultChannelName)
167+
.setReplyChannel(this.gatherChannel)
168+
.setErrorChannelName(this.errorChannelName)
169+
.build();
145170

146171
this.messagingTemplate.send(this.scatterChannel, scatterMessage);
147172

@@ -151,7 +176,7 @@ protected Object handleRequestMessage(Message<?> requestMessage) {
151176
.fromMessage(gatherResult)
152177
.removeHeader(GATHER_RESULT_CHANNEL)
153178
.setHeader(MessageHeaders.REPLY_CHANNEL, requestMessage.getHeaders().getReplyChannel())
154-
.build();
179+
.setHeader(MessageHeaders.ERROR_CHANNEL, requestMessage.getHeaders().getErrorChannel());
155180
}
156181

157182
return null;
@@ -179,7 +204,8 @@ public boolean isRunning() {
179204
private void checkClass(Class<?> gathererClass, String className, String type) throws LinkageError {
180205
try {
181206
Class<?> clazz = ClassUtils.forName(className, ClassUtils.getDefaultClassLoader());
182-
Assert.isAssignable(clazz, gathererClass, "the '" + type + "' must be an " + className + " instance");
207+
Assert.isAssignable(clazz, gathererClass, () -> "the '" + type + "' must be an " + className + " " +
208+
"instance");
183209
}
184210
catch (ClassNotFoundException e) {
185211
throw new IllegalStateException("The class for '" + className + "' cannot be loaded", e);

‎spring-integration-core/src/test/java/org/springframework/integration/dsl/routers/RouterTests.java‎

Lines changed: 53 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2016-2018 the original author or authors.
2+
* Copyright 2016-2019 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -39,8 +39,10 @@
3939
import org.springframework.beans.factory.annotation.Qualifier;
4040
import org.springframework.context.annotation.Bean;
4141
import org.springframework.context.annotation.Configuration;
42+
import org.springframework.core.task.TaskExecutor;
4243
import org.springframework.integration.IntegrationMessageHeaderAccessor;
4344
import org.springframework.integration.annotation.Router;
45+
import org.springframework.integration.annotation.ServiceActivator;
4446
import org.springframework.integration.channel.QueueChannel;
4547
import org.springframework.integration.config.EnableIntegration;
4648
import org.springframework.integration.config.EnableMessageHistory;
@@ -550,8 +552,9 @@ public void testNestedScatterGather() {
550552
Object payload = bestQuoteMessage.getPayload();
551553
assertThat(payload, instanceOf(String.class));
552554
List<?> topSequenceDetails =
553-
(List<?>) bestQuoteMessage.getHeaders().get(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS, List.class)
554-
.get(0);
555+
(List<?>) bestQuoteMessage.getHeaders()
556+
.get(IntegrationMessageHeaderAccessor.SEQUENCE_DETAILS, List.class)
557+
.get(0);
555558

556559
assertEquals(request.getHeaders().getId(),
557560
bestQuoteMessage.getHeaders().get(IntegrationMessageHeaderAccessor.CORRELATION_ID));
@@ -566,6 +569,26 @@ public void testNestedScatterGather() {
566569
topSequenceDetails.get(2));
567570
}
568571

572+
@Autowired
573+
@Qualifier("scatterGatherAndExecutorChannelSubFlow.input")
574+
private MessageChannel scatterGatherAndExecutorChannelSubFlowInput;
575+
576+
@Test
577+
public void testScatterGatherWithExecutorChannelSubFlow() {
578+
QueueChannel replyChannel = new QueueChannel();
579+
Message<?> testMessage =
580+
MessageBuilder.withPayload("test")
581+
.setReplyChannel(replyChannel)
582+
.build();
583+
584+
this.scatterGatherAndExecutorChannelSubFlowInput.send(testMessage);
585+
586+
Message<?> receive = replyChannel.receive(10_000);
587+
assertNotNull(receive);
588+
Object payload = receive.getPayload();
589+
assertThat(payload, instanceOf(List.class));
590+
assertThat(((List) payload).get(1), instanceOf(RuntimeException.class));
591+
}
569592

570593
@Configuration
571594
@EnableIntegration
@@ -689,7 +712,9 @@ public IntegrationFlow routeMethodInvocationFlow3() {
689712
@Bean
690713
public IntegrationFlow routeMultiMethodInvocationFlow() {
691714
return IntegrationFlows.from("routerMultiInput")
692-
.route(String.class, p -> p.equals("foo") || p.equals("bar") ? new String[] { "foo", "bar" } : null,
715+
.route(String.class, p -> p.equals("foo") || p.equals("bar")
716+
? new String[] { "foo", "bar" }
717+
: null,
693718
s -> s.suffix("-channel"))
694719
.get();
695720
}
@@ -832,6 +857,30 @@ public IntegrationFlow nestedScatterGatherFlow() {
832857
.collect(Collectors.joining("\n")));
833858
}
834859

860+
861+
@Bean
862+
public IntegrationFlow scatterGatherAndExecutorChannelSubFlow(TaskExecutor taskExecutor) {
863+
return f -> f
864+
.scatterGather(
865+
scatterer -> scatterer
866+
.applySequence(true)
867+
.recipientFlow(f1 -> f1.transform(p -> "Sub-flow#1"))
868+
.recipientFlow(f2 -> f2
869+
.channel(c -> c.executor(taskExecutor))
870+
.transform(p -> {
871+
throw new RuntimeException("Sub-flow#2");
872+
})),
873+
null,
874+
s -> s.errorChannel("scatterGatherErrorChannel"));
875+
}
876+
877+
@ServiceActivator(inputChannel = "scatterGatherErrorChannel")
878+
public Message<?> processAsyncScatterError(MessagingException payload) {
879+
return MessageBuilder.withPayload(payload.getCause().getCause())
880+
.copyHeaders(payload.getFailedMessage().getHeaders())
881+
.build();
882+
}
883+
835884
}
836885

837886
private static class RoutingTestBean {

0 commit comments

Comments
 (0)