Multiple Scattergathers in flow run in parallel

user_12353819 0
Viewed 31

I do have a requirement where I am having call to 3 APIs in which one of the API is dependent on the output of other. If we say we are having 3 APIs A, B and C A and B can run in parallel but C is dependent on the output of B.

Below is the pseudo code for the same.

public IntegrationFlow mainFlow()
    {
        return flow -> flow
                .scatterGather(scatterRequests(), aggregateResponse())
                .enrichHeaders(s -> s.headerExpressions(h -> h.put(AGGREGATED_RESPONSE, "payload")))
                .scatterGather(scatterOtherRequests(), aggregateResponse());
    }

private Consumer<RecipientListRouterSpec> scatterRequests()
         {
            return scatterer -> scatterer
                .applySequence(true)
                .recipientFlow(flowA())
                .recipientFlow(flowB());
        }
    
    private Consumer<RecipientListRouterSpec> scatterOtherRequests()
        {
            return scatterer -> scatterer
                .applySequence(true)
                .recipientFlow(flowC())
                .recipientFlow(aggregatedResponseSubFlow());
        }
    where aggregatedResponseSubFlow is as below
 private IntegrationFlow raggregatedResponseSubFlow() 
    {
         return subflow -> subflow
                 .handle((payload, headers) -> {
                     return headers.get(AGGREGATED_RESPONSE);
                 });
    }

Even though there are 2 Scattergathers one after the other, both of them run in parallel due to which I get inconsistent results. So how can I run it sequentially. i.e after the execution of 1st one the 2nd one should get started. If it cannot be done how else can the scenario be handled?

Note: I do need to aggregate responses from all 3 services

spring-integration-dsl spring-integration spring-boot
1 Answers

There is nothing in your code snippet confirming that process is done in parallel. But probably you just removed that part for simplicity of the code to show us. I assume that because it is not clear what is scatterOtherRequests(), but you show that scatterPendingRequests() instead.

Anyway it would be better to re-think your solution. It sounds more like A and B1 & B2 and make it in parallel like this:

  A
<
  B1 -> B2

Therefore it doesn't sound like you need a second scatter-gather. Even if you need a B1 result in the main flow, you still can combine it some way with B2 from that branch before replying back to the main flow.

user_27565470
Related