Spring Integration Aggregator combined with Publish Subscriber

Question

I have basic question in Spring integration. I tried searching in other threads but was not satisfied with the answers. So here are my questions. Any tip will be highly appreciated.

I am trying to get data from 4 different sources in parallel and aggregate the data.

My understanding (as per one of the threads I read earlier) is, any output channel of a splitter if is a Direct Channel, then the request is routed sequentially.

But output channel of the Splitter is a recepient-list router. When I debugged the request was being routed sequentially. So upon some analysis, I found we can use payload type router to process in parallel.

So I changed the router to payload type router and still the request is being routed sequentially.

So I changed my strategy and used a Publish-subscriber Channel and connected the output of all the subscribing channels to an aggregator so that I can aggregate the results. I was able to source the data in parallel but I do not see the response being routed to aggregated. Am I doing it wrong? Please suggest!

    <int:publish-subscribe-channel id="PublishSubscriberChannel"
    task-executor="taskExecutor" />

<task:executor id="taskExecutor" pool-size="5" />

<int:service-activator input-channel="PublishSubscriberChannel"
    ref="lookUpService" method="lookupSource1" output-channel="lookupAggregatorChannel" />

<int:service-activator input-channel="PublishSubscriberChannel"
    ref="lookUpService" method="lookupSource2" output-channel="lookupAggregatorChannel" />

<int:service-activator input-channel="PublishSubscriberChannel"
    ref="lookUpService" method="getVehicleInfoWithAdditionalAttributes"
    output-channel="lookupAggregatorChannel" />

<int:service-activator input-channel="PublishSubscriberChannel"
    ref="lookUpService" method="lookupSource4" output-channel="lookupAggregatorChannel" />

Karthik Mannepalli Karthik Mannepalli · Accepted Answer · 2016-06-28T17:36:44

So Finally, I found what I was doing wrong. Initially, I was working on Spring Integration Version 2.0. This is an old application. So when my code was exactly what is given in the question, I couldn't see the request being routed to Aggregator. Then when I upgraded the Spring Version to 4.3.2 RELEASE, I started getting error that Correlation Strategy might be null. That is when I found out the Aggregator needs a Correlation ID and if I had splitter correlation ID is automatically added and for Publish-Subscribe Channel, we need to add apply-sequence="true" in order to add a default correlation id which can be used by aggregator. So in essence, this is how my code looks now and works perfectly alright.

<int:publish-subscribe-channel id="PublishSubscriberChannel" apply-sequence="true"
    task-executor="taskExecutor" />

<task:executor id="taskExecutor" pool-size="5" />

<int:service-activator input-channel="PublishSubscriberChannel"
    ref="lookUpService" method="lookupSource1" output-channel="lookupAggregatorChannel" />

<int:service-activator input-channel="PublishSubscriberChannel"
    ref="lookUpService" method="lookupSource2" output-channel="lookupAggregatorChannel" />

<int:service-activator input-channel="PublishSubscriberChannel"
    ref="lookUpService" method="lookupSource3"
    output-channel="lookupAggregatorChannel" />

<int:service-activator input-channel="PublishSubscriberChannel"
    ref="lookUpService" method="lookupSource4" output-channel="lookupAggregatorChannel" />

Spring Integration Aggregator combined with Publish Subscriber

1 Answers