why "orderly shutdown" doesn't stop pollable message sources?

user_2038596 0
Viewed 40

I've having a hard time to understand why IntegrationMBeanExporter.stopActiveComponents method doesn't take care of stopping pollers like below.
What is the rational for not stopping below integrationflow from producing new messages after stopActiveComponents is called ?
(of course I can manually shutdown that poller myself but why isn't it handled by the framework ?)
Any way to modify below code so that the poller automatically stops sending new messages after stopActiveComponents method is called ?

IntegrationFlows
    .from(
        () -> MessageBuilder.<Object>withPayload("").build(),
        c -> c.poller(Pollers.fixedDelay(Duration.ofSeconds(10)))
    )
    .transform(p -> p)
    .log(Level.INFO, m-> "done")
    .get()
)

Thank you very much in advance for your time and your expertise.
Best Regards

spring-integration-dsl spring-integration
1 Answers
.from(
    () -> MessageBuilder.<Object>withPayload("").build(),

Produces a plain MessageSource into the target SourcePollingChannelAdapter:

        SourcePollingChannelAdapter spca = new SourcePollingChannelAdapter();
        spca.setSource(this.source);

(See SourcePollingChannelAdapterFactoryBean).

The IntegrationMBeanExporter, in turn, expects this:

else if (bean instanceof SourcePollingChannelAdapter) {
        SourcePollingChannelAdapter pollingChannelAdapter = (SourcePollingChannelAdapter) bean;
        MessageSource<?> messageSource = pollingChannelAdapter.getMessageSource();
        if (messageSource instanceof IntegrationInboundManagement) {
            IntegrationInboundManagement monitor = (IntegrationInboundManagement) extractTarget(messageSource);
            registerSource(monitor);
            this.sourceLifecycles.put(monitor, pollingChannelAdapter);
            this.runtimeBeans.add(monitor);
            return;
        }
    }

So, only IntegrationInboundManagement impl is considered as a lifecycle candidate for stopping eventually.

To fix your code just do this instead:

.fromSupplier(
    () -> "",
    c -> c.poller(Pollers.fixedDelay(Duration.ofSeconds(10)))
)

And the Framework will do the proper wrapping for you!

user_27565470
Related