INT-2898 Add Persistent FileListFilters

- Abstract implementation
- Implementation for the file system
- Implementation for remote files ((S)FTP)

JIRA: https://jira.springsource.org/browse/INT-2898

INT-2889 Polishing

- Add (S)FTP test cases
- Docs

spring-integration-core/src/main/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStore.java

@@ -28,6 +28,7 @@ import java.util.Properties;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.util.Assert;
@@ -62,6 +63,7 @@ public class PropertiesPersistingMetadataStore implements MetadataStore, Initial
 		this.baseDirectory = baseDirectory;
 	}
 
+	@Override
 	public void afterPropertiesSet() throws Exception {
 		File baseDir = new File(baseDirectory);
 		baseDir.mkdirs();
@@ -78,20 +80,22 @@ public class PropertiesPersistingMetadataStore implements MetadataStore, Initial
 		this.loadMetadata();
 	}
 
+	@Override
 	public void put(String key, String value) {
 		this.metadata.setProperty(key, value);
 	}
 
+	@Override
 	public String get(String key) {
 		return this.metadata.getProperty(key);
 	}
 
 	@Override
-	@SuppressWarnings("uchecked")
 	public String remove(String key) {
 		return (String) this.metadata.remove(key);
 	}
 
+	@Override
 	public void destroy() throws Exception {
 		this.saveMetadata();
 	}
