Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2967: Fix ScatterGatherH for headers copy (#2968)

Browse Source 
* GH-2967: Fix ScatterGatherH for headers copy

Fixes https://github.com/spring-projects/spring-integration/issues/2967

The `ChannelInterceptor` is added into the `this.gatherChannel` on each
request message making a subsequent requests for scatter-gather as
halting on reply.

* Add an interceptor into an injected `this.gatherChannel` only once
during `ScatterGatherHandler` initialization
* Introduce `ORIGINAL_REPLY_CHANNEL` and `ORIGINAL_ERROR_CHANNEL`
headers to carry a request reply and error channels from headers
* Populate `REPLY_CHANNEL` and `ERROR_CHANNEL` headers back before
sending scattering replies into gatherer
* Transfer a `GATHER_RESULT_CHANNEL` header now directly from the scatter
message to make it available in the reply from the gatherer
* Add note about those headers in the `scatter-gather.adoc`
* Modify `ScatterGatherTests` to be sure that `ScatterGatherHandler`
works for several requests

**Cherry-pick to 5.1.x**

* * Fix language in doc
This commit is contained in:
Artem Bilan
2019-06-19 16:14:12 -04:00
committed by Gary Russell
parent 36c14ef179
commit d79c06a987
4 changed files with 79 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests-context.xml
View File

@@ -4,7 +4,8 @@
xmlns="http://www.springframework.org/schema/integration"
xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/beans https://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd http://www.springframework.org/schema/task https://www.springframework.org/schema/task/spring-task.xsd">
http://www.springframework.org/schema/integration https://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/task https://www.springframework.org/schema/task/spring-task.xsd">
<channel id="output">
<queue/>
@@ -52,9 +53,12 @@
<!--Sync scenario-->
<gateway id="gateway" default-request-channel="gatewayAuction" default-reply-timeout="10000" />
<gateway id="gateway" default-request-channel="gatewayAuction" default-reply-timeout="10000"/>
<scatter-gather input-channel="gatewayAuction" output-channel="bridgeChannel" scatter-channel="auctionChannel">
<channel id="gatherChannel2"/>
<scatter-gather input-channel="gatewayAuction" output-channel="bridgeChannel" scatter-channel="auctionChannel"
gather-channel="gatherChannel2">
<gatherer release-strategy-expression="messages.^[payload gt 5] != null or size() == 3"/>
</scatter-gather>

49
spring-integration-core/src/test/java/org/springframework/integration/scattergather/config/ScatterGatherTests.java
View File

@@ -29,16 +29,19 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Artem Bilan
* @author Gary Russell
*
* @since 4.1
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class ScatterGatherTests {
@Autowired
@@ -58,36 +61,50 @@ public class ScatterGatherTests {
@Test
public void testAuction() {
this.inputAuction.send(new GenericMessage<String>("foo"));
this.inputAuction.send(new GenericMessage<>("foo"));
Message<?> bestQuoteMessage = this.output.receive(10000);
assertThat(bestQuoteMessage).isNotNull();
Object payload = bestQuoteMessage.getPayload();
assertThat(payload).isInstanceOf(List.class);
assertThat(((List<?>) payload).size()).isGreaterThanOrEqualTo(1);
assertThat(bestQuoteMessage)
.isNotNull()
.extracting(Message::getPayload)
.isInstanceOf(List.class)
.asList()
.hasSizeGreaterThanOrEqualTo(1);
}
@Test
public void testDistribution() {
this.inputDistribution.send(new GenericMessage<String>("foo"));
this.inputDistribution.send(new GenericMessage<>("foo"));
Message<?> bestQuoteMessage = this.output.receive(10000);
assertThat(bestQuoteMessage).isNotNull();
Object payload = bestQuoteMessage.getPayload();
assertThat(payload).isInstanceOf(List.class);
assertThat(((List<?>) payload).size()).isGreaterThanOrEqualTo(1);
assertThat(bestQuoteMessage)
.isNotNull()
.extracting(Message::getPayload)
.isInstanceOf(List.class)
.asList()
.hasSizeGreaterThanOrEqualTo(1);
}
@Test
public void testGatewayScatterGather() {
Message<?> bestQuoteMessage = this.gateway.exchange(new GenericMessage<String>("foo"));
assertThat(bestQuoteMessage).isNotNull();
Object payload = bestQuoteMessage.getPayload();
assertThat(payload).isInstanceOf(List.class);
assertThat(((List<?>) payload).size()).isGreaterThanOrEqualTo(1);
Message<?> bestQuoteMessage = this.gateway.exchange(new GenericMessage<>("foo"));
assertThat(bestQuoteMessage)
.isNotNull()
.extracting(Message::getPayload)
.isInstanceOf(List.class)
.asList()
.hasSizeGreaterThanOrEqualTo(1);
bestQuoteMessage = this.gateway.exchange(new GenericMessage<>("bar"));
assertThat(bestQuoteMessage)
.isNotNull()
.extracting(Message::getPayload)
.isInstanceOf(List.class)
.asList()
.hasSizeGreaterThanOrEqualTo(1);
}
@Test
public void testWithinChain() {
this.scatterGatherWithinChain.send(new GenericMessage<String>("foo"));
this.scatterGatherWithinChain.send(new GenericMessage<>("foo"));
for (int i = 0; i < 3; i++) {
Message<?> result = this.output.receive(10000);
assertThat(result).isNotNull();