INT-4115: (S)FtpPersistentFileFilter by Default

JIRA: https://jira.spring.io/browse/INT-4115

Apply `(S)FtpPersistentAcceptOnceFileListFilter` for the `(S)FtpInboundFileSynchronizer` by default to avoid cases to sync the same remote files to the local directory again.
Especially when `localFileName` strategy is applied and we end up with new local files, but with the same remote content

Accept `(S)FtpPersistentAcceptOnceFileListFilter` for streaming adapters

Make `doSetFilter()` as `protected final`

Fix tests after rebase

Fix "What's New" after rebase

Compose `PersistentAcceptOnceFileListFilter` together with the regex or pattern filters
Document such a behavior

Address PR comments for formatting and typo

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2012 the original author or authors.
+ * Copyright 2002-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -17,8 +17,10 @@
 package org.springframework.integration.ftp.config;
 
 import org.springframework.integration.file.config.AbstractRemoteFileInboundChannelAdapterParser;
+import org.springframework.integration.file.filters.AbstractPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.synchronizer.InboundFileSynchronizer;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
 import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
 import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizer;
@@ -29,6 +31,8 @@ import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizingMe
  *
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 2.0
  */
 public class FtpInboundChannelAdapterParser extends AbstractRemoteFileInboundChannelAdapterParser {
@@ -53,4 +57,9 @@ public class FtpInboundChannelAdapterParser extends AbstractRemoteFileInboundCha
 		return FtpRegexPatternFileListFilter.class;
 	}
 
+	@Override
+	protected Class<? extends AbstractPersistentAcceptOnceFileListFilter<?>> getPersistentAcceptOnceFileListFilterClass() {
+		return FtpPersistentAcceptOnceFileListFilter.class;
+	}
+
 }

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParser.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -18,8 +18,10 @@ package org.springframework.integration.ftp.config;
 
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.file.config.AbstractRemoteFileStreamingInboundChannelAdapterParser;
+import org.springframework.integration.file.filters.AbstractPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.RemoteFileOperations;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
 import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
 import org.springframework.integration.ftp.inbound.FtpStreamingMessageSource;
@@ -27,6 +29,8 @@ import org.springframework.integration.ftp.session.FtpRemoteFileTemplate;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 4.3
  *
  */
@@ -52,4 +56,9 @@ public class FtpStreamingInboundChannelAdapterParser extends AbstractRemoteFileS
 		return FtpRegexPatternFileListFilter.class;
 	}
 
+	@Override
+	protected Class<? extends AbstractPersistentAcceptOnceFileListFilter<?>> getPersistentAcceptOnceFileListFilterClass() {
+		return FtpPersistentAcceptOnceFileListFilter.class;
+	}
+
 }

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpInboundChannelAdapterSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 the original author or authors.
+ * Copyright 2014-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -22,17 +22,22 @@ import java.util.Comparator;
 import org.apache.commons.net.ftp.FTPFile;
 
 import org.springframework.integration.file.dsl.RemoteFileInboundChannelAdapterSpec;
+import org.springframework.integration.file.filters.CompositeFileListFilter;
+import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
 import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
 import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizer;
 import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizingMessageSource;
+import org.springframework.integration.metadata.SimpleMetadataStore;
 
 /**
  * A {@link RemoteFileInboundChannelAdapterSpec} for a
  * {@link FtpInboundFileSynchronizingMessageSource}.
  *
  * @author Artem Bilan
+ *
  * @since 5.0
  */
 public class FtpInboundChannelAdapterSpec
@@ -52,7 +57,7 @@ public class FtpInboundChannelAdapterSpec
 	 */
 	@Override
 	public FtpInboundChannelAdapterSpec patternFilter(String pattern) {
-		return filter(new FtpSimplePatternFileListFilter(pattern));
+		return filter(composeFilters(new FtpSimplePatternFileListFilter(pattern)));
 	}
 
 	/**
@@ -63,7 +68,15 @@ public class FtpInboundChannelAdapterSpec
 	 */
 	@Override
 	public FtpInboundChannelAdapterSpec regexFilter(String regex) {
-		return filter(new FtpRegexPatternFileListFilter(regex));
+		return filter(composeFilters(new FtpRegexPatternFileListFilter(regex)));
+	}
+
+	@SuppressWarnings("unchecked")
+	private CompositeFileListFilter<FTPFile> composeFilters(FileListFilter<FTPFile> fileListFilter) {
+		CompositeFileListFilter<FTPFile> compositeFileListFilter = new CompositeFileListFilter<>();
+		compositeFileListFilter.addFilters(fileListFilter,
+				new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "ftpMessageSource"));
+		return compositeFileListFilter;
 	}
 
 }

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/dsl/FtpStreamingInboundChannelAdapterSpec.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2016 the original author or authors.
+ * Copyright 2014-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -21,11 +21,15 @@ import java.util.Comparator;
 import org.apache.commons.net.ftp.FTPFile;
 
 import org.springframework.integration.file.dsl.RemoteFileStreamingInboundChannelAdapterSpec;