@@ -100,12 +104,12 @@ public class PropertiesPersistingMetadataStore implements MetadataStore, Initial
 		OutputStream outputStream = null;
 		try {
 			outputStream = new BufferedOutputStream(new FileOutputStream(this.file));
-			this.persister.store(this.metadata, outputStream, "Last feed entry");
+			this.persister.store(this.metadata, outputStream, "Last entry");
 		}
 		catch (IOException e) {
 			// not fatal for the functionality of the component
-			logger.warn("Failed to persist feed entry. This may result in a duplicate "
-					+ "feed entry after this component is restarted.", e);
+			logger.warn("Failed to persist entry. This may result in a duplicate "
+					+ "entry after this component is restarted.", e);
 		}
 		finally {
 			try {
@@ -128,8 +132,8 @@ public class PropertiesPersistingMetadataStore implements MetadataStore, Initial
 		}
 		catch (Exception e) {
 			// not fatal for the functionality of the component
-			logger.warn("Failed to load feed entry from the persistent store. This may result in a duplicate " +
-					"feed entry after this component is restarted", e);
+			logger.warn("Failed to load entry from the persistent store. This may result in a duplicate " +
+					"entry after this component is restarted", e);
 		}
 		finally {
 			try {

spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java

@@ -0,0 +1,92 @@
+/*
+ * Copyright 2013 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.file.filters;
+
+import org.springframework.integration.metadata.MetadataStore;
+import org.springframework.util.Assert;
+
+/**
+ * Stores "seen" files in a MetadataStore to survive application restarts.
+ * The default key is 'prefix' plus the absolute file name; value is the timestamp of the file.
+ * Files are deemed as already 'seen' if they exist in the store and have the
+ * same modified time as the current file.
+ *
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends AbstractFileListFilter<F> {
+
+	protected final MetadataStore store;
+
+	protected final String prefix;
+
+	private final Object monitor = new Object();
+
+	public AbstractPersistentAcceptOnceFileListFilter(MetadataStore store, String prefix) {
+		Assert.notNull(store, "'store' cannot be null");
+		Assert.notNull(prefix, "'prefix' cannot be null");
+		this.store = store;
+		this.prefix = prefix;
+	}
+
+	@Override
+	protected boolean accept(F file) {
+		String key = buildKey(file);
+		synchronized(monitor) {
+			String value = store.get(key);
+			if (value != null && isEqual(file, value)) {
+				return false;
+			}
+			store.put(key, value(file));
+		}
+		return true;
+	}
+
+	/**
+	 * The default value stored for the key is the last modified date.
+	 * @param file The file.
+	 * @return The value to store for the file.
+	 */
+	private String value(F file) {
+		return Long.toString(this.modified(file));
+	}
+
+	/**
+	 * Override this method if you wish to use something other than the
+	 * modified timestamp to determine equality.
+	 * @param file The file.
+	 * @param value The current value for the key in the store
+	 * @return
+	 */
+	protected boolean isEqual(F file, String value) {
+		return Long.valueOf(value).longValue() == this.modified(file);
+	}
+
+	/**
+	 * The default key is the {@link #prefix} plus the full filename.
+	 * @param file The file.
+	 * @return The key.
+	 */
+	protected String buildKey(F file) {
+		return this.prefix + this.fileName(file);
+	}
+
+	protected abstract long modified(F file);
+
+	protected abstract String fileName(F file);
+
+}

spring-integration-file/src/main/java/org/springframework/integration/file/filters/FileSystemPersistentAcceptOnceFileListFilter.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2013 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.file.filters;
+
+
+import java.io.File;
+
+import org.springframework.integration.metadata.MetadataStore;
+
+/**
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+public class FileSystemPersistentAcceptOnceFileListFilter extends AbstractPersistentAcceptOnceFileListFilter<File> {
+
+	public FileSystemPersistentAcceptOnceFileListFilter(MetadataStore store, String prefix) {
+		super(store, prefix);
+	}
+
+	@Override
+	protected long modified(File file) {
+		return file.lastModified();
+	}
+
+	@Override
+	protected String fileName(File file) {
+		return file.getAbsolutePath();
+	}
+
+}

spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourcePersistentFilterIntegrationTests-context.xml

@@ -0,0 +1,26 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	   xmlns:p="http://www.springframework.org/schema/p" xmlns:util="http://www.springframework.org/schema/util"
+	   xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans.xsd http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd">
+
+	<!-- under test -->
+	<bean id="pollableFileSource" class="org.springframework.integration.file.FileReadingMessageSource"
+		  p:directory="file:${java.io.tmpdir}/FileReadingMessageSourcePersistentFilterIntegrationTests"
+		  p:filter-ref="persistentFilter"/>
+
+	<!-- persistent filter -->
+	<bean id="persistentFilter" class="org.springframework.integration.file.filters.FileSystemPersistentAcceptOnceFileListFilter">
+		<constructor-arg ref="ppms" />
+		<constructor-arg value="frmsPersistTest" />
+	</bean>
+
+	<bean id="ppms" class="org.springframework.integration.metadata.PropertiesPersistingMetadataStore">
+		<property name="baseDirectory"
+			value="#{T(System).getProperty('java.io.tmpdir') + T(java.io.File).separator + 'FileReadingMessageSourcePersistentFilterIntegrationTests.meta'}"/>
+	</bean>
+
+	<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer" />
+
+</beans>

spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourcePersistentFilterIntegrationTests.java

@@ -0,0 +1,126 @@
+/*
+ * Copyright 2013 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.
+ * You may obtain a copy of the License at
+ *
+ *	  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.integration.file;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertNull;
+
+import java.io.File;
+
+import org.junit.After;
+import org.junit.AfterClass;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.springframework.beans.DirectFieldAccessor;
+import org.springframework.context.support.AbstractApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+import org.springframework.integration.Message;
+
+/**
+ * @author Iwein Fuld
+ * @author Gary Russell
+ */
+public class FileReadingMessageSourcePersistentFilterIntegrationTests {
+
+	AbstractApplicationContext context;
+
+	FileReadingMessageSource pollableFileSource;
+
+	private static File inputDir;
+
+	@AfterClass
+	public static void cleanUp() throws Throwable {
+	  if(inputDir.exists()) {
+		  inputDir.delete();
+	  }
+	}
+
+	@BeforeClass
+	public static void setupInputDir() {
+		inputDir = new File(System.getProperty("java.io.tmpdir") + "/"
+				+ FileReadingMessageSourcePersistentFilterIntegrationTests.class.getSimpleName());
+		inputDir.mkdir();
+	}
+
+	@Before
+	public void generateTestFiles() throws Exception {
+		File.createTempFile("test", null, inputDir).setLastModified(System.currentTimeMillis() - 1000);
+		File.createTempFile("test", null, inputDir).setLastModified(System.currentTimeMillis() - 1000);
+		File.createTempFile("test", null, inputDir).setLastModified(System.currentTimeMillis() - 1000);
+		this.loadContextAndGetMessageSource();
+	}
+
+	private void loadContextAndGetMessageSource() {
+		this.context = new ClassPathXmlApplicationContext(this.getClass().getSimpleName() + "-context.xml",
+				this.getClass());
+		this.pollableFileSource = context.getBean(FileReadingMessageSource.class);
+	}
+
+	@After
+	public void cleanoutInputDir() throws Exception {
+		File[] listFiles = inputDir.listFiles();
+		for (int i = 0; i < listFiles.length; i++) {
+			listFiles[i].delete();
+		}
+	}
+
+	@AfterClass
+	public static void removeInputDir() throws Exception {
+		inputDir.delete();
+		File persistDir = new File(System.getProperty("java.io.tmpdir") + "/"
+				+ FileReadingMessageSourcePersistentFilterIntegrationTests.class.getSimpleName()
+				+ ".meta");
+		File persist = new File(persistDir, "metadata-store.properties");
+		persist.delete();
+		persistDir.delete();
+	}
+
+
+	@Test
+	public void configured() throws Exception {
+		DirectFieldAccessor accessor = new DirectFieldAccessor(pollableFileSource);
+		assertEquals(inputDir, accessor.getPropertyValue("directory"));
+	}
+
+	@Test
+	public void getFiles() throws Exception {
+		Message<File> received1 = pollableFileSource.receive();
+		System.out.println("receive files round 1");
+		assertNotNull("This should return the first message", received1);
+		pollableFileSource.onSend(received1);
+		Message<File> received2 = pollableFileSource.receive();
+		assertNotNull(received2);
+		pollableFileSource.onSend(received2);
+		Message<File> received3 = pollableFileSource.receive();
+		assertNotNull(received3);
+		pollableFileSource.onSend(received3);
+		assertNotSame(received1 + " == " + received2, received1.getPayload(), received2.getPayload());
+		assertNotSame(received1 + " == " + received3, received1.getPayload(), received3.getPayload());
+		assertNotSame(received2 + " == " + received3, received2.getPayload(), received3.getPayload());
+		this.context.destroy();
+
+		this.loadContextAndGetMessageSource();
+		Message<File> received4 = pollableFileSource.receive();
+		assertNull(received4);
+		this.context.destroy();
+	}
+
+}

spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterTests.java

@@ -0,0 +1,47 @@
+/*
+ * Copyright 2013 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.file.filters;
+
+import static org.junit.Assert.assertTrue;
+
+import java.io.File;
+
+import org.junit.Test;
+
+import org.springframework.integration.metadata.MetadataStore;
+import org.springframework.integration.metadata.SimpleMetadataStore;
+
+/**
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+public class PersistentAcceptOnceFileListFilterTests {
+
+	@Test
+	public void testFileSystem() throws Exception {
+		MetadataStore store = new SimpleMetadataStore();
+		FileSystemPersistentAcceptOnceFileListFilter filter = new FileSystemPersistentAcceptOnceFileListFilter(store, "foo:");
+		File file = File.createTempFile("foo", ".txt");
+		assertTrue(filter.filterFiles(new File[] {file}).size() == 1);
+		assertTrue(filter.filterFiles(new File[] {file}).size() == 0);
+		file.setLastModified(27L);
+		assertTrue(filter.filterFiles(new File[] {file}).size() == 1);
+		assertTrue(filter.filterFiles(new File[] {file}).size() == 0);
+		file.delete();
+	}
+
+}

spring-integration-ftp/src/main/java/org/springframework/integration/ftp/filters/FtpPersistentAcceptOnceFileListFilter.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2013 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.ftp.filters;
+
+
+import org.apache.commons.net.ftp.FTPFile;
+
+import org.springframework.integration.file.filters.AbstractPersistentAcceptOnceFileListFilter;
+import org.springframework.integration.metadata.MetadataStore;
+
+/**
+ * Since the super class deems files as 'not seen' if the timestamp is different, remote file
+ * users should use the adapter's preserve-timestamp option. Otherwise if a file is re-fetched
+ * it will have a new timestamp.
+ *
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+public class FtpPersistentAcceptOnceFileListFilter extends AbstractPersistentAcceptOnceFileListFilter<FTPFile> {
+
+	public FtpPersistentAcceptOnceFileListFilter(MetadataStore store, String prefix) {
+		super(store, prefix);
+	}
+
+	@Override
+	protected long modified(FTPFile file) {
+		return file.getTimestamp().getTimeInMillis();
+	}
+
+	@Override
+	protected String fileName(FTPFile file) {
+		return file.getName();
+	}
+
+}

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

@@ -33,11 +33,14 @@ import java.io.OutputStream;
 import java.util.ArrayList;
 import java.util.Calendar;
 import java.util.Collection;
+import java.util.List;
+import java.util.Queue;
 
 import org.apache.commons.net.ftp.FTPClient;
 import org.apache.commons.net.ftp.FTPFile;
 import org.hamcrest.Matchers;
 import org.junit.After;
+import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mockito;
 
@@ -47,8 +50,13 @@ import org.springframework.expression.spel.SpelParserConfiguration;
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.Message;
 import org.springframework.integration.expression.ExpressionUtils;
+import org.springframework.integration.file.filters.CompositeFileListFilter;
+import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter;
 import org.springframework.integration.ftp.session.AbstractFtpSessionFactory;
+import org.springframework.integration.metadata.PropertiesPersistingMetadataStore;
+import org.springframework.integration.test.util.TestUtils;
 
 /**
  * @author Oleg Zhurakousky
@@ -61,6 +69,7 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 
 	private static FTPClient ftpClient = mock(FTPClient.class);
 
+	@Before
 	@After
 	public void cleanup(){
 		File file = new File("test");
@@ -86,7 +95,16 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 		synchronizer.setDeleteRemoteFiles(true);
 		synchronizer.setPreserveTimestamp(true);
 		synchronizer.setRemoteDirectory("remote-test-dir");
-		synchronizer.setFilter(new FtpRegexPatternFileListFilter(".*\\.test$"));
+		FtpRegexPatternFileListFilter patternFilter = new FtpRegexPatternFileListFilter(".*\\.test$");
+		PropertiesPersistingMetadataStore store = new PropertiesPersistingMetadataStore();
+		store.setBaseDirectory("test");
+		FtpPersistentAcceptOnceFileListFilter persistFilter =
+				new FtpPersistentAcceptOnceFileListFilter(store, "foo");
+		List<FileListFilter<FTPFile>> filters = new ArrayList<FileListFilter<FTPFile>>();
+		filters.add(persistFilter);
+		filters.add(patternFilter);
+		CompositeFileListFilter<FTPFile> filter = new CompositeFileListFilter<FTPFile>(filters);
+		synchronizer.setFilter(filter);
 		synchronizer.setIntegrationEvaluationContext(ExpressionUtils.createStandardEvaluationContext());
 
 		ExpressionParser expressionParser = new SpelExpressionParser(new SpelParserConfiguration(true, true));
@@ -120,28 +138,49 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 
 		assertTrue(new File("test/A.TEST.a").exists());
 		assertTrue(new File("test/B.TEST.a").exists());
+
+		TestUtils.getPropertyValue(ms, "localFileListFilter.seen", Queue.class).clear();
+
+		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();
+		assertNull(nothing);
+
 	}
 
 
 	public static class TestFtpSessionFactory extends AbstractFtpSessionFactory<FTPClient> {
 
+		private final Collection<Object> ftpFiles = new ArrayList<Object>();
+
+		private void init() {
+			String[] files = new File("remote-test-dir").list();
+			for (String fileName : files) {
+				FTPFile file = new FTPFile();
+				file.setName(fileName);
+				file.setType(FTPFile.FILE_TYPE);
+				Calendar calendar = Calendar.getInstance();
+				calendar.add(Calendar.DATE, 1);
+				file.setTimestamp(calendar);
+				ftpFiles.add(file);
+			}
+
+		}
+
 		@Override
 		protected FTPClient createClientInstance() {
+			if (this.ftpFiles.size() == 0) {
+				this.init();
+			}
+
 			try {
 				when(ftpClient.getReplyCode()).thenReturn(250);
 				when(ftpClient.login("kermit", "frog")).thenReturn(true);
 				when(ftpClient.changeWorkingDirectory(Mockito.anyString())).thenReturn(true);
 
 				String[] files = new File("remote-test-dir").list();
-				Collection<Object> ftpFiles = new ArrayList<Object>();
 				for (String fileName : files) {
-					FTPFile file = new FTPFile();
-					file.setName(fileName);
-					file.setType(FTPFile.FILE_TYPE);
-					Calendar calendar = Calendar.getInstance();
-					calendar.add(Calendar.DATE, 1);
-					file.setTimestamp(calendar);
-					ftpFiles.add(file);
 					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[]{}));

spring-integration-sftp/src/main/java/org/springframework/integration/sftp/filters/SftpPersistentAcceptOnceFileListFilter.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2013 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.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.integration.sftp.filters;
+
+
+
+import org.springframework.integration.file.filters.AbstractPersistentAcceptOnceFileListFilter;
+import org.springframework.integration.metadata.MetadataStore;
+
+import com.jcraft.jsch.ChannelSftp.LsEntry;
+
+/**
+ * Since the super class deems files as 'not seen' if the timestamp is different, remote file
+ * users should use the adapter's preserve-timestamp option. Otherwise if a file is re-fetched
+ * it will have a new timestamp.
+ *
+ * @author Gary Russell
+ * @since 3.0
+ *
+ */
+public class SftpPersistentAcceptOnceFileListFilter extends AbstractPersistentAcceptOnceFileListFilter<LsEntry> {
+
+	public SftpPersistentAcceptOnceFileListFilter(MetadataStore store, String prefix) {
+		super(store, prefix);
+	}
+
+	@Override
+	protected long modified(LsEntry file) {
+		return ((long) file.getAttrs().getMTime()) * 1000;
+	}
+
+	@Override
+	protected String fileName(LsEntry file) {
+		return file.getLongname();
+	}
+
+
+}

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java

@@ -30,19 +30,28 @@ import static org.mockito.Mockito.when;
 
 import java.io.File;
 import java.io.FileInputStream;
+import java.util.ArrayList;
 import java.util.Calendar;
+import java.util.List;
+import java.util.Queue;
 import java.util.Vector;
 
 import org.hamcrest.Matchers;
 import org.junit.After;
+import org.junit.Before;
 import org.junit.Test;
 
 import org.springframework.integration.Message;
 import org.springframework.integration.expression.ExpressionUtils;
+import org.springframework.integration.file.filters.CompositeFileListFilter;
+import org.springframework.integration.file.filters.FileListFilter;
 import org.springframework.integration.file.remote.session.Session;
+import org.springframework.integration.metadata.PropertiesPersistingMetadataStore;
+import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter;
 import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
 import org.springframework.integration.sftp.session.SftpTestSessionFactory;
+import org.springframework.integration.test.util.TestUtils;
 
 import com.jcraft.jsch.ChannelSftp;
 import com.jcraft.jsch.ChannelSftp.LsEntry;
@@ -59,6 +68,7 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
 
 	private static com.jcraft.jsch.Session jschSession = mock(com.jcraft.jsch.Session.class);
 
+	@Before
 	@After
 	public void cleanup(){
 		File file = new File("test");
@@ -87,7 +97,16 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
 		synchronizer.setDeleteRemoteFiles(true);
 		synchronizer.setPreserveTimestamp(true);
 		synchronizer.setRemoteDirectory("remote-test-dir");
-		synchronizer.setFilter(new SftpRegexPatternFileListFilter(".*\\.test$"));
+		SftpRegexPatternFileListFilter patternFilter = new SftpRegexPatternFileListFilter(".*\\.test$");
+		PropertiesPersistingMetadataStore store = new PropertiesPersistingMetadataStore();
+		store.setBaseDirectory("test");
+		SftpPersistentAcceptOnceFileListFilter persistFilter =
+				new SftpPersistentAcceptOnceFileListFilter(store, "foo");
+		List<FileListFilter<LsEntry>> filters = new ArrayList<FileListFilter<LsEntry>>();
+		filters.add(persistFilter);
+		filters.add(patternFilter);
+		CompositeFileListFilter<LsEntry> filter = new CompositeFileListFilter<LsEntry>(filters);
+		synchronizer.setFilter(filter);
 		synchronizer.setIntegrationEvaluationContext(ExpressionUtils.createStandardEvaluationContext());
 
 		SftpInboundFileSynchronizingMessageSource ms =
@@ -115,28 +134,48 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
 
 		assertTrue(new File("test/a.test").exists());
 		assertTrue(new File("test/b.test").exists());
+
+		TestUtils.getPropertyValue(ms, "localFileListFilter.seen", Queue.class).clear();
+
+		new File("test/a.test").delete();
+		new File("test/b.test").delete();
+		// the remote filter should prevent a re-fetch
+		nothing =  ms.receive();
+		assertNull(nothing);
+
 	}
 
 	public static class TestSftpSessionFactory extends DefaultSftpSessionFactory {
 
+		private final Vector<LsEntry> sftpEntries = new Vector<LsEntry>();
+
+		private void init() {
+			String[] files = new File("remote-test-dir").list();
+			for (String fileName : files) {
+				LsEntry lsEntry = mock(LsEntry.class);
+				SftpATTRS attributes = mock(SftpATTRS.class);
+				when(lsEntry.getAttrs()).thenReturn(attributes);
+
+				Calendar calendar = Calendar.getInstance();
+				calendar.add(Calendar.DATE, 1);
+				when(lsEntry.getAttrs().getMTime()).thenReturn(new Long(calendar.getTimeInMillis() / 1000).intValue());
+				when(lsEntry.getFilename()).thenReturn(fileName);
+				when(lsEntry.getLongname()).thenReturn(fileName);
+				sftpEntries.add(lsEntry);
+			}
+		}
 
 		@Override
 		public Session<LsEntry> getSession() {
+			if (this.sftpEntries.size() == 0) {
+				this.init();
+			}
+
 			try {
 				ChannelSftp channel = mock(ChannelSftp.class);
 
 				String[] files = new File("remote-test-dir").list();
-				Vector<LsEntry> sftpEntries = new Vector<LsEntry>();
 				for (String fileName : files) {
-					LsEntry lsEntry = mock(LsEntry.class);
-					SftpATTRS attributes = mock(SftpATTRS.class);
-					when(lsEntry.getAttrs()).thenReturn(attributes);
-
-					Calendar calendar = Calendar.getInstance();
-					calendar.add(Calendar.DATE, 1);
-					when(lsEntry.getAttrs().getMTime()).thenReturn(new Long(calendar.getTimeInMillis() / 1000).intValue());
-					when(lsEntry.getFilename()).thenReturn(fileName);
-					sftpEntries.add(lsEntry);
 					when(channel.get("remote-test-dir/"+fileName)).thenReturn(new FileInputStream("remote-test-dir/" + fileName));
 				}
 				when(channel.ls("remote-test-dir")).thenReturn(sftpEntries);
@@ -148,4 +187,5 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
 			}
 		}
 	}
+
 }

src/reference/docbook/file.xml

@@ -35,6 +35,14 @@
       <interfacename>FileListFilter</interfacename>. By default, an
       <classname>AcceptOnceFileListFilter</classname> is used. This filter
       ensures files are picked up only once from the directory.
+      <note>
+        The <classname>AcceptOnceFileListFilter</classname> stores its state in memory. If you wish the
+        state to survive a system restart, consider using the
+        <classname>FileSystemPersistentAcceptOnceFileListFilter</classname> instead. This filter stores
+        the accepted file names in a <interfacename>MetadataStore</interfacename>. The framework supplies
+        several store implementations (such as Redis), or you can provide your own. This filter matches on
+        the filename and modified time.
+      </note>
       <programlisting language="xml"><![CDATA[<bean id="pollableFileSource"
     class="org.springframework.integration.file.FileReadingMessageSource"
     p:inputDirectory="${input.directory}"

src/reference/docbook/ftp.xml

@@ -184,14 +184,37 @@ protected void postProcessClientBeforeConnect(T client) throws IOException {
 	(e.g. <code>filename-regex=".*\.test$"</code>). And of course if you need complete control you can use <code>filter</code>
 	attribute and provide a reference to any custom implementation of the
 	<classname>org.springframework.integration.file.filters.FileListFilter</classname>, a strategy interface for filtering a
-	list of files. This filter determines which remote files are retrieved.
+	list of files. This filter determines which remote files are retrieved. You can also combine a pattern based filter
+	with other filters, such as an <classname>AcceptOnceFileListFilter</classname> to avoid synchronizing files that
+	have previously been fetched, by using a <classname>CompositeFileListFilter</classname>.
   	</para>
+	<para>
+        The <classname>AcceptOnceFileListFilter</classname> stores its state in memory. If you wish the
+        state to survive a system restart, consider using the
+        <classname>FtpPersistentAcceptOnceFileListFilter</classname> instead. This filter stores
+        the accepted file names in a <interfacename>MetadataStore</interfacename>. The framework supplies
+        several store implementations (such as Redis), or you can provide your own. This filter matches on
+        the filename and the remote modified time.
+	</para>  	
 	<note>
-	Beginning with 3.0, you can also specify a filter used to filter the files locally, once they have
-	been retrieved. The default filter is an <classname>AcceptOnceFileListFilter</classname> which prevents processing
-	files with the same name multiple times in the same JVM execution; this can now be overridden
-	(for example with an <classname>AcceptAllFileListFilter</classname>), using the <code>local-filter</code> attribute.
-	Previously, the default	<classname>AcceptOnceFileListFilter</classname> could not be overridden.
+	  <para>
+		Beginning with <emphasis>version 3.0</emphasis>, you can also specify a filter used to filter the files locally, once they have
+		been retrieved. The default filter is an <classname>AcceptOnceFileListFilter</classname> which prevents processing
+		files with the same name multiple times in the same JVM execution; this can now be overridden
+		(for example with an <classname>AcceptAllFileListFilter</classname>), using the <code>local-filter</code> attribute.
+		Previously, the default	<classname>AcceptOnceFileListFilter</classname> could not be overridden.
+	  </para>
+	  <para>
+        The <classname>AcceptOnceFileListFilter</classname> stores its state in memory. If you wish the
+        state to survive a system restart, consider using the
+        <classname>FileSystemPersistentAcceptOnceFileListFilter</classname> as a local filter instead. This filter stores
+        the accepted file names in a <interfacename>MetadataStore</interfacename>. The framework supplies
+        several store implementations (such as Redis), or you can provide your own.
+        <important>
+        	This filter compares the filename and modified timestamp. If you wish to use this technique to avoid a
+        	re-synchronized file from being processed, you should use the <code>preserve-timestamp</code> attribute discussed above.
+        </important>
+	  </para>
 	</note>
 	<para>
 	The 'remote-file-separator' attribute allows you to configure a

src/reference/docbook/sftp.xml

@@ -297,14 +297,37 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp
 	(e.g. <code>filename-regex=".*\.test$"</code>). And of course if you need complete control you can use the <code>filter</code>
 	attribute to provide a reference to a custom implementation of the
 	<classname>org.springframework.integration.file.filters.FileListFilter</classname> - a strategy interface for filtering a
-	list of files. This filter determines which remote files are retrieved.
+	list of files. This filter determines which remote files are retrieved. You can also combine a pattern based filter
+	with other filters, such as an <classname>AcceptOnceFileListFilter</classname> to avoid synchronizing files that
+	have previously been fetched, by using a <classname>CompositeFileListFilter</classname>.
   	</para>
+	<para>
+        The <classname>AcceptOnceFileListFilter</classname> stores its state in memory. If you wish the
+        state to survive a system restart, consider using the
+        <classname>SftpPersistentAcceptOnceFileListFilter</classname> instead. This filter stores
+        the accepted file names in a <interfacename>MetadataStore</interfacename>. The framework supplies
+        several store implementations (such as Redis), or you can provide your own. This filter matches on
+        the filename and the remote modified time.
+	</para>  	
 	<note>
-	Beginning with 3.0, you can also specify a filter used to filter the files locally, once they have
-	been retrieved. The default filter is an <classname>AcceptOnceFileListFilter</classname> which prevents processing
-	files with the same name multiple times in the same JVM execution; this can now be overridden
-	(for example with an <classname>AcceptAllFileListFilter</classname>), using the <code>local-filter</code> attribute.
-	Previously, the default	<classname>AcceptOnceFileListFilter</classname> could not be overridden.
+	  <para>
+		Beginning with <emphasis>version 3.0</emphasis>, you can also specify a filter used to filter the files locally, once they have
+		been retrieved. The default filter is an <classname>AcceptOnceFileListFilter</classname> which prevents processing
+		files with the same name multiple times in the same JVM execution; this can now be overridden
+		(for example with an <classname>AcceptAllFileListFilter</classname>), using the <code>local-filter</code> attribute.
+		Previously, the default	<classname>AcceptOnceFileListFilter</classname> could not be overridden.
+	  </para>
+	  <para>
+        The <classname>AcceptOnceFileListFilter</classname> stores its state in memory. If you wish the
+        state to survive a system restart, consider using the
+        <classname>FileSystemPersistentAcceptOnceFileListFilter</classname> as a local filter instead. This filter stores
+        the accepted file names in a <interfacename>MetadataStore</interfacename>. The framework supplies
+        several store implementations (such as Redis), or you can provide your own.
+        <important>
+        	This filter compares the filename and modified timestamp. If you wish to use this technique to avoid a
+        	re-synchronized file from being processed, you should use the <code>preserve-timestamp</code> attribute discussed above.
+        </important>
+	  </para>
 	</note>
   	<para>
   	Please refer to the schema for more detail on these attributes.

src/reference/docbook/whats-new.xml

@@ -604,5 +604,13 @@
 				For more information, see <xref linkend="redis"/>.
 			</para>
 		</section>
+		<section id="3.0-filelistfilter">
+			<title>Persistend File List Filters (file, (S)FTP)</title>
+			<para>
+				New <classname>FileListFilter</classname>s that use a persistent <classname>MetadataStore</classname> are
+				now available. These can be used to prevent duplicate files after a system restart. See
+				<xref linkend="file-reading"/>, <xref linkend="ftp-inbound"/>, and <xref linkend="sftp-inbound"/> for more information. 
+			</para>
+		</section>
 	</section>
 </chapter>