我正在从ftp server foo.csv 中撤出一个文件，创建新文件 bar.csv ftp。然后，我正在使用咨询发送到成功渠道，并在"成功的建议"中删除有效载荷，该建议是从本地删除 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;
}
只需在第二个FTP频道适配器中添加另一个建议。