本地删除后未使用 Spring Integration 下载 FTP 文件

Question

我们正在编写一个批处理作业，它将文件作为来自 FTP 的输入，生成一些新文件并将它们写入 S3 存储桶，为此我们使用 Spring Integration。

FTP 中的文件是从数据库中提取的，每晚都会更新。

问题是，当我们第一次启动应用程序时，它连接良好的FTP，下载文件，并上传生成结果S3。然后我们在本地删除下载的文件，等待FTP中的文件的下一代重新启动该过程。但它永远不会再次下载文件。

有什么想法吗？

    @Bean
    public IntegrationFlow ftpInboundFlow() {
        return IntegrationFlows
                .from(ftpReader(),
                        spec -> spec.id("ftpInboundAdapter")
                                .autoStartup(true)
                                .poller(Pollers.fixedDelay(period)))
                .enrichHeaders(Map.of("CORRELATION_ID", "rcm"))
                .aggregate(aggregatorSpec -> aggregatorSpec
                        .correlationStrategy(message -> message.getHeaders().get("CORRELATION_ID"))
                        .releaseStrategy(group -> group.getMessages().size() == 2))
                .transform(stockUnmarshaller)
                .transform(stockTransformer)
                .transform(stockMarshaller)
                .transform(picturesDownloader)
                .transform(picturesZipper)
                .transform(stockIndexer)
                .handle(directoryCleaner)
                .nullChannel();
    }

    @Bean
    public FtpInboundChannelAdapterSpec ftpReader() {
        return Ftp.inboundAdapter(ftpSessionFactory())
                .preserveTimestamp(true)
                .remoteDirectory(rootFolder)
                .autoCreateLocalDirectory(true)
                .localDirectory(new File(localDirectory));
    }

    @Bean
    public SessionFactory<FTPFile> ftpSessionFactory() {
        DefaultFtpSessionFactory sessionFactory = new DefaultFtpSessionFactory();
        sessionFactory.setHost(host);
        sessionFactory.setUsername(userName);
        sessionFactory.setPassword(password);
        sessionFactory.setClientMode(FTPClient.PASSIVE_LOCAL_DATA_CONNECTION_MODE);
        return sessionFactory;
    }

提前致谢。

编辑：

我使用enrichHeaders 来确保在我们正好有 2 个文件时触发管道。也许标题没有被删除并且条件总是大于2？也许这是错误的方式？

再次感谢。

Answer 1

听起来你在谈论同一个文件。在这种情况下，从本地目录中删除它是不够的。进程中涉及一些FileListFilter 实例，这些实例包含已处理文件的条目。并且根据您的配置，您可以处理内存中的变体。他们真的对您的本地文件删除一无所知。

确切地说，您需要担心两个过滤器：FtpPersistentAcceptOnceFileListFilter 用于远程条目，FileSystemPersistentAcceptOnceFileListFilter 用于文件的本地副本。他们都在实现ResettableFileListFilter，所以，当你完成文件处理时，你可以调用他们的remove()。

Java DSL 中的FtpInboundChannelAdapterSpec 有以下选项：

/**
 * Configure a {@link FileListFilter} to be applied to the remote files before
 * copying them.
 * @param filter the filter.
 * @return the spec.
 */
public S filter(FileListFilter<F> filter) {

/**
 * A {@link FileListFilter} used to determine which files will generate messages
 * after they have been synchronized.
 * @param localFileListFilter the localFileListFilter.
 * @return the spec.
 * @see AbstractInboundFileSynchronizingMessageSource#setLocalFilter(FileListFilter)
 */
public S localFilter(FileListFilter<File> localFileListFilter) {

因此，您仍然可以将那些提到的过滤器作为默认过滤器，但您将它们作为 bean 提取并注入这些选项和您的 directoryCleaner 以从这些过滤器中删除。

还有一个选项，如：

/**
 * Switch the local {@link FileReadingMessageSource} to use its internal
 * {@code FileReadingMessageSource.WatchServiceDirectoryScanner}.
 * @param useWatchService the {@code boolean} flag to switch to
 * {@code FileReadingMessageSource.WatchServiceDirectoryScanner} on {@code true}.
 * @since 5.0
 */
public void setUseWatchService(boolean useWatchService) {

并且DELETE 事件也被配置为观察者。发生这种情况时，已删除的文件也会从本地过滤器中删除。

您也可以在配置时正确处理远程文件：

/**
 * Set to true to enable the preservation of the remote file timestamp when transferring.
 * @param preserveTimestamp true to preserve.
 * @return the spec.
 */
public S preserveTimestamp(boolean preserveTimestamp) {

这样，具有相同名称的较新文件将被视为不同的文件，并且其在上述过滤器中的条目将被覆盖。虽然我看到你已经在使用它了，但是你仍然抱怨它不起作用。当 FileSystemPersistentAcceptOnceFileListFilter 不用于本地文件时，某些旧版本的 Spring Integration 可能会出现这种情况。

Answer 2

入站通道适配器有两个过滤器.filter 和.localFilter。

第一个在下载前过滤远程文件，第二个过滤文件系统上的文件。

默认情况下，filter 是 FtpPersistentAcceptOnceFileListFilter，它只会获取新的或更改的文件。

默认情况下，localFilter 是 FileSystemPersistentAcceptOnceFileListFilter，如果文件的时间戳已更改，它只会再次传递文件。

所以文件只有在时间戳改变时才会被重新处理。

我建议你在调试器中运行，看看为什么它没有通过过滤器。