Spring Integration 轮询目录

Question

我有一个用例，我需要将 .fif 文件从本地上传到指定目录结构中的远程 sftp 服务器。如下：

现在其他一些服务器/应用程序将处理这些 .fif 文件并生成 .npz 文件并将其放置在相应访问/id 的输出文件夹中。

我可以使用 sftpOutboundAdapter 上传文件，它工作正常。

现在，我不知道如何添加轮询以知道在此访问的此输出目录中创建了输出文件。我不想下载输出文件夹中的文件。我只想知道创建了 4 个文件，以便我可以更新访问状态。

我目前的代码是

@Value("${sftp.host}")
private String host;

@Value("${sftp.port:22}")
private int port;

@Value("${sftp.user}")
private String user;

@Value("${sftp.privateKey:#{null}}")
private Resource privateKey;

@Value("${sftp.privateKeyPassphrase:}")
private String privateKeyPassphrase;

@Value("${sftp.password:#{null}}")
private String password;

private final FileService fileService;

@Autowired
public SftpConfig(FileService fileService) {
    this.fileService = fileService;
}

@Bean
public SessionFactory<LsEntry> sftpSessionFactory() {
    DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
    factory.setHost(host);
    factory.setPort(port);
    factory.setUser(user);
    if (privateKey != null) {
        factory.setPrivateKey(privateKey);
        factory.setPrivateKeyPassphrase(privateKeyPassphrase);
    } else {
        factory.setPassword(password);
    }
    factory.setAllowUnknownKeys(true);
    return new CachingSessionFactory<LsEntry>(factory);
}

@Bean
public IntegrationFlow sftpOutboundFlow() {
    return IntegrationFlows.from("toSftpChannel")
            .handle(Sftp.outboundAdapter(this.sftpSessionFactory(), FileExistsMode.FAIL)
                            .remoteDirectoryExpression("headers['path']")
                            .useTemporaryFileName(false)
                            .autoCreateDirectory(true)
                            .fileNameGenerator(message -> (String) message.getHeaders().get("fileName"))
                    , c -> c.advice(expressionAdvice())
            ).get();
}

@Bean
public Advice expressionAdvice() {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
    advice.setSuccessChannelName("integrationFlow.input");
    advice.setOnSuccessExpressionString("payload");
    advice.setFailureChannelName("integrationFlow.input");
    advice.setOnFailureExpressionString("payload");
    advice.setTrapException(true);
    return advice;
}

@Bean
public IntegrationFlow integrationFlow() {
    return f -> f.handle((MessageHandler) fileService::OnFilesUpload);
}

@MessagingGateway
public interface UploadGateway {

    @Gateway(requestChannel = "toSftpChannel")
    void upload(@Payload File file, @Header("fileName") String fileName, @Header("path") String path,
                @Header("parentId") Long parentId);

}

请给出我如何轮询父目录（即 Providers）的 java 配置，并知道在其中创建 .npz 文件的各个访问/ID 的输出文件夹的路径。然后获取创建的文件数或文件列表。

Answer 1

在 LS 命令中使用outbound gateway（如果需要，它有一个递归选项）。

只需将远程目录发送到网关，它就会返回列表。

编辑

在下面回答您的评论；你没有显示你的调用代码，但是为什么你不能做这样的事情......

@MessagingGateway
public interface UploadGateway {

    @Gateway(requestChannel = "toSftpChannel")
    void upload(@Payload File file, @Header("fileName") String fileName, @Header("path") String path,
            @Header("parentId") Long parentId);

    @Gateway(requestChannel = "toSftpLsGatewayChannel")
    List<LsEntry> getOutputFiles(String path);

}

和

this.gateway.upload(file, fileName, path, parentId);
List<LsEntry> outputs;
String outputPath = path.replace("/input", "/output");
do {
    outputs = this.gateway.getOutputFiles(outputPath);
    Thread.sleep(5_000);
} while (outputs == null || outputs.size() == 0);

如果您希望它异步运行，只需在任务执行器上运行第二部分即可。