spring integration dsl buffer

2
votes

I have a requirement where I need to hold/buffer the messages that are received on a channel and persist in database based on number of messages or timeout mean no messages received for 1min. Is there a way to achieve this in spring integration

IntegrationFlows
                .from(Jms.messageDrivenChannelAdapter(connectionFactory)
                    .destination(sourceQueue))
                .transform(someTransform, "transform")
                .handle(someService, "save")
                .get();
1
spring-integrationspring-integration-dsl

1 Answers

1
votes

There is an .aggregate() operator based on the Aggregator EI-pattern implementation.

That one you can configure with the JdbcMessageStore to buffer messages and store them into DB.

You can hold them there until some condition via ReleaseStrategy (based on each message arriving) or release them due to group timeout.

If you are not interested to have them all afterward as a single aggregated message, you can consider to use a SimpleMessageGroupProcessor which just produces a Collection<Message<?>> and iterates over them to send to the output one by one.

See more info about aggregator in the Reference Manual: https://docs.spring.io/spring-integration/reference/html/messaging-routing-chapter.html#aggregator