监视远程目录以获取添加的文件并将其流式传输以通过 SFTP 读取数据

Question

我想在远程机器上添加一个监视新添加的 CSV 文件或未读文件。识别文件后，根据文件名中的时间戳读取文件。该文件将使用流而不是处理本地计算机来读取。在读取文件时，将 _reading 附加到文件名并在读取文件后附加 _read。该文件将通过 SFTP 协议读取，我计划使用 spring 集成 sftp。如果在读取文件时出错或文件中的数据不符合预期，我想将该文件移动到子目录中。

我尝试轮询远程目录并读取一次 CSV 文件。一旦读取，我将从目录中删除文件。

 <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-sftp</artifactId>
            <version>5.1.0.RELEASE</version>
        </dependency>
        <dependency>
            <groupId>org.springframework.integration</groupId>
            <artifactId>spring-integration-core</artifactId>
            <version>5.0.6.RELEASE</version>
        </dependency>

Spring boot version 2.0.3.RELEASE   

 @Bean
    public SessionFactory<ChannelSftp.LsEntry> sftpSessionFactory() {
        DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
        factory.setHost(hostname);
        factory.setPort(22);
        factory.setUser(username);
        factory.setPassword(password);
        factory.setAllowUnknownKeys(true);
        return new CachingSessionFactory<ChannelSftp.LsEntry>(factory);
    }

    @Bean
    public MessageSource<InputStream> sftpMessageSource() {
        SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template());
        messageSource.setRemoteDirectory(path);
        messageSource.setFilter(compositeFilters());
        return messageSource;
    }

    public CompositeFileListFilter compositeFilters() {
        return new CompositeFileListFilter()
                .addFilter(new SftpRegexPatternFileListFilter(".*csv"));
    }

    @Bean
    public SftpRemoteFileTemplate template() {
        return new SftpRemoteFileTemplate(sftpSessionFactory());
    }

    @Bean
    public IntegrationFlow sftpOutboundListFlow() {
        return IntegrationFlows.from(this.sftpMessageSource(), e -> e.poller(Pollers.fixedDelay(5, TimeUnit.SECONDS)))
                .handle(Sftp.outboundGateway(template(), NLST, path).options(Option.RECURSIVE)))
                .filter(compositeFilters())
                .transform(sorter())
                .split()
                .handle(Sftp.outboundGateway(template(), GET, "headers['file_remoteDirectory'] + headers['file_remoteFile']").options(STREAM))
                .transform(csvToPojoTransformer())
                .handle(service())
                .handle(Sftp.outboundGateway(template(), MV, "headers['file_remoteDirectory'] + headers['file_remoteFile'] + _read"))
                .handle(after())
                .get();
    }

    @Bean
    public MessageHandler sorter() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                List<String> fileNames = (List<String>) message.getPayload();
                Collections.sort(fileNames);
            }
        };
    }

    @Bean
    public MessageHandler csvToPojoTransformer() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                InputStream streamData = (InputStream) message.getPayload();
                convertStreamtoObject(streamData, Class.class);
            }
        };
    }

    public List<?> convertStreamtoObject(InputStream inputStream, Class clazz) {
        HeaderColumnNameMappingStrategy ms = new HeaderColumnNameMappingStrategy();
        ms.setType(clazz);
        Reader reader = new InputStreamReader(inputStream);

        CsvToBean cb = new CsvToBeanBuilder(reader)
                .withType(clazz)
                .withMappingStrategy(ms)
                .withSkipLines(0)
                .withSeparator('|')
                .withThrowExceptions(true)
                .build();
        return cb.parse();
    }

    @Bean
    public MessageHandler service() {
        return new MessageHandler() {
            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                List<Class> csvDataAsListOfPojo = List < Class > message.getPayload();
                // use this
            }
        };
    }
    @Bean
    public ExpressionEvaluatingRequestHandlerAdvice after() {
        ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
        advice.setSuccessChannelName("success.input");
        advice.setOnSuccessExpressionString("payload + ' was successful'");
        advice.setFailureChannelName("failure.input");
        advice.setOnFailureExpressionString("payload + ' was bad, with reason: ' + #exception.cause.message");
        advice.setTrapException(true);
        return advice;
    }
    @Bean
    public IntegrationFlow success() {
        return f -> f.handle(System.out::println);
    }

    @Bean
    public IntegrationFlow failure() {
        return f -> f.handle(System.out::println);
    }

更新代码

Answer 1

对于复杂的场景（列表、移动、获取、删除等），您应该使用SFTP remote file gateways instead。

SFTP 出站网关提供一组有限的命令，可让您与远程 SFTP 服务器交互：

ls（列出文件）

nlst（列出文件名）

获取（检索文件）

mget（检索多个文件）

rm（删除文件）

mv（移动和重命名文件）

put（发送文件）

mput（发送多个文件）

或者直接从您的代码中使用SftpRemoteFileTemplate。

编辑

响应您的 cmets；你需要这样的东西

• 入站通道适配器（带轮询器） - 返回目录名称
• LS 网关
• 过滤（删除所有已提取的文件）
• 转换器（列表排序）
• 分离器
• GET 网关（流选项）
• 转换器（csv 到 POJO）
• 服务（流程 POJO）

如果你添加

• RM 网关

在您的服务（删除远程文件）之后，您不需要过滤步骤。

您可能会发现 Java DSL 更易于组装此流程...

@Bean
public IntegrationFlow flow() {
    return IntegrationFlows.from(() -> "some/dir", e -> e.poller(...))
        .handle(...) // LS Gateway
        .filter(...)
        .transform(sorter())
        .split
        .handle(...) // GET Gateway
        .transform(csvToPojoTransformer())
        .handle(myService())
        .get()
}