Блог

У меня есть следующий код, который работает нормально, извлекая файлы с FTP-сервера в поток, но мне нужно получить строку каждого файла, кажется, мне нужно использовать преобразователь, передающий кодировку, но чего мне не хватает? Как именно получить строку содержимого каждого переданного файла?

Заранее большое спасибо

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

}
  

Используйте a StreamTransformer , чтобы получить весь файл в виде одной строки, или a 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;
}
  

РЕДАКТИРОВАТЬ 2 — удалить удаленный файл с рекомендациями

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