Using Control Bus EIP in Spring Integration to start/stop channels dynamically

I am interested in using Spring Integration to fetch files from various endpoints (FTP servers, email inboxes, S3, etc.) and load them into my system (essentially, ETL).

There are times when I will want these channels active and running, and other times when I will want them paused/stopped. Meaning, even if there are files available at the source, I do not want the channel consuming the data and doing anything with it.

Is a control bus an appropriate start/stop solution here:

@Bean
public IntegrationFlow controlBusFlow() {
    return IntegrationFlow.from("controlBus")
              .controlBus()
              .get();
}

If so, how would I stop/restart a specific channel (route between an S3 bucket and the rest of my system) using the Java DSL/API? And if not, then what is the recommended practice/EIP to apply here?

>Solution :

Yes, the Control Bus is exactly a pattern and tool designed for your goal: https://www.enterpriseintegrationpatterns.com/ControlBus.html.

Yes, to use it you need to send messages to input channel of that control bus endpoint. The payload of message to sent must be a command to do some control activity for endpoint. Typically we call start and stop.

So, let’s imagine you have an S3 source polling channel adapter:

@Bean
IntegrationFlow s3Flow(S3InboundFileSynchronizingMessageSource s3InboundFileSynchronizingMessageSource) {
    return IntegrationFlow.from(s3InboundFileSynchronizingMessageSource, e -> e.id("myS3SourceEndpoint"))
          ...;
}

So, to stop that myS3SourceEndpoint via Control Bus, you need to send a message with a payload @myS3SourceEndpoint.stop().

Pay attention that we don’t talk here about message channels neither message sources. The active components in the flow are really endpoints.