What is the recommended way to communicate between IntergrationFlow and services?

Question

I'm looking into spring integration, more specifically the Java DSL.

The DSL is presented by the IntegrationFlows factory for the IntegrationFlowBuilder. This produces the IntegrationFlow component, which should be registered as a Spring bean (by using the @Bean annotation). The builder pattern is used to express arbitrarily complex structures as a hierarchy of methods that can accept lambdas as arguments.

In my mind then, I define a configuration class with a @Bean.

    @Bean
    fun ftpInboundIntegrationFlow(ftpSf: DefaultFtpSessionFactory?): IntegrationFlow {
        val ftpSpecification = Ftp
            .inboundStreamingAdapter(template())
            .patternFilter("*.csv")
            .remoteDirectory(fromFolder)

        return IntegrationFlows
            .from(ftpSpecification) { pc: SourcePollingChannelAdapterSpec ->
                pc.poller { pm: PollerFactory ->
                    pm.cron(cronSyncExpression)
                }
            }
            .transform(StreamTransformer())
            .channel(doSomeBusinessLogic)
            .get()
    }

But I am rather confused how to, in my more specific service layer, handle this message.

I could, and it seems to work, create another @Bean in the service class.

    @Bean
    fun handleDoSomeBusinessLogicFlow(): IntegrationFlow {
        return IntegrationFlows.from("doSomeBusinessLogic")
            .handle { customers: List<SomeDomainModel>, headers: MessageHeaders ->
                //Some repository and service logic
            }
            .get()
    

I tried looking at examples (https://github.com/spring-projects/spring-integration-samples/tree/833f55d662fb17edda6bcb000d952a3d00aa1dd8). But almost all of them are just creating all the logic in the same place. It feels strange to define a @Bean in the @Service class.

Answer 1

You are right: you can just have a business method in your service and use it from the .handle() instead of channel. You also can just use a @ServiceActivator on that method if you’d prefer loosely coupled distribution via channels. There is also an IntegrationFlowAdapter to have a service incapsulation, but still gain from flow definition.