+import org.springframework.integration.file.filters.CompositeFileListFilter;
+import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.AbstractFileInfo;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
 import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
 import org.springframework.integration.ftp.inbound.FtpStreamingMessageSource;
+import org.springframework.integration.metadata.SimpleMetadataStore;
 
 /**
  * A {@link RemoteFileStreamingInboundChannelAdapterSpec} for a
@@ -51,7 +55,7 @@ public class FtpStreamingInboundChannelAdapterSpec
 	 */
 	@Override
 	public FtpStreamingInboundChannelAdapterSpec patternFilter(String pattern) {
-		return filter(new FtpSimplePatternFileListFilter(pattern));
+		return filter(composeFilters(new FtpSimplePatternFileListFilter(pattern)));
 	}
 
 	/**
@@ -62,7 +66,15 @@ public class FtpStreamingInboundChannelAdapterSpec
 	 */
 	@Override
 	public FtpStreamingInboundChannelAdapterSpec regexFilter(String regex) {
-		return filter(new FtpRegexPatternFileListFilter(regex));
+		return filter(composeFilters(new FtpRegexPatternFileListFilter(regex)));
+	}
+
+	@SuppressWarnings("unchecked")
+	private CompositeFileListFilter<FTPFile> composeFilters(FileListFilter<FTPFile> fileListFilter) {
+		CompositeFileListFilter<FTPFile> compositeFileListFilter = new CompositeFileListFilter<>();
+		compositeFileListFilter.addFilters(fileListFilter,
+				new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "ftpStreamingMessageSource"));
+		return compositeFileListFilter;
 	}
 
 }

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -22,6 +22,8 @@ import org.springframework.expression.common.LiteralExpression;
 import org.springframework.integration.file.remote.session.Session;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
+import org.springframework.integration.metadata.SimpleMetadataStore;
 
 /**
  * An implementation of {@link AbstractInboundFileSynchronizer} for FTP.
@@ -36,12 +38,12 @@ public class FtpInboundFileSynchronizer extends AbstractInboundFileSynchronizer<
 
 	/**
 	 * Create a synchronizer with the {@link SessionFactory} used to acquire {@link Session} instances.
-	 *
 	 * @param sessionFactory The session factory.
 	 */
 	public FtpInboundFileSynchronizer(SessionFactory<FTPFile> sessionFactory) {
 		super(sessionFactory);
-		setRemoteDirectoryExpression(new LiteralExpression(null));
+		doSetRemoteDirectoryExpression(new LiteralExpression(null));
+		doSetFilter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "ftpMessageSource"));
 	}
 
 	@Override

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSource.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -26,12 +26,16 @@ import org.apache.commons.net.ftp.FTPFile;
 import org.springframework.integration.file.remote.AbstractFileInfo;
 import org.springframework.integration.file.remote.AbstractRemoteFileStreamingMessageSource;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.session.FtpFileInfo;
+import org.springframework.integration.metadata.SimpleMetadataStore;
 
 /**
  * Message source for streaming FTP remote file contents.
  *
  * @author Gary Russell
+ * @author Artem Bilan
+ *
  * @since 4.3
  *
  */
