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

Question

我想在远程机器上添加一个手表，用于新添加的CSV文件或未读文件。一旦文件被识别，根据文件名中的时间戳读取。文件将使用流媒体而不是本地机器读取。读取文件时，将读取附加到文件名，并在读取文件后附加读取。该文件将通过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远程文件网关。

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

ls（列表文件）

nlst（列出文件名）

获取（检索文件）

mget（检索多个文件）

rm（删除文件）

mv（移动和重命名文件）

put（发送文件）

mput（发送多个文件）

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

编辑

回应您的评论；你需要这样的东西

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

如果你加上

• RM网关

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

您可能会发现Java DSL更容易组装此流。。。

java prettyprint-override">@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()
}