CODEWITHSUNDEEP
springspring-integration
Debopam
Debopam

Reputation: 3356

Spring integration service activator with multiple messages

I would like to process multiple messages at a time e.g. get 10 messages from the channel at a time and write them to a log file at once.

Given the scenario, can I write a service activator which will get messages in predefined set i.e. 5 or 10 messages and process it? If this is not possible then how to achieve this using Spring Integration.

Upvotes: 0

Views: 1113

Answers (2)

Artem Bilan
Artem Bilan

Reputation: 121552

That is exactly what you can get with the Aggregator. You can collect several messages to the group using simple expression like size() == 10. When the group is complete, the DefaultAggregatingMessageGroupProcessor emits a single message with the list of payloads of messages in the group. The result you can send to the service-activator for handling the batch at once.

UPDATE

Something like this:

.aggregate(aggregator -> aggregator
                        .correlationStrategy(message -> 1)
                        .releaseStrategy(group -> group.size() == 10)
                        .outputProcessor(g -> new GenericMessage<Collection<Message<?>>>(g.getMessages()))
                        .expireGroupsUponCompletion(true))

So, we correlate messages (group or buffer them) by the static 1 key. The group (or buffer size is 10) and when we reach it we emit a single message which contains all the message from the group. After emitting the result we clean the store from this group to allow to form a new one for a fresh sequence of messages.

Upvotes: 1

Gary Russell
Gary Russell

Reputation: 174749

It depends on what is creating the messages in the first place; if a message-driven channel adapter, the concurrency in that adapter is the key.

For other message sources, you can use an ExecutorChannel as the input channel to the service activator, with an executor with a pool size of 10.

Depending on what is sending messages, you need to be careful about losing messages in the event of a server failure.

It's difficult to provide a general answer without more information about your application.

Upvotes: 0

Related Questions