Example Spring integration DSL for JPA Inbound Channel adapter

Question

I can't find a useful example for polling a JPA source for inbound data. I know how to do this in XML but can't figure out how do it in DSL.

In short what I want to do is periodically poll a JPA repository for records then put the records into a flow that will do the usual filtering/transforming/executing.

Kind regards

David Smith

Answer 1

Wire up a JpaPollingChannelAdapter as a @Bean and use

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

See the DSL Reference for configuration options.

This one's near the top (with a different message source).

Answer 2

You are right: there is no yet JPA components support in the Spring Integration Java DSL. Feel free to raise a JIRA (JavaDSL component) on the matter and we'll take care about this demand. Feel free to contribute as well!

Meanwhile I can help you to figure out how to do that without high-level API.

The <int-jpa:inbound-channel-adapter> is based on the JpaPollingChannelAdapter and JpaExecutor objects (exactly them we will use for DSL API). You just must to configure @Bean for JpaExecutor and use it like this:

@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()
       ....
}

Everything else will be done by framework as usual for existing DSL components API.

UPDATE

How to provide auto-startup= property when creating JpaPollingChannelAdapter programmatically? Also, is it possible to get this bean and invoke .start(), .stop() using control-bus?

See, Gary's answer. The Lifecycle control is a responsibility of Endpoint in our case it is SourcePollingChannelAdapter. So, you should specify that second Lambda argument, configure the .autoStartup() and .id() there to be able to inject the SourcePollingChannelAdapter for your JpaPollingChannelAdapter and operate with it for your purpose. That id really can be used from control-bus to start()/stop() at runtime.

Yes, I agree JpaPollingChannelAdapter is unfortunate name for that class because it is really a MessageSource implementation.