Spring Integration DSL for SFTP Outbound with delete

Question

I'm using

• Sprint Integration(File, SFTP, etc) 4.3.6
• Spring Boot 1.4.3
• Spring Integration Java DSL 1.1.4

and I'm trying to set up an SFTP outbound adapter that would allow me move a file to a directory on the remote system and also remove or rename the file in my local system.

So, for instance I would like to place a file, a.txt, in a local directory and have it SFTP'ed to a remote server in the directory inbound. Once the transfer is complete I would like to have the local copy of a.txt deleted, or renamed.

I was toying with a couple of ways to this. So here is my common SessionFactory for the test.

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

This is a transformer that I have to add some of the headers to the message

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

I then have an integration flow that uses a poller to watch a local directory and invoke this :

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

This works fine, but leaves the local file. Any ideas on how to have this remove local file after the upload completes? Should I be looking at the SftpOutboundGateway instead?

Thanks in advance!

Artem's answer worked perfectly! Here is a quick example that deletes the local file after it is pushed.

@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

For this purpose you can use several approaches.

All of them are based on the fact that you do something else with the original request message for the Sftp.outboundAdapter().

1. .publishSubscribeChannel() allows you to send the same message to several subscribers, when the second will receive it only the first one finishes its work. By default, if you don't specify Executor

2. routeToRecipients() allows you make the same result but via different component.

3. ExpressionEvaluatingRequestHandlerAdvice - you add this one to the .advice() of the Sftp.outboundAdapter() endpoint definition - the second .handle() argument and perform file.delete() via onSuccessExpression:

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