Spring集成流程：在流程中执行任务答案

【问题标题】：Spring integration flow: perform task within flowSpring集成流程：在流程中执行任务
【发布时间】：2016-05-21 03:30:01
【问题描述】：

希望这是我在 Spring 集成中提出的最后一个问题。

面临以下问题：在相当长的 IntegrationFlow dsl 表的末尾有一段代码：

   return IntegrationFlows.
   //...
       .enrichHeaders(headerEnricherSpec -> headerEnricherSpec.header("jms_replyTo", responseQueue(), true)) // IntegrationMessageHeaderAccessor.CORRELATION_ID is not acceptable though message came from outgoingGateway of another application with this header been set
       .handle(requestRepository::save)
       .handle(
            Jms.outboundAdapter(queueConnectionFactory()).destination(serverQueue())
       )
       .get();

问题是像requestRepository::save 处理程序链这样的代码被破坏了。这个技巧只有在有网关作为处理程序参数传入时才有效。

我怎样才能克服这个限制？我认为在这里使用wireTap不会达成交易，因为它是异步的。在这里，实际上，我保存消息以存储它的jms_replyTo 标头，并在相应的回复从服务器返回后将其替换为已保存的消息（智能代理企业集成模式）。

有什么建议吗？

【问题讨论】：

标签： java spring spring-integration dsl

【解决方案1】：

不知道你为什么说“最后一个问题”。你要放弃 Spring Integration 吗？ :-(

我猜你的问题在于下一个.handle()，因为你的requestRepository::save 是单向MessageHandler（void 从save() 方法返回）。或者你的save() 返回null。

IntegrationFlow 是一个执行链，下一个将在前一个之后调用，其结果为非空。

所以，请分享您的requestRepository::save！

更新

两者都没有帮助将 MessageHandler bean 声明为 (m) -> requestRepository.save(m) 并将其作为参数传递给 handle(..) 方法。

是的...我想看看您的requestRepository::save 的签名。

所以，看。使用.handle() 的方法参考，您应该确定您的方案。如果您使用流停止处理one-way，那么org.springframework.messaging.MessageHandler 合约就足够了。你的方法签名应该是这样的：

public void myHandle(Message<?> message)

如果您想继续流程，您应该从您的服务方法中返回任何内容。该结果将成为下一个.handle() 的payload。在这种情况下，您的方法应该遵循org.springframework.integration.dsl.support.GenericHandler 合约。您的方法签名可能如下所示：

public Bar myHandle(Foo foo, Map<String, Object> headers)

这就是 .handle() 的方法引用的工作原理。

您应该了解method chain 样式的工作原理。前一个方法的输出是下一个方法的输入。在这种情况下，我们保护流免受死代码的影响，例如 MessageHandler 和 void 返回，但还有下一个流成员。这就是您看到This is the end of the integration flow. 错误的原因。

【讨论】：

对，请求 repo 的 save 方法没有返回任何内容。我现在没有代码，但它只是将消息保存到数据库和声明为void 的方法中。（不，我说最后一个，因为我有一种感觉，我经常要求。）感谢您的建议，将尝试返回值。
:-)。问没问题！其他很多事情我都感觉不好，所以我也不得不问。另一方面，我的工作是回答类似的问题并帮助人们理解和使用框架。从另一方面来说，像您这样的反馈有助于我们改进框架。就目前而言，文档或 JavaDocs 似乎还不清楚，应该更好地解释这个调用链。
没错，阿尔乔姆。这就是我想注意的。 Spring 框架是一个杰作，它的文档一直很漂亮而且很有帮助。但是 spring 集成似乎比其他模块要复杂一些。文档分别显示了各个方面，但很难理解如何将所有部分组合在一起。我同事的意见都是一样的。如果您稍微改进一下文档来描述如何组织流程，那将非常高兴。（还有一个问题 - 您将来会在本模块中实现任何集成模式吗？）谢谢。
从save 方法返回字符串或布尔值没有意义。我继续使用上面的代码，框架说：The 'currentComponent' (c.e.m.integration.orchestrator.conf.IntegrationConfig$$Lambda$3/632206764@127a7272) is a one-way 'MessageHandler' and it isn't appropriate to configure 'outputChannel'. This is the end of the integration flow.
两者都没有帮助将 MessageHandler bean 声明为 (m) -> requestRepository.save(m) 并将其作为参数传递给 handle(..) 方法。

【解决方案2】：

终于想出了解决办法：

@Bean
public MessageHandler requestPersistingHandler() {
    return new ServiceActivatingHandler(message -> {
        requestRepository.save(message);
        return message.getPayload();
    });
}

//...
@Bean
public IntegrationFlow requestFlow() {
    return IntegrationFlows.from(
            Jms.messageDrivenChannelAdapter(queueConnectionFactory()).destination(bookingQueue())
    )
            .wireTap(controlBusMessageChannel())
            .enrichHeaders(headerEnricherSpec -> headerEnricherSpec.header(JMS_REPLY_HEADER, responseQueue(), true))
            .handle(requestPersistingHandler())
            .handle(
                    Jms.outboundAdapter(queueConnectionFactory()).destination(serverQueue())
            )
            .get();
}

我只是不确定是否有更直接的方法。

剩下的唯一问题是如何在 enrichHeaders 方法中更改“来自服务器”IntegrationFlow 中的标头：不知道如何使用规范访问现有标头。

【讨论】：