Получить строку из 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, чтобы получить сообщение для каждой строки.

EDIT (фильтр конфигурации)

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