SFTP 流入站频道适配器

版本 4.3 引入了流入站通道适配器。 此适配器生成 payloads 为 类型的 message ，允许您获取文件而无需写入本地文件系统。 由于会话保持打开状态，因此使用应用程序负责在使用文件时关闭会话。 会话在标头 （） 中提供。 标准框架组件（如 和 ）会自动关闭会话。 有关这些组件的更多信息，请参见File Splitter 和 Stream Transformer。 以下示例显示如何配置 SFTP 流入站通道适配器：InputStreamcloseableResourceIntegrationMessageHeaderAccessor.CLOSEABLE_RESOURCEFileSplitterStreamTransformerspring-doc.cn

<int-sftp: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-sftp:inbound-streaming-channel-adapter>

您只能使用 、 、 或 中的一个。filename-patternfilename-regexfilterfilter-expressionspring-doc.cn

从版本 5.0 开始，默认情况下，适配器通过使用 based on the in-memory 来防止远程文件重复。 默认情况下，此过滤器也与文件名模式（或 regex）一起应用。 如果需要允许重复项，可以使用 . 您可以使用 （或 ） 处理任何其他使用案例。 后面显示的 Java 配置显示了一种在处理后删除远程文件以避免重复的技术。SftpStreamingMessageSourceSftpPersistentAcceptOnceFileListFilterSimpleMetadataStoreAcceptAllFileListFilterCompositeFileListFilterChainFileListFilter

有关 的更多信息，请参见远程持久性文件列表过滤器。SftpPersistentAcceptOnceFileListFilterspring-doc.cn

当需要提取时，你可以使用该属性来限制每次轮询时提取的文件数。 将其设置为在集群环境中运行时使用持久过滤器。 有关更多信息，请参见入站通道适配器：控制远程文件获取。max-fetch-size1spring-doc.cn

适配器将远程目录和文件名放在 headers （ 和 ， ） 中。 从版本 5.0 开始，标头提供额外的远程文件信息（以 JSON 格式）。 如果在 to 上设置属性，则标头包含一个对象。 您可以使用该方法访问底层提供的对象。 当您使用 XML 配置时，该属性不可用，但您可以通过将 the 注入到您的一个配置类中来设置它。 另请参阅远程文件信息。FileHeaders.REMOTE_DIRECTORYFileHeaders.REMOTE_FILEFileHeaders.REMOTE_FILE_INFOfileInfoJsonSftpStreamingMessageSourcefalseSftpFileInfoSftpClient.DirEntrySftpClientSftpFileInfo.getFileInfo()fileInfoJsonSftpStreamingMessageSourcespring-doc.cn

使用 Java 配置进行配置

以下 Spring Boot 应用程序显示了如何使用 Java 配置入站适配器的示例：spring-doc.cn

@SpringBootApplication
public class SftpJavaApplication {

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

    @Bean
    @InboundChannelAdapter(channel = "stream")
    public MessageSource<InputStream> ftpMessageSource() {
        SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template());
        messageSource.setRemoteDirectory("sftpSource/");
        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 SftpRemoteFileTemplate template() {
        return new SftpRemoteFileTemplate(sftpSessionFactory());
    }

    @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;
    }

}

请注意，在此示例中，转换器下游的消息处理程序具有一个，该处理程序在处理后删除远程文件。advicespring-doc.cn