Spring Integration Interaction

Spring Integration Framework扩展了 Spring 编程模型，以支持著名的企业集成模式。它能够在基于 Spring 的应用程序中进行轻量级消息传递，并支持通过声明性适配器与外部系统进行集成。它还提供高级 DSL，可以将各种操作（端点）组合成逻辑集成流。借助此 DSL 配置的 lambda 样式，Spring Integration 已经获得了相当好的 java.util.function`接口采用度。@MessagingGateway`代理界面还可以作为 `Function`或 `Consumer`出现，根据 Spring Cloud 函数环境，代理界面可以注册到函数目录中。有关 Spring Integration 如何支持函数的更多信息，请参阅 Spring Integration ReferenceManual。

另一方面，从版本 4.0.3 开始，Spring Cloud Function 引入了 spring-cloud-function-integration 模块，该模块从 Spring Integration DSL 角度为与 FunctionCatalog 交互提供了更深层次、更特定于云且基于自动配置的 API。FunctionFlowBuilder 是使用 FunctionCatalog 自动配置和自动装配的，并且表示目标 IntegrationFlow 实例针对函数的 DSL 的入口点。除了标准 IntegrationFlow.from() 工厂（为方便起见）外，FunctionFlowBuilder 还公开了一个 fromSupplier(String supplierDefinition) 工厂，用于在提供的 FunctionCatalog 中查找目标 Supplier。然后，此 FunctionFlowBuilder 会将 FunctionFlowDefinition 导致结果。此 FunctionFlowDefinition 是 IntegrationFlowExtension 的实现，并公开 apply(String functionDefinition) 和 accept(String consumerDefinition) 运算符，用于分别从 FunctionCatalog 中查找 Function 或 Consumer。有关详细信息，请参阅其 Javadocs。

以下示例展示了 FunctionFlowBuilder 的实际操作以及其余 IntegrationFlow API 的强大功能：

@Configuration
public class IntegrationConfiguration {

    @Bean
    Supplier<byte[]> simpleByteArraySupplier() {
        return "simple test data"::getBytes;
    }

    @Bean
    Function<String, String> upperCaseFunction() {
        return String::toUpperCase;
    }

    @Bean
    BlockingQueue<String> results() {
        return new LinkedBlockingQueue<>();
    }

    @Bean
    Consumer<String> simpleStringConsumer(BlockingQueue<String> results) {
        return results::add;
    }

    @Bean
    QueueChannel wireTapChannel() {
        return new QueueChannel();
    }

    @Bean
    IntegrationFlow someFunctionFlow(FunctionFlowBuilder functionFlowBuilder) {
        return functionFlowBuilder
                .fromSupplier("simpleByteArraySupplier")
                .wireTap("wireTapChannel")
                .apply("upperCaseFunction")
                .log(LoggingHandler.Level.WARN)
                .accept("simpleStringConsumer");
    }

}

由于 FunctionCatalog.lookup() 功能不仅限于简单的函数名称，因此上述 apply() 和 accept() 运算符中还可以使用函数组合功能：

@Bean
IntegrationFlow functionCompositionFlow(FunctionFlowBuilder functionFlowBuilder) {
    return functionFlowBuilder
            .from("functionCompositionInput")
            .accept("upperCaseFunction|simpleStringConsumer");
}

当我们在 Spring Cloud 应用程序中添加用于预定义函数的自动配置依赖项时，此 API 将变得更加相关。例如， Stream Applications项目除了应用程序映像之外，还提供具有适用于各种集成用例的函数的工件，例如 debezium-supplier、elasticsearch-consumer、`aggregator-function`等。

以下配置分别基于 http-supplier、spel-function 和 file-consumer：

@Bean
IntegrationFlow someFunctionFlow(FunctionFlowBuilder functionFlowBuilder) {
    return functionFlowBuilder
            .fromSupplier("httpSupplier", e -> e.poller(Pollers.trigger(new OnlyOnceTrigger())))
            .<Flux<?>>handle((fluxPayload, headers) -> fluxPayload, e -> e.async(true))
            .channel(c -> c.flux())
            .apply("spelFunction")
            .<String, String>transform(String::toUpperCase)
            .accept("fileConsumer");
}

我们只需要将其配置添加到 application.properties 中（如有必要）：

http.path-pattern=/testPath
spel.function.expression=new String(payload)
file.consumer.name=test-data.txt