为什么我的处理程序方法在定义为 lambda 时没有被触发？

Question

我正在定义一个 IntegrationFlow 以通过这种方式使用 DSL 语法从 SFTP 流式传输到 S3：

return IntegrationFlows.from(Sftp.inboundStreamingAdapter(remoteFileTemplate)
                        .remoteDirectory("remoteDirectory"),
                e -> e.poller(Pollers.fixedDelay(POLL, TimeUnit.SECONDS)))
                .transform(new StreamTransformer())
                .handle(s3UploadMessageHandler(outputFolderPath, "headers['file_remoteFile']")) // Upload on S3
                .get();

private S3MessageHandler s3UploadMessageHandler(String folderPath, String spelFileName) {
        S3MessageHandler s3MessageHandler = new S3MessageHandler(amazonS3, s3ConfigProperties.getBuckets().getCardManagementData());
        s3MessageHandler.setKeyExpression(new SpelExpressionParser().parseExpression(String.format("'%s/'.concat(%s)", folderPath, spelFileName)));
        s3MessageHandler.setCommand(S3MessageHandler.Command.UPLOAD);
        return s3MessageHandler;
    }

它按预期工作：文件已很好地上传到我的 S3 存储桶。但是，我想避免SPEL 语法，并将消息中的标头注入s3uploadMessageHandler 方法，这样我可以使用简单的ValueExpression 在s3UploadMessageHandler 方法中设置keyExpression。 为此，我改变了

handle(s3UploadMessageHandler(outputFolderPath, "headers['file_remoteFile']")) // Upload on S3

到

handle(m -> s3UploadMessageHandler(outputFolderPath, (String) m.getHeaders().get("file_remoteFile"))) // Upload on S3

但是现在这个处理程序似乎不再被触发了。日志中没有错误，我从日志中知道 SFTP 轮询仍在工作。

我试图找到这背后的原因，我看到在IntegrationFlowdefinition.java中输入handle方法时，messageHandler的类类型是不同的：它是一个S3MessageHandler，在没有lambda的情况下调用，一个MyCallingClass$lambda使用 lambda 表达式调用时。

为了让我的场景正常工作，我错过了什么？

Answer 1

有两种方法可以处理消息。一种是通过MessageHandler 实现——这是最有效的方法，并且在通道适配器实现的框架中完成，例如S3MessageHandler。另一种方法是 POJO 方法调用 - 当您不需要担心任何框架接口时，这是最用户友好的方法。

因此，当您像这样使用.handle(s3UploadMessageHandler(...)) 时，您指的是MessageHandler，并且框架知道必须注册该MessageHandler 的bean，因为您的s3UploadMessageHandler() 不是@Bean。

当您将其用作 lambda 时，框架将其视为 POJO 方法调用，并且为 MethodInvokingMessageHandler 注册了一个 bean，但不是您的 S3MessageHandler。

无论如何，即使您将s3UploadMessageHandler() 更改为@Bean 方法，它也不会起作用，因为您不允许框架调用S3MessageHandler.handleMessage()。您在这里所做的只是在运行时调用 private 方法来针对每个请求消息创建一个 S3MessageHandler 实例：MethodInvokingMessageHandler 在其 handleMessage() 中调用您的 lambda，仅此而已 - S3 不会发生任何事情。

ValueExpression 在这里无法为您提供帮助，因为您需要针对每条请求消息评估目标文件。因此，您需要一个运行时表达式。 new SpelExpressionParser().parseExpression() 确实没有问题。仅仅因为我们别无选择，只能有一个无状态的S3MessageHandler，并且不要像您尝试使用可疑的 lambda 和 ValueExpression 实现的每个请求在运行时重新创建它。