@@ -42,7 +46,7 @@ public class FtpStreamingMessageSource extends AbstractRemoteFileStreamingMessag
 	 * @param template the template.
 	 */
 	public FtpStreamingMessageSource(RemoteFileTemplate<FTPFile> template) {
-		super(template, null);
+		this(template, null);
 	}
 
 	/**
@@ -55,6 +59,7 @@ public class FtpStreamingMessageSource extends AbstractRemoteFileStreamingMessag
 	public FtpStreamingMessageSource(RemoteFileTemplate<FTPFile> template,
 			Comparator<AbstractFileInfo<FTPFile>> comparator) {
 		super(template, comparator);
+		doSetFilter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "ftpStreamingMessageSource"));
 	}
 
 	@Override

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/FtpParserInboundTests-context.xml

@@ -3,12 +3,9 @@
 	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
 	   xmlns:ftp="http://www.springframework.org/schema/integration/ftp"
 	   xmlns:int="http://www.springframework.org/schema/integration"
-	   xmlns:context="http://www.springframework.org/schema/context"
-	   xmlns:ftps="http://www.springframework.org/schema/integration/ftps"
 	   xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
-	    http://www.springframework.org/schema/integration/ftps http://www.springframework.org/schema/integration/ftp/spring-integration-ftps.xsd
         http://www.springframework.org/schema/integration  http://www.springframework.org/schema/integration/spring-integration.xsd
-       http://www.springframework.org/schema/integration/ftp http://www.springframework.org/schema/integration/ftp/spring-integration-ftp.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
+       http://www.springframework.org/schema/integration/ftp http://www.springframework.org/schema/integration/ftp/spring-integration-ftp.xsd">
 
 	<bean id="ftpSessionFactory" class="org.springframework.integration.ftp.session.DefaultFtpSessionFactory">
 		<property name="host" value="localhost"/>

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -16,10 +16,12 @@
 
 package org.springframework.integration.ftp.config;
 
+import static org.hamcrest.Matchers.instanceOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -27,6 +29,8 @@ import static org.mockito.Mockito.when;
 import java.lang.reflect.Method;
 import java.util.Collection;
 import java.util.Comparator;
+import java.util.Iterator;
+import java.util.Set;
 import java.util.concurrent.PriorityBlockingQueue;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -39,9 +43,11 @@ import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.ApplicationContext;
 import org.springframework.expression.Expression;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
+import org.springframework.integration.file.filters.CompositeFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.session.CachingSessionFactory;
 import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
 import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizer;
 import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizingMessageSource;
@@ -59,6 +65,7 @@ import org.springframework.util.ReflectionUtils;
  * @author Mark Fisher
  * @author Gary Russell
  * @author Gunnar Hillert
+ * @author Artem Bilan
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
@@ -103,8 +110,16 @@ public class FtpInboundChannelAdapterParserTests {
 		String remoteFileSeparator = (String) TestUtils.getPropertyValue(fisync, "remoteFileSeparator");
 		assertNotNull(remoteFileSeparator);
 		assertEquals("", remoteFileSeparator);
-		FtpSimplePatternFileListFilter filter = (FtpSimplePatternFileListFilter) TestUtils.getPropertyValue(fisync, "filter");
+
+		FileListFilter<?> filter = TestUtils.getPropertyValue(fisync, "filter", FileListFilter.class);
 		assertNotNull(filter);
+		assertThat(filter, instanceOf(CompositeFileListFilter.class));
+		Set<?> fileFilters = TestUtils.getPropertyValue(filter, "fileFilters", Set.class);
+
+		Iterator<?> filtersIterator = fileFilters.iterator();
+		assertThat(filtersIterator.next(), instanceOf(FtpSimplePatternFileListFilter.class));
+		assertThat(filtersIterator.next(), instanceOf(FtpPersistentAcceptOnceFileListFilter.class));
+
 		Object sessionFactory = TestUtils.getPropertyValue(fisync, "remoteFileTemplate.sessionFactory");
 		assertTrue(DefaultFtpSessionFactory.class.isAssignableFrom(sessionFactory.getClass()));
 		FileListFilter<?> acceptAllFilter = context.getBean("acceptAllFilter", FileListFilter.class);

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpStreamingInboundChannelAdapterParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016 the original author or authors.
+ * Copyright 2016-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -26,13 +26,19 @@ import static org.junit.Assert.assertThat;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import java.util.Iterator;
+import java.util.Set;
+
 import org.junit.Test;
 import org.junit.runner.RunWith;
 
 import org.springframework.beans.factory.FactoryBean;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
+import org.springframework.integration.file.filters.CompositeFileListFilter;
+import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.session.CachingSessionFactory;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter;
 import org.springframework.integration.ftp.inbound.FtpStreamingMessageSource;
 import org.springframework.integration.ftp.session.DefaultFtpSessionFactory;
@@ -45,6 +51,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 /**
  * @author Gary Russell
+ * @author Artem Bilan
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
@@ -71,7 +78,16 @@ public class FtpStreamingInboundChannelAdapterParserTests {
 
 		assertNotNull(TestUtils.getPropertyValue(source, "comparator"));
 		assertThat(TestUtils.getPropertyValue(source, "remoteFileSeparator", String.class), equalTo("X"));
-		assertThat(TestUtils.getPropertyValue(source, "filter"), instanceOf(FtpSimplePatternFileListFilter.class));
+
+		FileListFilter<?> filter = TestUtils.getPropertyValue(source, "filter", FileListFilter.class);
+		assertNotNull(filter);
+		assertThat(filter, instanceOf(CompositeFileListFilter.class));
+		Set<?> fileFilters = TestUtils.getPropertyValue(filter, "fileFilters", Set.class);
+
+		Iterator<?> filtersIterator = fileFilters.iterator();
+		assertThat(filtersIterator.next(), instanceOf(FtpSimplePatternFileListFilter.class));
+		assertThat(filtersIterator.next(), instanceOf(FtpPersistentAcceptOnceFileListFilter.class));
+
 		assertSame(this.csf, TestUtils.getPropertyValue(source, "remoteFileTemplate.sessionFactory"));
 		assertEquals(31, TestUtils.getPropertyValue(source, "maxFetchSize"));
 	}

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-2017 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -76,9 +76,8 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 	public void cleanup() {
 		File file = new File("test");
 		if (file.exists()) {
-			String[] files = file.list();
-			for (String fileName : files) {
-				new File(file, fileName).delete();
+			for (File f : file.listFiles()) {
+				f.delete();
 			}
 			file.delete();
 		}
@@ -127,7 +126,7 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 		localFileListFilter.addFilter(localAcceptOnceFilter);
 		ms.setLocalFilter(localFileListFilter);
 		ms.afterPropertiesSet();
-		Message<File> atestFile =  ms.receive();
+		Message<File> atestFile = ms.receive();
 		assertNotNull(atestFile);
 		assertEquals("A.TEST.a", atestFile.getPayload().getName());
 		// The test remote files are created with the current timestamp + 1 day.
@@ -135,13 +134,13 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 
 		assertEquals("A.TEST.a", atestFile.getHeaders().get(FileHeaders.FILENAME));
 
-		Message<File> btestFile =  ms.receive();
+		Message<File> btestFile = ms.receive();
 		assertNotNull(btestFile);
 		assertEquals("B.TEST.a", btestFile.getPayload().getName());
 		// The test remote files are created with the current timestamp + 1 day.
 		assertThat(atestFile.getPayload().lastModified(), Matchers.greaterThan(System.currentTimeMillis()));
 
-		Message<File> nothing =  ms.receive();
+		Message<File> nothing = ms.receive();
 		assertNull(nothing);
 
 		// two times because on the third receive (above) the internal queue will be empty, so it will attempt
@@ -155,7 +154,7 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 		new File("test/A.TEST.a").delete();
 		new File("test/B.TEST.a").delete();
 		// the remote filter should prevent a re-fetch
-		nothing =  ms.receive();
+		nothing = ms.receive();
 		assertNull(nothing);
 
 		ms.stop();
@@ -164,6 +163,35 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 	}
 
 
+	@Test
+	public void testSyncRemoteFileOnlyOnceByDefault() throws Exception {
+		File localDirectory = new File("test");
+		localDirectory.mkdir();
+
+		TestFtpSessionFactory ftpSessionFactory = new TestFtpSessionFactory();
+		ftpSessionFactory.setUsername("kermit");
+		ftpSessionFactory.setPassword("frog");
+		FtpInboundFileSynchronizer synchronizer = spy(new FtpInboundFileSynchronizer(ftpSessionFactory));
+		synchronizer.setRemoteDirectory("remote-test-dir");
+
+		synchronizer.setBeanFactory(mock(BeanFactory.class));
+		synchronizer.afterPropertiesSet();
+
+		synchronizer.synchronizeToLocalDirectory(localDirectory);
+
+
+		File[] files = localDirectory.listFiles();
+		assertEquals(3, files.length);
+
+		for (File f : files) {
+			f.delete();
+		}
+
+		synchronizer.synchronizeToLocalDirectory(localDirectory);
+
+		assertEquals(0, localDirectory.list().length);
+	}
+
 	public static class TestFtpSessionFactory extends AbstractFtpSessionFactory<FTPClient> {
 
 		private final Collection<FTPFile> ftpFiles = new ArrayList<FTPFile>();
@@ -198,7 +226,8 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 					when(ftpClient.retrieveFile(Mockito.eq("remote-test-dir/" + fileName),
 							Mockito.any(OutputStream.class))).thenReturn(true);
 				}
-				when(ftpClient.listFiles("remote-test-dir")).thenReturn(ftpFiles.toArray(new FTPFile[ftpFiles.size()]));
+				when(ftpClient.listFiles("remote-test-dir"))
+						.thenReturn(ftpFiles.toArray(new FTPFile[ftpFiles.size()]));
 				when(ftpClient.deleteFile(Mockito.anyString())).thenReturn(true);
 				return ftpClient;
 			}
@@ -206,6 +235,7 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 				throw new RuntimeException("Failed to create mock client", e);
 			}
 		}
+
 	}
 
 }

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpStreamingMessageSourceTests.java

@@ -39,6 +39,7 @@ import org.springframework.integration.config.EnableIntegration;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
 import org.springframework.integration.file.FileHeaders;
+import org.springframework.integration.file.filters.AcceptAllFileListFilter;
 import org.springframework.integration.file.remote.FileInfo;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.ftp.FtpTestSupport;
@@ -128,6 +129,7 @@ public class FtpStreamingMessageSourceTests extends FtpTestSupport {
 		public MessageSource<InputStream> ftpMessageSource() {
 			FtpStreamingMessageSource messageSource = new FtpStreamingMessageSource(template(),
 					Comparator.comparing(FileInfo::getFilename));
+			messageSource.setFilter(new AcceptAllFileListFilter<>());
 			messageSource.setRemoteDirectory("ftpSource/");
 			return messageSource;
 		}