Spring integration - AbstractInboundFileSynchronizer not updating file

Question

I would have expected the ftp synchronization mechanism to update a changed file. However, from what I can see here, the file is only downloaded if it does not exist yet. As for now, the file is not saved locally even though the timestamp/ content has changed.

So here is what I've discovered so far:

The class org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer

@Override
    public void synchronizeToLocalDirectory(final File localDirectory) {
        final String remoteDirectory = this.remoteDirectoryExpression.getValue(this.evaluationContext, String.class);
        try {
            int transferred = this.remoteFileTemplate.execute(new SessionCallback<F, Integer>() {

                @Override
                public Integer doInSession(Session<F> session) throws IOException {
                    F[] files = session.list(remoteDirectory);
                    if (!ObjectUtils.isEmpty(files)) {
                        List<F> filteredFiles = filterFiles(files);
                        for (F file : filteredFiles) {
                            try {
                                if (file != null) {
                                    copyFileToLocalDirectory(
                                            remoteDirectory, file, localDirectory,
                                            session);
                                }
                            }
                            catch (RuntimeException e) {
                                if (AbstractInboundFileSynchronizer.this.filter instanceof ReversibleFileListFilter) {
                                    ((ReversibleFileListFilter<F>) AbstractInboundFileSynchronizer.this.filter)
                                            .rollback(file, filteredFiles);
                                }
                                throw e;
                            }
                            catch (IOException e) {
                                if (AbstractInboundFileSynchronizer.this.filter instanceof ReversibleFileListFilter) {
                                    ((ReversibleFileListFilter<F>) AbstractInboundFileSynchronizer.this.filter)
                                            .rollback(file, filteredFiles);
                                }
                                throw e;
                            }
                        }
                        return filteredFiles.size();
                    }
                    else {
                        return 0;
                    }
                }
            });
            if (this.logger.isDebugEnabled()) {
                this.logger.debug(transferred + " files transferred");
            }
        }
        catch (Exception e) {
            throw new MessagingException("Problem occurred while synchronizing remote to local directory", e);
        }
    }

filters the files to be downloaded. I'd like to use the org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter, which compares filenames and last modified dates.

It then invokes the copyFileToLocalDirectory function with the filtered files (to be copied).

protected void copyFileToLocalDirectory(String remoteDirectoryPath, F remoteFile, File localDirectory,
            Session<F> session) throws IOException {
        String remoteFileName = this.getFilename(remoteFile);
        String localFileName = this.generateLocalFileName(remoteFileName);
        String remoteFilePath = remoteDirectoryPath != null
                ? (remoteDirectoryPath + this.remoteFileSeparator + remoteFileName)
                : remoteFileName;
        if (!this.isFile(remoteFile)) {
            if (this.logger.isDebugEnabled()) {
                this.logger.debug("cannot copy, not a file: " + remoteFilePath);
            }
            return;
        }

        File localFile = new File(localDirectory, localFileName);
        if (!localFile.exists()) {
            String tempFileName = localFile.getAbsolutePath() + this.temporaryFileSuffix;
            File tempFile = new File(tempFileName);
            OutputStream outputStream = new BufferedOutputStream(new FileOutputStream(tempFile));
            try {
                session.read(remoteFilePath, outputStream);
            }
            catch (Exception e) {
                if (e instanceof RuntimeException) {
                    throw (RuntimeException) e;
                }
                else {
                    throw new MessagingException("Failure occurred while copying from remote to local directory", e);
                }
            }
            finally {
                try {
                    outputStream.close();
                }
                catch (Exception ignored2) {
                }
            }

            if (tempFile.renameTo(localFile)) {
                if (this.deleteRemoteFiles) {
                    session.remove(remoteFilePath);
                    if (this.logger.isDebugEnabled()) {
                        this.logger.debug("deleted " + remoteFilePath);
                    }
                }
            }
            if (this.preserveTimestamp) {
                localFile.setLastModified(getModified(remoteFile));
            }
        }
    }

However, this method checks (only based on the file name) if the file is already present on the local disk and only downloads if it is not present. So basically there is no chance the updated file with (with the new timestamp) gets downloaded.

I played around trying to change the FtpInboundFileSynchronizer but it gets way too complicated. What is the best way to "customize" the synchronize- / copyToLocalDirectory methods?

Answer 1

It is possible to update AbstractInboundFileSynchronizer to recognize updated files, but it is brittle and you run into other issues.

