我正在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
中设置。为此,我改变了keyExpression
s3UploadMessageHandler
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
,messageHandler
类类型不同:它是一个S3MessageHandler
没有 lambda 的MyCallingClass$lambda
调用时,一个使用 lambda 表达式调用的时候。
我错过了什么让我的场景工作?