FTP Streaming Inbound Channel Adapter

版本 4.3 引入了流入站通道适配器。此适配器会生成带有 InputStream`类型负载的消息，允许获取文件而无需写入本地文件系统。由于会话保持打开状态，因此消耗应用程序负责在文件消耗完毕后关闭会话。会话是在 `closeableResource`头（`IntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCE）中提供的。FileSplitter`和 `StreamTransformer`等标准框架组件会自动关闭会话。有关这些组件的更多信息，请参阅 File Splitter和 Stream Transformer。以下示例显示了如何配置 `inbound-streaming-channel-adapter：

<int-ftp:inbound-streaming-channel-adapter id="ftpInbound"
            channel="ftpChannel"
            session-factory="sessionFactory"
            filename-pattern="*.txt"
            filename-regex=".*\.txt"
            filter="filter"
            filter-expression="@myFilterBean.check(#root)"
            remote-file-separator="/"
            comparator="comparator"
            max-fetch-size="1"
            remote-directory-expression="'foo/bar'">
        <int:poller fixed-rate="1000" />
</int-ftp:inbound-streaming-channel-adapter>

只允许 filename-pattern、filename-regex、filter 或 filter-expression 中的一个。

从 5.0 版本开始，默认情况下 FtpStreamingMessageSource 适配器会基于内存中的 SimpleMetadataStore 来防止 FtpPersistentAcceptOnceFileListFilter 的远程文件出现重复。默认情况下，此过滤器也会与文件名模式（或正则表达）一起应用。如果您需要允许重复，则可以使用 AcceptAllFileListFilter。任何其他用例都可以由 CompositeFileListFilter（或 ChainFileListFilter）来处理。Java 配置（later in the document）展示了一种在处理后移除远程文件以避免重复的技术。

有关 FtpPersistentAcceptOnceFileListFilter 的详细信息以及如何使用它，请参见 Remote Persistent File List Filters。使用 max-fetch-size 属性来限制每次轮询获取的文件数（如果有必要）。在集群环境中运行时，将其设置为 1 并使用持久过滤器。有关详细信息，请参阅 Inbound Channel Adapters: Controlling Remote File Fetching。适配器分别将远程目录和文件名放在 FileHeaders.REMOTE_DIRECTORY 和 FileHeaders.REMOTE_FILE 标头中。从 5.0 版开始，FileHeaders.REMOTE_FILE_INFO 标头提供其他远程文件信息（默认情况下以 JSON 表示）。如果你将 FtpStreamingMessageSource 上的 fileInfoJson 属性设置为 false，标头将包含一个 FtpFileInfo 对象。该 FtpFileInfo 对象由基础 Apache Net 库提供，可以使用 FtpFileInfo.getFileInfo() 方法访问它。使用 XML 配置时，fileInfoJson 属性不可用，但你可以通过将 FtpStreamingMessageSource 注入到你的某个配置类中来设置该属性。另请参阅 Remote File Information。从版本 5.1 开始，comparator 的泛型类型为 FTPFile。以前，它是 AbstractFileInfo<FTPFile>。这是因为现在在过滤和应用 maxFetch 之前，排序在处理中执行得更早。

Configuring with Java Configuration

以下 Spring Boot 应用展示了如何通过 Java 配置来配置入站适配器的一个示例：

@SpringBootApplication
public class FtpJavaApplication {

    public static void main(String[] args) {
        new SpringApplicationBuilder(FtpJavaApplication.class)
            .web(false)
            .run(args);
    }

    @Bean
    @InboundChannelAdapter(channel = "stream")
    public MessageSource<InputStream> ftpMessageSource() {
        FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template());
        messageSource.setRemoteDirectory("ftpSource/");
        messageSource.setFilter(new AcceptAllFileListFilter<>());
        messageSource.setMaxFetchSize(1);
        return messageSource;
    }

    @Bean
    @Transformer(inputChannel = "stream", outputChannel = "data")
    public org.springframework.integration.transformer.Transformer transformer() {
        return new StreamTransformer("UTF-8");
    }

    @Bean
    public FtpRemoteFileTemplate template() {
        return new FtpRemoteFileTemplate(ftpSessionFactory());
    }

    @ServiceActivator(inputChannel = "data", adviceChain = "after")
    @Bean
    public MessageHandler handle() {
        return System.out::println;
    }

    @Bean
    public ExpressionEvaluatingRequestHandlerAdvice after() {
        ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
        advice.setOnSuccessExpression(
                "@template.remove(headers['file_remoteDirectory'] + headers['file_remoteFile'])");
        advice.setPropagateEvaluationFailures(true);
        return advice;
    }

}

请注意，在此示例中，转换器下游的消息处理程序有一个“建议”，可在处理后移除远程文件。