Update 13/Nov/2016: Found out how to get modification timestamps in seconds.

The main problem with updating the AbstractInboundFileSynchronizer is that it has setter-methods but no (protected) getter-methods. If, in the future, the setter-methods do something smart, the updated version presented here will break.

The main issue with updating files in the local directory is concurrency: if you are processing a local file at the same time that an update is being received, you can run into all sorts of trouble. The easy way out is to move the local file to a (temporary) processing directory so that an update can be received as a new file which in turn removes the need to update AbstractInboundFileSynchronizer. See also Camel's timestamp remarks.

By default FTP servers provide modification timestamps in minutes. For testing I updated the FTP client to use the MLSD command which provides modification timestamps in seconds (and milliseconds if you are lucky), but not all FTP servers support this.

As mentioned on the Spring FTP reference the local file filter needs to be a FileSystemPersistentAcceptOnceFileListFilter to ensure local files are picked up when the modification timestamp changes.

Below my version of the updated AbstractInboundFileSynchronizer, followed by some test classes I used.

public class FtpUpdatingFileSynchronizer extends FtpInboundFileSynchronizer {

    protected final Log logger = LogFactory.getLog(this.getClass());

    private volatile Expression localFilenameGeneratorExpression;
    private volatile EvaluationContext evaluationContext;
    private volatile boolean deleteRemoteFiles;
    private volatile String remoteFileSeparator = "/";
    private volatile boolean  preserveTimestamp;

    public FtpUpdatingFileSynchronizer(SessionFactory<FTPFile> sessionFactory) {
        super(sessionFactory);
        setPreserveTimestamp(true);
    }

    @Override
    public void setLocalFilenameGeneratorExpression(Expression localFilenameGeneratorExpression) {
        super.setLocalFilenameGeneratorExpression(localFilenameGeneratorExpression);
        this.localFilenameGeneratorExpression = localFilenameGeneratorExpression;
    }

    @Override
    public void setIntegrationEvaluationContext(EvaluationContext evaluationContext) {
        super.setIntegrationEvaluationContext(evaluationContext);
        this.evaluationContext = evaluationContext;
    }

    @Override
    public void setDeleteRemoteFiles(boolean deleteRemoteFiles) {
        super.setDeleteRemoteFiles(deleteRemoteFiles);
        this.deleteRemoteFiles = deleteRemoteFiles;
    }

    @Override
    public void setRemoteFileSeparator(String remoteFileSeparator) {
        super.setRemoteFileSeparator(remoteFileSeparator);
        this.remoteFileSeparator = remoteFileSeparator;
    }

    @Override
    public void setPreserveTimestamp(boolean preserveTimestamp) {
        // updated
        Assert.isTrue(preserveTimestamp, "for updating timestamps must be preserved");
        super.setPreserveTimestamp(preserveTimestamp);
        this.preserveTimestamp = preserveTimestamp;
    }

    @Override
    protected void copyFileToLocalDirectory(String remoteDirectoryPath, FTPFile remoteFile, File localDirectory,
            Session<FTPFile> session) throws IOException {

        String remoteFileName = this.getFilename(remoteFile);
        String localFileName = this.generateLocalFileName(remoteFileName);
        String remoteFilePath = (remoteDirectoryPath != null
                ? (remoteDirectoryPath + this.remoteFileSeparator + remoteFileName)
                        : remoteFileName);

        if (!this.isFile(remoteFile)) {
            if (this.logger.isDebugEnabled()) {
                this.logger.debug("cannot copy, not a file: " + remoteFilePath);
            }
            return;
        }

        // start update
        File localFile = new File(localDirectory, localFileName);
        boolean update = false;
        if (localFile.exists()) {
            if (this.getModified(remoteFile) > localFile.lastModified()) {
                this.logger.info("Updating local file " + localFile);
                update = true;
            } else {
                this.logger.info("File already exists: " + localFile);
                return;
            }
        }
        // end update

        String tempFileName = localFile.getAbsolutePath() + this.getTemporaryFileSuffix();
        File tempFile = new File(tempFileName);
        OutputStream outputStream = new BufferedOutputStream(new FileOutputStream(tempFile));
        try {
            session.read(remoteFilePath, outputStream);
        } catch (Exception e) {
            if (e instanceof RuntimeException) {
                throw (RuntimeException) e;
            }
            else {
                throw new MessagingException("Failure occurred while copying from remote to local directory", e);
            }
        } finally {
            try {
                outputStream.close();
            }
            catch (Exception ignored2) {
            }
        }
        // updated
        if (update && !localFile.delete()) {
            throw new MessagingException("Unable to delete local file [" + localFile + "] for update.");
        }
        if (tempFile.renameTo(localFile)) {
            if (this.deleteRemoteFiles) {
                session.remove(remoteFilePath);
                if (this.logger.isDebugEnabled()) {
                    this.logger.debug("deleted " + remoteFilePath);
                }
            }
            // updated
            this.logger.info("Stored file locally: " + localFile);
        } else {
            // updated
            throw new MessagingException("Unable to rename temporary file [" + tempFile + "] to [" + localFile + "]");
        }
        if (this.preserveTimestamp) {
            localFile.setLastModified(getModified(remoteFile));
        }
    }

