Spring Integration FTP 使用后删除本地文件（Spring Boot）

Question

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

@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

从 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);


        }

Answer 2

我建议您将 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，以便根据有效负载删除本地文件。