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

Question

我尝试将集成咖啡馆演示转换为Java8 DSL，在官方Spring集成示例中有一些现有的示例。

我想把这些例子中所有好的部分结合起来。

• Java8 DSL而不是XML
• AMQP在较小的流中滑动工作流。
• 用Jackson处理JSON中的载荷

代码（不是预期的工作，一些问题）在这里。

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

并且在应用时，会抱怨“需要回复通道或输出通道”。

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

更新1：当我写coldDrinks流时感到困惑。

XML来自原始的caf-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>

如何有效地将其转换为JavaDSL。我把我的想法添加到内联

 @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示例中，我没有在那里设置回复通道，但消息返回到roscket路由，并由客户端（OutboundGateway）接收。

更新：终于工作了，请检查这里。当我试图使用Amqp发送和接收对象消息时，我在这里遇到了一个问题，抛出了一些类强制转换异常，当使用<代码>句柄etcMessageHandler时，标题中的TypeId没有改变，必须像基于xml的caf-amqp那样转换bwteenjson/对象才能使其最终工作。这里缺少什么？

Answer 1

该回购中已经有一个官方JavaDSL样本：https://github.com/spring-projects/spring-integration-samples/tree/master/dsl/cafe-dsl.

是的，不存在AMQP变体，但是如果您遵循来自sping-整合-amqp的Amqp工厂，那么将该XML配置转换为JavaDSL应该不会太难。

不确定您在测试中使用的“坏选项”是什么，但在spring集成测试中我们提供了足够的UTIL来简化流测试：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();
}

没有类似的回复频道部分。这就是为什么需要回复通道或输出通道的原因，因为您从路由发送，该路由不需要任何回复。也许你还没有完成整个逻辑的应用程序设计。。。