使用协议适配器

到目前为止展示的所有示例都说明了 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;
}