Spring Integration DSL для SFTP Outbound с удаления

Question

Я использую

• Sprint интеграции (File, SFTP и т.д.) 4.3.6
• Spring Загрузочный 1.4.3
• Spring Integration Java DSL 1.1. 4

и я пытаюсь настроить исходящий адаптер SFTP, который позволит мне переместить файл в каталог на удаленной системе, а также удалить или переименовать файл в моей локальной системе.

Так, например, я хотел бы поместить файл, a.txt, в локальном каталоге и он SFTP'ed на удаленный сервер в каталоге въездной. Как только передача будет завершена, я хочу, чтобы локальная копия a.txt была удалена или переименована.

Я играл с несколькими способами. Итак, вот мой общий SessionFactory для теста.

protected SessionFactory<ChannelSftp.LsEntry> buildSftpSessionFactory() { 
    DefaultSftpSessionFactory sessionFactory = new DefaultSftpSessionFactory(); 
    sessionFactory.setHost("localhost"); 
    sessionFactory.setUser("user"); 
    sessionFactory.setAllowUnknownKeys(true); 
    sessionFactory.setPassword("pass"); 
    CachingSessionFactory<ChannelSftp.LsEntry> cachingSessionFactory = new CachingSessionFactory<>(sessionFactory, 1); 
    return cachingSessionFactory; 
} 

Это трансформатор, который я должен добавить некоторые из заголовков сообщения

@Override 
public Message<File> transform(Message<File> source) { 
    System.out.println("here is the thing : "+source); 
    File file = (File)source.getPayload(); 
    Message<File> transformedMessage = MessageBuilder.withPayload(file) 
      .copyHeaders(source.getHeaders()) 
      .setHeaderIfAbsent(FileHeaders.ORIGINAL_FILE, file) 
      .setHeaderIfAbsent(FileHeaders.FILENAME, file.getName()) 
      .build(); 
    return transformedMessage; 
} 

я тогда в поток интеграции, который использует Poller наблюдать локальный каталог и вызвать это:

@Bean 
public IntegrationFlow pushTheFile(){ 
    return IntegrationFlows 
      .from(s -> s.file(new File(DIR_TO_WATCH)) 
          .patternFilter("*.txt").preventDuplicates(), 
        e -> e.poller(Pollers.fixedDelay(100))) 
      .transform(outboundTransformer) 
      .handle(Sftp.outboundAdapter(this.buildSftpSessionFactory()) 
        .remoteFileSeparator("/") 
        .useTemporaryFileName(false) 
        .remoteDirectory("inbound/") 
      ) 
      .get(); 
} 

Это прекрасно работает, но выходит из локального файла. Любые идеи о том, как удалить этот локальный файл после завершения загрузки? Должен ли я смотреть на SftpOutboundGateway вместо этого?

Заранее благодарен!

Ответ Артема работал отлично! Вот краткий пример, который удаляет локальный файл после его нажатия.

@Bean 
public IntegrationFlow pushTheFile(){ 
    return IntegrationFlows 
      .from(s -> s.file(new File(DIR_TO_WATCH)) 
          .patternFilter("*.txt").preventDuplicates(), 
        e -> e.poller(Pollers.fixedDelay(100))) 
      .transform(outboundTransformer) 
      .handle(Sftp.outboundAdapter(this.buildSftpSessionFactory()) 
        .remoteFileSeparator("/") 
        .useTemporaryFileName(false) 
        .remoteDirectory("inbound/"), c -> c.advice(expressionAdvice(c)) 
      ) 
      .get(); 
} 

@Bean 
public Advice expressionAdvice(GenericEndpointSpec<FileTransferringMessageHandler<ChannelSftp.LsEntry>> c) { 
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice(); 
    advice.setOnSuccessExpression("payload.delete()"); 
    advice.setOnFailureExpression("payload + ' failed to upload'"); 
    advice.setTrapException(true); 
    return advice; 
} 

Answer 1

Для этой цели вы можете использовать несколько подходов.

Все они основаны на том факте, что вы делаете что-то еще с исходным сообщением запроса для Sftp.outboundAdapter().

1. .publishSubscribeChannel() позволяет отправить одно сообщение нескольким абонентам, когда второй получит это только первый один заканчивает свою работу. По умолчанию, если вы не указали Executor

2. routeToRecipients() позволяет сделать тот же результат, но с помощью другого компонента.

3. ExpressionEvaluatingRequestHandlerAdvice - добавить этот к .advice() определения Sftp.outboundAdapter() конечной точки - второй аргумент .handle() и выполнить file.delete() через onSuccessExpression:

   .transform((Integer p) -> p * 2, c -> c.advice(this.expressionAdvice()))