Я извлекаю файл с FTP-сервера Foo.csv , создаю новый файл bar.csv и отправляю новый файл обратно на FTP. Затем я использую advice для отправки на канал успеха, и удаляю полезную нагрузку в сообщении об успехе, который удаляет bar.csv из локального, затем я перехожу на канал успеха выражение заголовка с исходным файлом ( foo.csv ), который я сначала извлек с FTP-сервера и отправляю обратно в папку истории FTP для архивации.
Теперь я хочу удалить его из моего локального хранилища и после отправки обратно в историю FTP, но не могу найти, как мне это сделать в успешном процессе. Есть ли какое-либо решение, которое я могу реализовать?
Это то, что я имею до сих пор:
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.headerExpression("file_originalFile", "new java.io.File('"+ myBranch.getBranchCode() +"/FEFOexport" + myBranch.getBranchCode() + ".csv')",true))
.transform(p -> {
LOG1.info("Sending file " + p + " to FTP branch " + myBranch.getBranchCode());
return p;
})
.log()
.transform(m -> {
this.defaultSessionFactoryLocator.addSessionFactory(myBranch.getBranchCode(),createNewFtpSessionFactory(myBranch));
LOG1.info("Adding factory to delegation");
return m;
})
.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)
* @return Advice
*/
@Bean
public Advice expressionAdvice() {
ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
advice.setSuccessChannelName("success.input");
advice.setOnSuccessExpressionString("payload.delete() + ' was successful'");
//advice.setOnSuccessExpressionString("inputMessage.headers['file_originalFile'].renameTo(new java.io.File(payload.absolutePath + '.success.to.send'))");
//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.
* @return ftpSessionFactory
*/
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;
}
/**
* Creating a default FTP connection.
* @return ftpSessionFactory
*/
@Bean
public SessionFactory<FTPFile> createNewFtpSessionFactory() {
final DefaultFtpSessionFactory factory = new DefaultFtpSessionFactory();
factory.setHost("xxxxxx");
factory.setUsername("xxx");
factory.setPort(21);
factory.setPassword("xxxxx");
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.
* @return metadataStore
* */
@Bean
public ConcurrentMetadataStore metadataStore(final DataSource dataSource) {
return new JdbcMetadataStore(dataSource);
}
/*
* Success channel that will handle the AdviceMessage from the outbound adapter and sends the inputMessage file_originalFile to FTP destination folder specified.
*
* */
@Bean
public IntegrationFlow success(){
return f -> f.transform("inputMessage.headers['file_originalFile']")
.transform(e -> {
//getting the Branch code from the Input message and calling the correct factory based on it
delegatingSessionFactoryAuto().setThreadKey(e.toString().substring(0,3));
return e;
})
.handle(Ftp.outboundAdapter(delegatingSessionFactoryAuto(), FileExistsMode.REPLACE)
.useTemporaryFileName(true)
.autoCreateDirectory(true)
.fileNameGenerator(new FileNameGenerator() {
@Override
public String generateFileName(Message<?> message) {
return new SimpleDateFormat("yyyy-MM-dd-hh-mm-ss").format(new Date()) + ".csv";
}
})
.remoteDirectory("/ftp/erbranch/EDMS/FEFO/History/" + new SimpleDateFormat("yyyy-MM-dd").format(new Date())) // + dateFormat.format(date)
.get(),e -> e.advice(expressionAdviceForSuccess()));
}
@Bean
public Advice expressionAdviceForSuccess() {
ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
//advice.setSuccessChannelName("success.input");
advice.setOnSuccessExpressionString("payload.delete() + ' was successful'");
//advice.setOnSuccessExpressionString("inputMessage.headers['file_originalFile'].renameTo(new java.io.File(payload.absolutePath + '.success.to.send'))");
//advice.setFailureChannelName("failure.input");
advice.setOnFailureExpressionString("payload + ' was bad, with reason: ' + #exception.cause.message");
advice.setTrapException(true);
return advice;
}
@Bean
public DelegatingSessionFactory<FTPFile> delegatingSessionFactoryAuto(){
SessionFactoryLocator<FTPFile> sff = createNewFtpSessionFactoryAndAddItToTheLocator();
return new DelegatingSessionFactory<FTPFile>(sff);
}
@Bean
public SessionFactoryLocator<FTPFile> createNewFtpSessionFactoryAndAddItToTheLocator(){
this.defaultSessionFactoryLocator.addSessionFactory("BEY",createNewFtpSessionFactory());
return this.defaultSessionFactoryLocator;
}