Spring integration Scatter-Gather pattern with JMS transport

Question

I need to implement the following architecture:

1. I have data that must be sent to systems (Some external application ) using JMS.

2. Depending on the data you need to send only to the necessary systems (For example, if the number of systems is 4, then you can send from 1 to 4 )

3. It is necessary to wait for a response from the systems to which the messages were sent, after receiving all the answers, it is required to process the received data (or to process at least one timeout)

4. The correlation id is contained in the header of both outgoing and incoming JMS messages

5. Each new such process can be started asynchronously and in parallel

Now I have it implemented only with the help of Spring JMS. I synchronize the threads manually, also manually I manage the thread pools.

The correlation ids and information about the systems in which messages were sent are stored as a state and update it after receiving new messages, etc.

But I want to simplify the logic and use Spring-integration Java DSL, Scatter gather pattern (Which is just my case) and other useful Spring features.

Can you help me show an example of how such an architecture can be implemented with the help of Spring-integration/IntregrationFlow?

Answer 1

Here is some sample from our test-cases:

    @Bean
    public IntegrationFlow scatterGatherFlow() {
        return f -> f
                .scatterGather(scatterer -> scatterer
                                .applySequence(true)
                                .recipientFlow(m -> true, sf -> sf.handle((p, h) -> Math.random() * 10))
                                .recipientFlow(m -> true, sf -> sf.handle((p, h) -> Math.random() * 10))
                                .recipientFlow(m -> true, sf -> sf.handle((p, h) -> Math.random() * 10)),
                        gatherer -> gatherer
                                .releaseStrategy(group ->
                                        group.size() == 3 ||
                                                group.getMessages()
                                                        .stream()
                                                        .anyMatch(m -> (Double) m.getPayload() > 5)),
                        scatterGather -> scatterGather
                                .gatherTimeout(10_000));
    }

So, there is the parts:

• scatterer - to send messages to recipients. In your case all those JMS services. That can be a scatterChannel though. Typically PublishSubscribeChannel, so Scatter-Gather might not know subscrbibers in adavance.

• gatherer - well, it is just an aggregator with all its possible options.

• scatterGather - is just for convenience for the direct properties of the ScatterGatherHandler and common endpoint options.