Implementing Polling in Middle of spring Integration flow DSL

Question

I am writing a spring integration DSL flow. Which will look like below diagram.

As you can see in flow I need to read 1 mil entities from database. I want o avoid reading those in single go.

I want to implement polling which will read N entities in fixed interval and send it for processing.

In the examples I read for polling, The polling is used as the first step of the Flow. in my case I want to implement in in middle of the flow.

Please let me know how do I implement this.

Any help is appreciated. Thanks in Advance.

Answer 1

If you want to trigger the start of some polled flow using some external stimulus, use a control bus:

@SpringBootApplication
public class So63337649Application {

    public static void main(String[] args) {
        SpringApplication.run(So63337649Application.class, args);
    }

    @Bean
    IntegrationFlow trigger(ConnectionFactory connectionFactory) {
        return IntegrationFlows.from(Amqp.inboundAdapter(connectionFactory, "foo"))
                .transform(msg -> "@poller.start()")
                .channel("control.input")
                .get();
    }

    @Bean
    IntegrationFlow control() {
        return f -> f.controlBus();
    }

    @Bean
    IntegrationFlow mainFlow() {
        return IntegrationFlows.from(() -> "foo", e -> e
                    .id("poller")
                    .autoStartup(false)
                    .poller(Pollers.fixedDelay(5000)))
                .handle(System.out::println)
                .get();
    }

}