Move file after successful ftp transfer using Java

问题:

I've defined a flow using spring-integration java dsl to ftp transfer a file, handle it, then transfer it back in an "archive" dir, and at last move it in a local archive dir. Which is something "quite easy" as:

@Bean
public IntegrationFlow ftpInboundFlow() {
    if (ftpProperties.getEnabled() == false) {
        return null;
    }
    logger.trace("Starting ftp flow");
    return IntegrationFlows
            .from(source -> source.ftp(ftpSessionFactory()).deleteRemoteFiles(true).preserveTimestamp(true)
                    .filter(compositeWithAcceptOnceFilter())                        .remoteDirectory(ftpProperties.getRemoteDirectory())
                    .localDirectory(new File(ftpProperties.getLocalDirectory())).autoCreateLocalDirectory(true),
            consumer -> consumer.id("ftpInboundAdapter")) /* Fine from() */
            .handle(new GenericHandler<File>() {

                @Override
                @Transactional
                public Object handle(File payload, Map<String, Object> headers) {
                    logger.debug("Data arrived {} {}", payload, payload.getClass().getName());
                    return payload;
                }

            }) /* Fine GenericHandler */
            .handleWithAdapter(a -> a.ftp(ftpSessionFactory())
                    .remoteDirectory(ftpProperties.getRemoteArchiveDirectory()).autoCreateDirectory(true))              
            .get();
}

If I append

.handleWithAdapter(a -> a.file("'" + ftpProperties.getLocalArchiveDirectory() + "'")
.autoCreateDirectory(true).deleteSourceFiles(true))

after the ftp adapter configuration the bean initializer replies with the following error message:

Caused by: org.springframework.beans.factory.BeanCreationException: The 'currentComponent' (org.springframework.integration.file.remote.handler.FileTransferringMessageHandler@80bfa9d) is a one-way 'MessageHandler' and it isn't appropriate to configure 'outputChannel'. This is the end of the integration flow.

How should I fix it?

回答1:

As a workaround I refactored my code in giving up with the file adapter:

@Bean
public IntegrationFlow ftpInboundFlow() {
    return IntegrationFlows
        .from(source -> source.ftp(ftpSessionFactory()).deleteRemoteFiles(true).preserveTimestamp(true)
                .filter(compositeWithAcceptOnceFilter())                        .remoteDirectory(ftpProperties.getRemoteDirectory())
                .localDirectory(new File(ftpProperties.getLocalDirectory())).autoCreateLocalDirectory(true),
        consumer -> consumer.id("ftpInboundAdapter")) /* Fine from() */
        .handle(new GenericHandler<File>() {

            @Override
            @Transactional
            public Object handle(File payload, Map<String, Object> headers) {
                logger.debug("Data arrived {} {}", payload, payload.getClass().getName());
                /* handle file content here ... */
                /* ... then move it to archive */
                File dstFile = new File("archive", payload);
                FileUtils.moveFile(payload, dstFile);
                return dstFile;
            }

        }) /* Fine GenericHandler */
        /* Archive the remote file */
        .handleWithAdapter(a -> a.ftp(ftpSessionFactory())
                .remoteDirectory(ftpProperties.getRemoteArchiveDirectory()).autoCreateDirectory(true))              
        .get();

}

回答2:

With the XML configuration we have a convention for one-way component like adapter, e.g. <int-ftp:outbound-channel-adapter> or <int-file:outbound-channel-adapter>. For the Java DSL we use the convention like Gateway suffix for the factory methods which produces request/reply endpoint. Otherwise it is one way, like yours a.ftp(ftpSessionFactory()).

From other side with the XML configuration we don't have any choice to go ahead with downstream flow definition because there is no output-channel on the <outbound-channel-adapter>.

With the Java DSL we don't have so much choice to prevent such a error like your. But I did there so comprehensive exception message which should follow you to the different way for your flow definition.

And the answer for you is .publishSubscribeChannel(), when you can subscribe several endpoints to it to accept the same message to make different logic for it.

Please, find my article where I explain most of SI Java DSL features line by line.