Пример интеграции DSL Spring для адаптера входящего канала JPA

Question

Я не могу найти полезный пример для опроса источника JPA для входящих данных. Я знаю, как сделать это в XML, но не могу понять, как это сделать в DSL.

Короче говоря, я хочу периодически опрашивать репозиторий JPA для записей, а затем помещать записи в поток, который будет выполнять обычную фильтрацию / преобразование / выполнение.

С уважением

Дэвид Смит

Answer 1

Вы правы: в Java DSL Spring Integration нет поддержки компонентов JPA. Не стесняйтесь поднять JIRA (JavaDSL компонент) по этому вопросу, и мы позаботимся об этом требовании. Не стесняйтесь вносить свой вклад!

Тем временем я могу помочь вам понять, как это сделать без высокоуровневого API.

<int-jpa:inbound-channel-adapter> основан на JpaPollingChannelAdapter а также JpaExecutor объекты (именно их мы будем использовать для API DSL). Вы просто должны настроить @Bean за JpaExecutor и используйте это так:

@Bean
public JpaExecutor jpaExecutor(EntityManagerFactory entityManagerFactory) {
     JpaExecutor jpaExecutor = new JpaExecutor(entityManagerFactory);
     jpaExecutor.setJpaQuery("from Foo");
     ....
     return jpaExecutor;
}

@Bean
public IntegrationFlow jpaFlow(JpaExecutor jpaExecutor) {
    return IntegrationFlows.from(new JpaPollingChannelAdapter(jpaExecutor))
                      .split()
                      .transform()
       ....
}

Все остальное будет сделано фреймворком, как обычно для существующих API компонентов DSL.

ОБНОВИТЬ

Как обеспечить автозапуск = свойство при программном создании JpaPollingChannelAdapter? Кроме того, возможно ли получить этот bean-компонент и вызвать.start(), .stop(), используя control-bus?

Видишь, ответ Гэри. Lifecycle контроль является обязанностью Endpoint в нашем случае это SourcePollingChannelAdapter, Итак, вы должны указать второй лямбда-аргумент, настроить .autoStartup() а также .id() там, чтобы быть в состоянии ввести SourcePollingChannelAdapter для тебя JpaPollingChannelAdapter и работать с ним для ваших целей. Тот id действительно может быть использован из control-bus в start()/stop() во время выполнения.

Да, я согласен JpaPollingChannelAdapter неудачное название для этого класса, потому что это действительно MessageSource реализация.

Answer 2

Подключите JpaPollingChannelAdapter как @Bean и использовать

IntegrationFlows.from(jpaMessageSource(), 
                      c -> c.poller(Pollers.fixedDelay(1000)))
                .transform(...)
                ...

Смотрите DSL Reference для опций конфигурации.

Этот находится ближе к началу (с другим источником сообщения).