将 Spring Integration Cafe 演示从 xml 配置转换为 Java 8 DSL

Question

我尝试将集成咖啡馆演示转换为Java 8 DSL，官方Spring Integration Samples中有一些现有示例。

我想结合这些例子的所有优点。

• Java 8 DSL 代替 XML
• AMQP 将工作流拆分为较小的流程。
• 使用 Jackson 处理 JSON 格式的负载。

代码（不按预期工作，有些问题）是here。

在官方 XML 示例中，很容易在 Amqp 网关中设置请求/回复通道。但是转到 Java 8 DSL，缺少选项。

并且在申请的时候会报错“需要回复通道或者输出通道”。

顺便说一句，有没有更好的选项来调试/单元测试 Spring 集成应用程序？

更新1：我在写coldDrinks流程的时候很困惑。

XML 来自原始 cafe-amqp 项目。

<!-- To receive an AMQP Message from a Queue, and respond to its reply-to address, configure an inbound-gateway. -->
    <int-amqp:inbound-gateway
        id="coldDrinksBarista"
        request-channel="coldJsonDrinks"
        queue-names="cold-drinks"
        connection-factory="rabbitConnectionFactory" />

    <int:chain input-channel="coldJsonDrinks">
        <int:json-to-object-transformer type="org.springframework.integration.samples.cafe.OrderItem"/>
        <int:service-activator method="prepareColdDrink">
            <bean class="org.springframework.integration.samples.cafe.xml.Barista"/>
        </int:service-activator>
        <int:object-to-json-transformer content-type="text/x-json"/>
    </int:chain>

如何有效地将其转换为 Java DSL。 我在内联添加了我的想法

 @Bean
    public IntegrationFlow coldDrinksFlow(AmqpTemplate amqpTemplate) {
        return IntegrationFlows
                .from("coldDrinks")
                .handle(
                        Amqp.outboundGateway(amqpTemplate)
                                .exchangeName(TOPIC_EXCHANGE_CAFE_DRINKS)
                                .routingKey(ROUTING_KEY_COLD_DRINKS)
                )
                .log("coldDrinksFlow")
                .channel(preparedDrinksChannel())
                .get();
    }

    @Bean
    public IntegrationFlow coldDrinksBaristaFlow(ConnectionFactory connectionFactory, Barista barista) {
        return IntegrationFlows
                .from(Amqp.inboundGateway(connectionFactory, QUEUE_COLD_DRINKS)
                        .configureContainer(
                                c -> c.receiveTimeout(10000)
                        )// If setup replyChannel the below `handle` is not worked as expected.


                )
                .handle(OrderItem.class, (payload, headers) -> (Drink) barista.prepareColdDrink(payload))
//If adding a channel here, the flow will NOT return back the `coldDrinksFlow` will cause another exception, "its requiresReply is set to true..."
                .get();
    }

在我之前的经验中，我想通过协议（HTTP、FTP 等）作为小流中的边缘（开始时入站，结束时出站）来打破整个流程。入站/出站网关很容易设置为工作而无需设置回复通道等，默认情况下它应该使用其内置协议而不是通道通过原始路由回复。在我的inboundGateway RSocket example 中，我没有在那里设置回复通道，但是消息返回到 roscket 路由并由client side(outboudGateway) 接收。

更新：终于成功了，检查here。我在尝试使用 Amqp 发送和接收对象消息时遇到的一个问题，抛出了一些类强制转换异常，使用 handle 等 MessageHandler 时，标头中的 TypeId 没有改变，有像基于 xml 的 cafe-amqp 那样转换 bwteen json/object 以使其最终起作用。这里缺少什么？

Answer 1

该存储库中已经有一个官方 Java DSL 示例：https://github.com/spring-projects/spring-integration-samples/tree/master/dsl/cafe-dsl。

是的，没有 AMQP 变体，但如果您从 spring-integration-amqp 遵循 Amqp 工厂，那么将该 XML 配置转换为 Java DSL 应该不会太难。

不确定您使用的测试“坏选项”是什么，但在spring-integration-test 中，我们提供了足够的实用程序来简化流测试：https://docs.spring.io/spring-integration/docs/5.3.2.RELEASE/reference/html/testing.html#testing

更新

你的问题在这里：

<int-amqp:outbound-gateway
    id="coldDrinksBarista"
    request-channel="coldDrinks"
    reply-channel="preparedJsonDrinks"
    exchange-name="cafe-drinks"
    routing-key="drink.cold"
    amqp-template="amqpTemplate" />

你的配置在哪里：

@Bean
public IntegrationFlow coldDrinksFlow(AmqpTemplate amqpTemplate) {
    return IntegrationFlows
            .from("coldDrinks")
            .handle(
                    Amqp.outboundGateway(amqpTemplate)
                            .exchangeName(TOPIC_EXCHANGE_CAFE_DRINKS)
                            .routingKey(ROUTING_KEY_COLD_DRINKS)
            )
            .get();
}

没有类似的replyChannel 部分。这就是为什么您会收到 reply channel or output channel is required 的原因，因为您是从 route 发送的，它不希望得到任何回复。可能您还没有完成为整个逻辑设计应用程序...