    private String generateLocalFileName(String remoteFileName) {

        if (this.localFilenameGeneratorExpression != null) {
            return this.localFilenameGeneratorExpression.getValue(this.evaluationContext, remoteFileName, String.class);
        }
        return remoteFileName;
    }

}

Following some of the test classes I used. I used dependencies org.springframework.integration:spring-integration-ftp:4.3.5.RELEASE and org.apache.ftpserver:ftpserver-core:1.0.6 (plus the usual logging and testing dependencies).

public class TestFtpSync {

    static final Logger log = LoggerFactory.getLogger(TestFtpSync.class);
    static final String FTP_ROOT_DIR = "target" + File.separator + "ftproot";
    // org.apache.ftpserver:ftpserver-core:1.0.6
    static FtpServer server;

    @BeforeClass
    public static void startServer() throws FtpException {

        File ftpRoot = new File (FTP_ROOT_DIR);
        ftpRoot.mkdirs();
        TestUserManager userManager = new TestUserManager(ftpRoot.getAbsolutePath());
        FtpServerFactory serverFactory = new FtpServerFactory();
        serverFactory.setUserManager(userManager);
        ListenerFactory factory = new ListenerFactory();
        factory.setPort(4444);
        serverFactory.addListener("default", factory.createListener());
        server = serverFactory.createServer();
        server.start();
    }

    @AfterClass
    public static void stopServer() {

        if (server != null) {
            server.stop();
        }
    }

    File ftpFile = Paths.get(FTP_ROOT_DIR, "test1.txt").toFile();
    File ftpFile2 = Paths.get(FTP_ROOT_DIR, "test2.txt").toFile();

    @Test
    public void syncDir() {

        // org.springframework.integration:spring-integration-ftp:4.3.5.RELEASE
        AnnotationConfigApplicationContext ctx = new AnnotationConfigApplicationContext();
        try {
            ctx.register(FtpSyncConf.class);
            ctx.refresh();
            PollableChannel msgChannel = ctx.getBean("inputChannel", PollableChannel.class);
            for (int j = 0; j < 2; j++) {
                for (int i = 0; i < 2; i++) {
                    storeFtpFile();
                }
                for (int i = 0; i < 4; i++) {
                    fetchMessage(msgChannel);
                }
            }
        } catch (Exception e) {
            throw new AssertionError("FTP test failed.", e);
        } finally {
            ctx.close();
            cleanup();
        }
    }

    boolean tswitch = true;

    void storeFtpFile() throws IOException, InterruptedException {

        File f = (tswitch ? ftpFile : ftpFile2);
        tswitch = !tswitch;
        log.info("Writing message " + f.getName());
        Files.write(f.toPath(), ("Hello " + System.currentTimeMillis()).getBytes());
    }

    Message<?> fetchMessage(PollableChannel msgChannel) {

        log.info("Fetching message.");
        Message<?> msg = msgChannel.receive(1000L);
        if (msg == null) {
            log.info("No message.");
        } else {
            log.info("Have a message: " + msg);
        }
        return msg;
    }

    void cleanup() {

        delFile(ftpFile);
        delFile(ftpFile2);
        File d = new File(FtpSyncConf.LOCAL_DIR);
        if (d.isDirectory()) {
            for (File f : d.listFiles()) {
                delFile(f);
            }
        }
        log.info("Finished cleanup");
    }

    void delFile(File f) {

        if (f.isFile()) {
            if (f.delete()) {
                log.info("Deleted " + f);
            } else {
                log.error("Cannot delete file " + f);
            }
        }
    }

}

