使用协议适配器

到目前为止，所有显示的示例都说明了 DSL 如何通过使用 Spring Integration 编程模型来支持消息传递架构。但是，我们还没有进行任何真正的集成。这样做需要通过 HTTP、JMS、AMQP、TCP、JDBC、FTP、SMTP 等访问远程资源，或者访问本地文件系统。Spring Integration 支持所有这些以及更多。理想情况下，DSL 应该为所有这些提供一流的支持，但实现所有这些并随着 Spring Integration 中添加新的适配器而保持更新是一项艰巨的任务。因此，预期 DSL 将持续赶上 Spring Integration。

因此，我们提供了高级 API 来无缝定义特定于协议的消息传递。我们通过工厂和构建器模式以及 lambda 来做到这一点。您可以将工厂类视为“命名空间工厂”，因为它们与来自具体协议特定 Spring Integration 模块的组件的 XML 命名空间发挥着相同的作用。目前，Spring Integration Java DSL 支持Amqp、Feed、Jms、Files、(S)Ftp、Http、JPA、MongoDb、TCP/UDP、Mail、WebFlux 和Scripts 命名空间工厂。以下示例显示了如何使用其中的三个（Amqp、Jms 和Mail）

@Bean
public IntegrationFlow amqpFlow() {
    return IntegrationFlow.from(Amqp.inboundGateway(this.rabbitConnectionFactory, queue()))
            .transform("hello "::concat)
            .transform(String.class, String::toUpperCase)
            .get();
}

@Bean
public IntegrationFlow jmsOutboundGatewayFlow() {
    return IntegrationFlow.from("jmsOutboundGatewayChannel")
            .handle(Jms.outboundGateway(this.jmsConnectionFactory)
                        .replyContainer(c ->
                                    c.concurrentConsumers(3)
                                            .sessionTransacted(true))
                        .requestDestination("jmsPipelineTest"))
            .get();
}

@Bean
public IntegrationFlow sendMailFlow() {
    return IntegrationFlow.from("sendMailChannel")
            .handle(Mail.outboundAdapter("localhost")
                            .port(smtpPort)
                            .credentials("user", "pw")
                            .protocol("smtp")
                            .javaMailProperties(p -> p.put("mail.debug", "true")),
                    e -> e.id("sendMailEndpoint"))
            .get();
}

前面的示例显示了如何将“命名空间工厂”用作内联适配器声明。但是，您可以从@Bean 定义中使用它们，以使IntegrationFlow 方法链更易于阅读。

在我们将精力投入到其他方面之前，我们正在征求社区对这些命名空间工厂的反馈。我们也感谢您对我们接下来应该支持哪些适配器和网关进行优先级排序的任何意见。

您可以在本参考手册中各个协议特定的章节中找到更多 Java DSL 示例。

所有其他协议通道适配器都可以配置为通用 bean 并连接到IntegrationFlow，如下例所示

@Bean
public QueueChannelSpec wrongMessagesChannel() {
    return MessageChannels
            .queue()
            .wireTap("wrongMessagesWireTapChannel");
}

@Bean
public IntegrationFlow xpathFlow(MessageChannel wrongMessagesChannel) {
    return IntegrationFlow.from("inputChannel")
            .filter(new StringValueTestXPathMessageSelector("namespace-uri(/*)", "my:namespace"),
                    e -> e.discardChannel(wrongMessagesChannel))
            .log(LoggingHandler.Level.ERROR, "test.category", m -> m.getHeaders().getId())
            .route(xpathRouter(wrongMessagesChannel))
            .get();
}

@Bean
public AbstractMappingMessageRouter xpathRouter(MessageChannel wrongMessagesChannel) {
    XPathRouter router = new XPathRouter("local-name(/*)");
    router.setEvaluateAsString(true);
    router.setResolutionRequired(false);
    router.setDefaultOutputChannel(wrongMessagesChannel);
    router.setChannelMapping("Tags", "splittingChannel");
    router.setChannelMapping("Tag", "receivedChannel");
    return router;
}