пружинная интеграция буфера dsl

Question

У меня есть требование, когда мне нужно хранить / буферизовать сообщения, полученные по каналу и сохраняемые в базе данных на основе количества сообщений или тайм-аута, что означает отсутствие сообщений в течение 1 минуты. Есть ли способ добиться этого весной интеграции

IntegrationFlows
                .from(Jms.messageDrivenChannelAdapter(connectionFactory)
                    .destination(sourceQueue))
                .transform(someTransform, "transform")
                .handle(someService, "save")
                .get();

Answer 1

Существует оператор .aggregate(), основанный на реализации шаблона Aggregator EI.

Это можно настроить с помощью JdbcMessageStore для буферизации сообщений и их сохранения в БД.

Вы можете удерживать их там до некоторого состояния через ReleaseStrategy (в зависимости от каждого приходящего сообщения) или освобождать их из-за group timeout.

Если вас не интересует, чтобы все они впоследствии были единым агрегированным сообщением, вы можете использовать SimpleMessageGroupProcessor, который просто создает Collection<Message<?>> и перебирает их для отправки на выход по одному.

Подробнее об агрегаторе см. В Справочном руководстве: https://docs.spring.io/spring-integration/reference/html/messaging-routing-chapter.html#aggregator