Spring Integration FTP在使用后删除本地文件(Spring Boot)

Question

我正在尝试编写一个程序，该程序可以通过ftp从一台服务器获取文件，并通过ftp将其放置在另一台服务器上。但是，在写入本地文件后，我在删除该文件时遇到问题。只要是临时的，能够在本地保存就不是问题。我曾尝试将ExpressionEvaluatingRequestHandlerAdvice与OnSuccessExpression一起使用，但无法使其实际使用该表达式。代码在这里：

@Configuration
@EnableConfigurationProperties(FTPConnectionProperties.class)
public class FTPConfiguration {
private FTPConnectionProperties ftpConnectionProperties;
public FTPConfiguration(FTPConnectionProperties ftpConnectionProperties) {
this.ftpConnectionProperties = ftpConnectionProperties;
}
@Bean
public SessionFactory<FTPFile> ftpInputSessionFactory() {
DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
sf.setHost(ftpConnectionProperties.getInputServer());
sf.setUsername(ftpConnectionProperties.getInputFtpUser());
sf.setPassword(ftpConnectionProperties.getInputFtpPassword());
return new CachingSessionFactory<>(sf);
}
@Bean
public SessionFactory<FTPFile> ftpOutputSessionFactory() {
DefaultFtpSessionFactory sf = new DefaultFtpSessionFactory();
sf.setHost(ftpConnectionProperties.getOutputServer());
sf.setUsername(ftpConnectionProperties.getOutputFtpUser());
sf.setPassword(ftpConnectionProperties.getOutputFtpPassword());
return new CachingSessionFactory<>(sf);
}
@Bean
public FtpInboundFileSynchronizer ftpInboundFileSynchronizer() {
FtpInboundFileSynchronizer fileSynchronizer = new FtpInboundFileSynchronizer(ftpInputSessionFactory());
fileSynchronizer.setDeleteRemoteFiles(true);
fileSynchronizer.setRemoteDirectory(ftpConnectionProperties.getInputDirectory());
fileSynchronizer.setFilter(new FtpSimplePatternFileListFilter("*.TIF"));
return fileSynchronizer;
}
@Bean
@InboundChannelAdapter(channel = "input", poller = @Poller(fixedDelay = "5000"))
public MessageSource<File> ftpMessageSource() {
FtpInboundFileSynchronizingMessageSource source = new FtpInboundFileSynchronizingMessageSource(ftpInboundFileSynchronizer());
source.setLocalDirectory(new File("ftp-inbound"));
source.setAutoCreateLocalDirectory(true);
source.setLocalFilter(new FileSystemPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), ""));
return source;
}
@Bean
@ServiceActivator(inputChannel = "input")
public MessageHandler handler() {
FtpMessageHandler handler = new FtpMessageHandler(ftpOutputSessionFactory());
handler.setRemoteDirectoryExpression(new LiteralExpression(ftpConnectionProperties.getOutputDirectory()));
handler.setFileNameGenerator(message -> {
if (message.getPayload() instanceof File) {
return ((File) message.getPayload()).getName();
} else {
throw new IllegalArgumentException("File expected as payload.");
}
});
return handler;
}
}

它完全按照预期处理远程文件，从源中删除远程文件并将其放入输出中，但在使用后不会删除本地文件。

Answer 1

我建议您将input频道作为PublishSubscribeChannel，并添加一个简单的订户：

@Bean
public PublishSubscribeChannel input() {
return new PublishSubscribeChannel();
}

@Bean
@ServiceActivator(inputChannel = "input")
public MessageHandler handler() {
...
}

@Bean
@ServiceActivator(inputChannel = "input")
public MessageHandler deleteLocalFileService() {
return m ->  ((File) message.getPayload()).delete();
}

通过这种方式，带有File有效载荷的相同消息将首先发送到您的FtpMessageHandler，然后才发送到这个新的deleteLocalFileService，以便根据有效载荷删除本地文件。

Answer 2

从SFTP服务器获取文件，然后将该文件移动到其他具有不同名称的文件夹的简单解决方案

@Bean
public SessionFactory<ChannelSftp.LsEntry> sftpSessionFactory() {
DefaultSftpSessionFactory factory = new DefaultSftpSessionFactory(true);
if (sftpServerProperties.getSftpPrivateKey() != null) {
factory.setPrivateKey(sftpServerProperties.getSftpPrivateKey());
factory.setPrivateKeyPassphrase(sftpServerProperties.getSftpPrivateKeyPassphrase());
} else {
factory.setPassword(sftpServerProperties.getPassword());
}
factory.setHost(sftpServerProperties.getSftpHost());
factory.setPort(sftpServerProperties.getSftpPort());
factory.setUser(sftpServerProperties.getSftpUser());
factory.setAllowUnknownKeys(true);
return new CachingSessionFactory<>(factory);
}
@Bean
public SftpInboundFileSynchronizer sftpInboundFileSynchronizer() {
SftpInboundFileSynchronizer fileSynchronizer = new SftpInboundFileSynchronizer(sftpSessionFactory());
fileSynchronizer.setDeleteRemoteFiles(false);
fileSynchronizer.setRemoteDirectory(sftpServerProperties.getSftpRemoteDirectoryDownload());
fileSynchronizer.setFilter(new SftpSimplePatternFileListFilter(sftpServerProperties.getSftpRemoteDirectoryDownloadFilter()));
return fileSynchronizer;
}
@Bean
@InboundChannelAdapter(channel = "fromSftpChannel", poller = @Poller(cron = "*/10 * * * * *"))
public MessageSource<File> sftpMessageSource() {
SftpInboundFileSynchronizingMessageSource source = new SftpInboundFileSynchronizingMessageSource(
sftpInboundFileSynchronizer());
source.setLocalDirectory(util.createDirectory(Constants.FILES_DIRECTORY));
source.setAutoCreateLocalDirectory(true);
return source;
}
@Bean
@ServiceActivator(inputChannel = "fromSftpChannel")
public MessageHandler resultFileHandler() {
return (Message<?> message) -> {
String csvFilePath = util.getDirectory(Constants.FILES_DIRECTORY) + Constants.INSIDE + message.getHeaders().get("file_name");
util.readCSVFile(csvFilePath, String.valueOf(message.getHeaders().get("file_name")));
File file = (File) message.getPayload();
File newFile = new File(file.getPath() + System.currentTimeMillis());
try {
FileUtils.copyFile(file, newFile);
sftpGateway.sendToSftp(newFile);
} catch (Exception e) {
e.printStackTrace();
}
if (file.exists()) {
file.delete();
}
if (newFile.exists()) {
newFile.delete();
}
};
}
@Bean
@ServiceActivator(inputChannel = "toSftpChannelDest")
public MessageHandler handlerOrderBackUp() {
SftpMessageHandler handler = new SftpMessageHandler(sftpSessionFactory());
handler.setAutoCreateDirectory(true);
handler.setRemoteDirectoryExpression(new LiteralExpression(sftpServerProperties.getSftpRemoteBackupDirectory()));
return handler;
}

@MessagingGateway
public interface SFTPGateway {
@Gateway(requestChannel = "toSftpChannelDest")
void sendToSftp(File file);

}