创建 sftp 回复通道以回复错误或未成功发送的消息

Question

我正在使用 java dsl 配置 sfp 出站流。

网关：

@MessagingGateway
 public interface SftpGateway {
 @Gateway(requestChannel = "sftp-channel")
 void sendFiles(List<Message> messages);
}

配置：

@Bean
public IntegrationFlow sftpFlow(DefaultSftpSessionFactory sftpSessionFactory) {
    return IntegrationFlows
            .from("sftp-channel")
            .split()
            .handle(Sftp.outboundAdapter(sftpSessionFactory, FileExistsMode.REPLACE)
            .useTemporaryFileName(false)
            .remoteDirectory(REMOTE_DIR_TO_CREATE).autoCreateDirectory(true)).get();
}

@Bean
public DefaultSftpSessionFactory sftpSessionFactory() {
...
}

如何配置流程以使我的网关回复失败的消息？ 换句话说，我希望我的网关能够返回失败的消息列表，而不是无效的。

我用 @MessagingGateway(errorChannel = "errorChannel")

并写了错误通道

@Bean
public IntegrationFlow errorFlow() {
    return IntegrationFlows.from("errorChannel").handle(new GenericHandler<MessagingException>() {

        public Message handle(MessagingException payload, Map headers) {
            System.out.println(payload.getFailedMessage().getHeaders());
            return payload.getFailedMessage();
        }
    })
            .get();
}

@Bean
public MessageChannel errorChannel() {
    return MessageChannels.direct().get();
}

如果出现一些错误（即没有连接到 SFTP），我只会收到一个错误（列表中第一条消息的有效负载）。 我应该把 Advice 放在哪里来汇总所有消息？

Answer 1

这不是 Spring Integration Java DSL 的问题。

这主要是一项设计和架构任务。

目前您没有任何选择，因为您使用的是单向Sftp.outboundAdapter()，因此没有任何回复。并且您的 SftpGateway 已准备好使用 void 返回类型进行该行为。

如果您有下游错误，您只能将它们抛出或捕获并发送到某些error-channel。

根据您的要求：

我希望我的网关能够返回失败的消息列表，而不是无效的。

我会说这取决于。实际上它只是从您的网关返回。因此，如果您将一个空列表返回到网关，这可能意味着没有错误。

由于 Java 不提供多返回功能，我们别无选择，除非在我们的流中做一些事情来构建要返回的单个消息。正如我们决定的失败消息列表。

既然您有.split()，您应该查看.aggregate() 以建立一个回复。

聚合器与拆分器的关联非常简单，默认为applySequence = true。

要发送到聚合器，我建议查看Sftp.outboundAdapter() 端点上的ExpressionEvaluatingRequestHandlerAdvice（.handle() 的第二个参数）。这样，您应该将好消息和坏消息发送到同一 .aggregate() 流。比您可以迭代结果列表以从良好结果中清除它。之后的结果可以使用replyChannel 标头发送到SftpGateway。

我知道这听起来有点复杂，但您想要的并不是开箱即用的。需要自己思考和玩弄自己才能弄清楚可以达到的目标。