public class MlistFtpSessionFactory extends AbstractFtpSessionFactory<MlistFtpClient> {

    @Override
    protected MlistFtpClient createClientInstance() {
        return new MlistFtpClient();
    }

}

public class MlistFtpClient extends FTPClient {

    @Override
    public FTPFile[] listFiles(String pathname) throws IOException {
        return super.mlistDir(pathname);
    }
}

@EnableIntegration
@Configuration
public class FtpSyncConf {

    private static final Logger log = LoggerFactory.getLogger(FtpSyncConf.class);

    public static final String LOCAL_DIR = "/tmp/received";

    @Bean(name = "ftpMetaData")
    public ConcurrentMetadataStore ftpMetaData() {
        return new SimpleMetadataStore();
    }

    @Bean(name = "localMetaData")
    public ConcurrentMetadataStore localMetaData() {
        return new SimpleMetadataStore();
    }

    @Bean(name = "ftpFileSyncer")
    public FtpUpdatingFileSynchronizer ftpFileSyncer(
            @Qualifier("ftpMetaData") ConcurrentMetadataStore metadataStore) {

        MlistFtpSessionFactory ftpSessionFactory = new MlistFtpSessionFactory();
        ftpSessionFactory.setHost("localhost");
        ftpSessionFactory.setPort(4444);
        ftpSessionFactory.setUsername("demo");
        ftpSessionFactory.setPassword("demo");

        FtpPersistentAcceptOnceFileListFilter fileFilter = new FtpPersistentAcceptOnceFileListFilter(metadataStore, "ftp");
        fileFilter.setFlushOnUpdate(true);
        FtpUpdatingFileSynchronizer ftpFileSync = new FtpUpdatingFileSynchronizer(ftpSessionFactory);
        ftpFileSync.setFilter(fileFilter);
        // ftpFileSync.setDeleteRemoteFiles(true);
        return ftpFileSync;
    }

    @Bean(name = "syncFtp")
    @InboundChannelAdapter(value = "inputChannel", poller = @Poller(fixedDelay = "500", maxMessagesPerPoll = "1"))
    public MessageSource<File> syncChannel(
            @Qualifier("localMetaData") ConcurrentMetadataStore metadataStore,
            @Qualifier("ftpFileSyncer") FtpUpdatingFileSynchronizer ftpFileSync) throws Exception {

        FtpInboundFileSynchronizingMessageSource messageSource = new FtpInboundFileSynchronizingMessageSource(ftpFileSync);
        File receiveDir = new File(LOCAL_DIR);
        receiveDir.mkdirs();
        messageSource.setLocalDirectory(receiveDir);
        messageSource.setLocalFilter(new FileSystemPersistentAcceptOnceFileListFilter(metadataStore, "local"));
        log.info("Message source bean created.");
        return messageSource;
    }

    @Bean(name = "inputChannel")
    public PollableChannel inputChannel() {

        QueueChannel channel = new QueueChannel();
        log.info("Message channel bean created.");
        return channel;
    }

}

/**
 * Copied from https://github.com/spring-projects/spring-integration-samples/tree/master/basic/ftp/src/test/java/org/springframework/integration/samples/ftp/support
 * @author Gunnar Hillert
 *
 */
public class TestUserManager extends AbstractUserManager {
    private BaseUser testUser;
    private BaseUser anonUser;

    private static final String TEST_USERNAME = "demo";
    private static final String TEST_PASSWORD = "demo";

    public TestUserManager(String homeDirectory) {
        super("admin", new ClearTextPasswordEncryptor());

        testUser = new BaseUser();
        testUser.setAuthorities(Arrays.asList(new Authority[] {new ConcurrentLoginPermission(1, 1), new WritePermission()}));
        testUser.setEnabled(true);
        testUser.setHomeDirectory(homeDirectory);
        testUser.setMaxIdleTime(10000);
        testUser.setName(TEST_USERNAME);
        testUser.setPassword(TEST_PASSWORD);

        anonUser = new BaseUser(testUser);
        anonUser.setName("anonymous");
    }

    public User getUserByName(String username) throws FtpException {
        if(TEST_USERNAME.equals(username)) {
            return testUser;
        } else if(anonUser.getName().equals(username)) {
            return anonUser;
        }

        return null;
    }

    public String[] getAllUserNames() throws FtpException {
        return new String[] {TEST_USERNAME, anonUser.getName()};
    }

