Search code examples
spring-integration

The success channel sending .msg file instead of the .csv file

I'm sending a new payload header to a success channel using an advice and in the success channel I'm sending the file_originalFile to FTP server using outboundAdapter but I'm facing an issue where the success channel is sending the file as .msg and in side the .msg I can see a text showing the name of the file_originalFile while I want to send foo.csv it self to the ftp server.

Any possible solution for that?

Here is the code of the application

public IntegrationFlow localToFtpFlow(Branch myBranch) {

    return IntegrationFlows.from(Files.inboundAdapter(new File(myBranch.getBranchCode()))
                    .filter(new ChainFileListFilter<File>()
                            .addFilter(new RegexPatternFileListFilter("final" + myBranch.getBranchCode() + ".csv"))
                            .addFilter(new FileSystemPersistentAcceptOnceFileListFilter(metadataStore(dataSource), "foo"))),//FileSystemPersistentAcceptOnceFileListFilter
            e -> e.poller(Pollers.fixedDelay(10_000)))
            .enrichHeaders(h ->h.header("file_originalFile", "new java.io.File('/BEY/FEFOexport" + myBranch.getBranchCode() + ".csv')",true))

            .transform(p -> {
                LOG1.info("Sending file " + p + " to FTP branch " + myBranch.getBranchCode());

                return p;
            })
            .log()
            .handle(Ftp.outboundAdapter(createNewFtpSessionFactory(myBranch), FileExistsMode.REPLACE)
                    .useTemporaryFileName(true)
                    .autoCreateDirectory(false)
                    .remoteDirectory(myBranch.getFolderPath()), e -> e.advice(expressionAdvice()))

            .get();
}

/*
* Creating the advice for routing the payload of the outbound message on different expressions (success, failure)
*
* */

@Bean
public Advice expressionAdvice() {
    ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
    advice.setSuccessChannelName("success.input");
    advice.setOnSuccessExpressionString("payload.delete() + ' was successful'");
    //advice.setFailureChannelName("failure.input");
    advice.setOnFailureExpressionString("payload + ' was bad, with reason: ' + #exception.cause.message");
    advice.setTrapException(true);
    return advice;
}

/*
* Creating FTP connection based on the branch ftp data entered.
* */

public DefaultFtpSessionFactory createNewFtpSessionFactory(Branch branch) {
    final DefaultFtpSessionFactory factory = new DefaultFtpSessionFactory();
    factory.setHost(branch.getHost());
    factory.setUsername(branch.getUsern());
    factory.setPort(branch.getFtpPort());
    factory.setPassword(branch.getPassword());
    return factory;
}

public DefaultFtpSessionFactory createNewFtpSessionFactory() {
    final DefaultFtpSessionFactory factory = new DefaultFtpSessionFactory();
    factory.setHost("xxx");
    factory.setUsername("xxxxx");
    factory.setPort(xx);
    factory.setPassword("xxxx");
    return factory;
}

/*
* Creating a metadata store to be used across the application flows to prevent reprocessing the file if it is already processed.
* This will save the new file in a metadata table in the DB with the state of the report, so when a new copy comes with different date it will be processed only.
* */
@Bean
public ConcurrentMetadataStore metadataStore(final DataSource dataSource) {
    return new JdbcMetadataStore(dataSource);
}

/*
* Success channel that will handle the AdviceMessage from the outbound adapter
*
* */

@Bean
public IntegrationFlow success(){
    return f -> f.transform("inputMessage.headers['file_originalFile']")
                 .handle(Ftp.outboundAdapter(createNewFtpSessionFactory(), FileExistsMode.REPLACE)
                 .useTemporaryFileName(true)
                 .autoCreateDirectory(true)
                 .remoteDirectory("/ftp/erbranch/EDMS/FEFO/History/"));
           //f.handle(System.out::println);
}

Debug:

2019-02-08 20:18:34.264 DEBUG 1368 --- [ask-scheduler-2] o.s.integration.channel.DirectChannel    : postSend (sent=true) on channel '1o.channel#3', message: GenericMessage [payload=BEY\finalBEY.csv, headers={file_originalFile=new java.io.File('/BEY/FEFOexportBEY.csv'), id=e1b48bf4-ba44-bdd0-23bc-e335e3c377a1, file_name=finalBEY.csv, file_relativePath=finalBEY.csv, timestamp=1549649899453}]
2019-02-08 20:18:34.267 DEBUG 1368 --- [ask-scheduler-2] o.s.integration.channel.DirectChannel    : postSend (sent=true) on channel '1o.channel#1', message: GenericMessage [payload=BEY\finalBEY.csv, headers={file_originalFile=new java.io.File('/BEY/FEFOexportBEY.csv'), id=0c599e08-a972-2ffc-cc9b-f45ff266ca98, file_name=finalBEY.csv, file_relativePath=finalBEY.csv, timestamp=1549649899439}]
2019-02-08 20:18:34.268 DEBUG 1368 --- [ask-scheduler-2] o.s.integration.channel.DirectChannel    : postSend (sent=true) on channel '1o.channel#0', message: GenericMessage [payload=BEY\finalBEY.csv, headers={file_originalFile=BEY\finalBEY.csv, id=14336975-2e33-1990-63e0-3182726b47fe, file_name=finalBEY.csv, file_relativePath=finalBEY.csv, timestamp=1549649899439}]
Solution

  • .enrichHeaders(h ->h.header("file_originalFile","/someDirectory/FEFOexport" + myBranch.getBranchCode() + ".csv",true))

    The header contains just the file name, not a File. When you send a String, it becomes the content of the remote file. In that case, the remote file name uses the FileHeaders.FILENAME by default, or <messageId>.msg if there is no such header.

    Use 

    .enrichHeaders(h ->h.headerExpression("file_originalFile", "new java.io.File('/someDirectory/FEFOexport" + myBranch.getBranchCode() + ".csv')",true))