从 Spring FTP 流式入站通道适配器获取字符串

Question

我有以下代码可以正常将文件从 FTP 服务器检索到流中，但我需要获取每个文件的字符串，似乎我需要使用 Transformer 传递一个字符集，但我缺少什么？究竟如何获取传输的每个文件的内容字符串？

提前非常感谢

@SpringBootApplication
@EnableIntegration
public class FtpinboundApp extends SpringBootServletInitializer implements WebApplicationInitializer {

    final static Logger logger = Logger.getLogger(FtpinboundApp.class);

    public static void main(String[] args) {
        SpringApplication.run(FtpinboundApp.class, args);
    }

    @Bean
    public SessionFactory<FTPFile> ftpSessionFactory() {
        DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
        sf.setHost("X.X.X.X");
        sf.setPort(21);
        sf.setUsername("xxx");
        sf.setPassword("XXX");
        return new CachingSessionFactory<FTPFile>(sf);
    }


    @Bean
    @ServiceActivator(inputChannel = "stream")
    public MessageHandler handler() {
        return new MessageHandler() {

            @Override
            public void handleMessage(Message<?> message) throws MessagingException {
                System.out.println("trasnferred file:" + message.getPayload());

            }

        };
    }

    @Bean
    @InboundChannelAdapter(value = "stream", poller = @Poller(fixedRate = "1000"))
    public MessageSource<InputStream> ftpMessageSource() {

        FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(), null);
        messageSource.setRemoteDirectory("/X/X/X");
        messageSource.setFilter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "streaming"));
        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());
       }


    @Bean(name = PollerMetadata.DEFAULT_POLLER)
    public PollerMetadata defaultPoller() {
        PollerMetadata pollerMetadata = new PollerMetadata();
        pollerMetadata.setTrigger(new PeriodicTrigger(5000));
        return pollerMetadata;
    }

}

Answer 1

使用StreamTransformer 将整个文件作为单个字符串获取，或使用FileSplitter 获取每一行的消息。

编辑（过滤器配置）

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

public FileListFilter<FTPFile> filter() {
    CompositeFileListFilter<FTPFile> filter = new CompositeFileListFilter<>();
    filter.addFilter(new FtpSimplePatternFileListFilter("*.txt"));
    filter.addFilter(acceptOnceFilter());
    return filter;
}

@Bean
public FtpPersistentAcceptOnceFileListFilter acceptOnceFilter() {
    FtpPersistentAcceptOnceFileListFilter filter = new FtpPersistentAcceptOnceFileListFilter(meta(),
            "streaming"); // keys will be, e.g. "streamingfoo.txt"
    filter.setFlushOnUpdate(true);
    return filter;
}

@Bean
public ConcurrentMetadataStore meta() {
    PropertiesPersistingMetadataStore meta = new PropertiesPersistingMetadataStore();
    meta.setBaseDirectory("/tmp/foo");
    meta.setFileName("ftpStream.properties");
    return meta;
}

EDIT2 - 删除带有建议的远程文件

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