    public void delete(String username) throws FtpException {
        throw new UnsupportedOperationException("Deleting of FTP Users is not supported.");
    }

    public void save(User user) throws FtpException {
        throw new UnsupportedOperationException("Saving of FTP Users is not supported.");
    }

    public boolean doesExist(String username) throws FtpException {
        return (TEST_USERNAME.equals(username) || anonUser.getName().equals(username)) ? true : false;
    }

    public User authenticate(Authentication authentication) throws AuthenticationFailedException {
        if(UsernamePasswordAuthentication.class.isAssignableFrom(authentication.getClass())) {
            UsernamePasswordAuthentication upAuth = (UsernamePasswordAuthentication) authentication;

            if(TEST_USERNAME.equals(upAuth.getUsername()) && TEST_PASSWORD.equals(upAuth.getPassword())) {
                return testUser;
            }

            if(anonUser.getName().equals(upAuth.getUsername())) {
                return anonUser;
            }
        } else if(AnonymousAuthentication.class.isAssignableFrom(authentication.getClass())) {
            return anonUser;
        }

        return null;
    }
}

Update 15/Nov/2016: Note on xml-configuration.

The xml-element inbound-channel-adapter is directly linked to the FtpInboundFileSynchronizer via org.springframework.integration.ftp.config.FtpInboundChannelAdapterParser via FtpNamespaceHandler via spring-integration-ftp-4.3.5.RELEASE.jar!/META-INF/spring.handlers.
Following the xml-custom reference guide, specifying a custom FtpNamespaceHandler in a local META-INF/spring.handlers file should allow you to use the FtpUpdatingFileSynchronizer instead of the FtpInboundFileSynchronizer. It did not work for me with unit-tests though and a proper solution would probably involve creating extra/modified xsd-files so that the regular inbound-channel-adapter is using the regular FtpInboundFileSynchronizer and a special inbound-updating-channel-adapter is using the FtpUpdatingFileSynchronizer. Doing this properly is a bit out of scope for this answer.
A quick hack can get you started though. You can overwrite the default FtpNamespaceHandler by creating package org.springframework.integration.ftp.config and class FtpNamespaceHandler in your local project. Contents shown below:

package org.springframework.integration.ftp.config;

public class FtpNamespaceHandler extends org.springframework.integration.config.xml.AbstractIntegrationNamespaceHandler {

    @Override
    public void init() {
        System.out.println("Initializing FTP updating file synchronizer.");
        // one updated line below, rest copied from original FtpNamespaceHandler
        registerBeanDefinitionParser("inbound-channel-adapter", new MyFtpInboundChannelAdapterParser());
        registerBeanDefinitionParser("inbound-streaming-channel-adapter",
                new FtpStreamingInboundChannelAdapterParser());
        registerBeanDefinitionParser("outbound-channel-adapter", new FtpOutboundChannelAdapterParser());
        registerBeanDefinitionParser("outbound-gateway", new FtpOutboundGatewayParser());
    }

}

package org.springframework.integration.ftp.config;

import org.springframework.integration.file.remote.synchronizer.InboundFileSynchronizer;
import org.springframework.integration.ftp.config.FtpInboundChannelAdapterParser;

public class MyFtpInboundChannelAdapterParser extends FtpInboundChannelAdapterParser {

    @Override
    protected Class<? extends InboundFileSynchronizer> getInboundFileSynchronizerClass() {
        System.out.println("Returning updating file synchronizer.");
        return FtpUpdatingFileSynchronizer.class;
    }

}

Also add preserve-timestamp="true" to the xml-file to prevent the new IllegalArgumentException: for updating timestamps must be preserved.

Answer 2

You can customise the FtpInboundFileSynchonizer with remote filter during construction.

Set the filter to FtpPersistentAcceptOnceFileListFilter to configure filtering based on timestamp changes.

Remote filter when set will be used in the synchronizeToLocalDirectory to filter the remote view and download the files.

This will be utilised in this line in doInSession

List<F> filteredFiles = filterFiles(files); 

You will also need to set the local filter to FileSystemPersistentAcceptOnceFileListFilter to process local side changes accrodingly based on the remote file changes.

Also, you'll need to provide the your own metadata store or use PropertiesPersistingMetadataStore for simple case to persist the local and remote file metadata changes so it can survive the server restarts.

More here http://docs.spring.io/spring-integration/reference/html/ftp.html#ftp-inbound