From 72dd39bccf5ac235deb2a4926a41895271d899b6 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 6 Nov 2013 19:44:52 -0500 Subject: [PATCH] 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 --- .../PropertiesPersistingMetadataStore.java | 16 ++- ...actPersistentAcceptOnceFileListFilter.java | 92 +++++++++++++ ...temPersistentAcceptOnceFileListFilter.java | 44 ++++++ ...rsistentFilterIntegrationTests-context.xml | 26 ++++ ...ourcePersistentFilterIntegrationTests.java | 126 ++++++++++++++++++ ...rsistentAcceptOnceFileListFilterTests.java | 47 +++++++ ...FtpPersistentAcceptOnceFileListFilter.java | 49 +++++++ ...oundRemoteFileSystemSynchronizerTests.java | 57 ++++++-- ...ftpPersistentAcceptOnceFileListFilter.java | 51 +++++++ ...oundRemoteFileSystemSynchronizerTests.java | 62 +++++++-- src/reference/docbook/file.xml | 8 ++ src/reference/docbook/ftp.xml | 35 ++++- src/reference/docbook/sftp.xml | 35 ++++- src/reference/docbook/whats-new.xml | 8 ++ 14 files changed, 618 insertions(+), 38 deletions(-) create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java create mode 100644 spring-integration-file/src/main/java/org/springframework/integration/file/filters/FileSystemPersistentAcceptOnceFileListFilter.java create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourcePersistentFilterIntegrationTests-context.xml create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourcePersistentFilterIntegrationTests.java create mode 100644 spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterTests.java create mode 100644 spring-integration-ftp/src/main/java/org/springframework/integration/ftp/filters/FtpPersistentAcceptOnceFileListFilter.java create mode 100644 spring-integration-sftp/src/main/java/org/springframework/integration/sftp/filters/SftpPersistentAcceptOnceFileListFilter.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStore.java b/spring-integration-core/src/main/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStore.java index 26c35231ec..b6599cef0d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/metadata/PropertiesPersistingMetadataStore.java +++ b/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 { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/AbstractPersistentAcceptOnceFileListFilter.java new file mode 100644 index 0000000000..0f752473f5 --- /dev/null +++ b/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 extends AbstractFileListFilter { + + 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); + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/filters/FileSystemPersistentAcceptOnceFileListFilter.java b/spring-integration-file/src/main/java/org/springframework/integration/file/filters/FileSystemPersistentAcceptOnceFileListFilter.java new file mode 100644 index 0000000000..6fb2a2c19f --- /dev/null +++ b/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 { + + 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(); + } + +} diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourcePersistentFilterIntegrationTests-context.xml b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourcePersistentFilterIntegrationTests-context.xml new file mode 100644 index 0000000000..1292e3905a --- /dev/null +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourcePersistentFilterIntegrationTests-context.xml @@ -0,0 +1,26 @@ + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourcePersistentFilterIntegrationTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/FileReadingMessageSourcePersistentFilterIntegrationTests.java new file mode 100644 index 0000000000..293738d1c5 --- /dev/null +++ b/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 received1 = pollableFileSource.receive(); + System.out.println("receive files round 1"); + assertNotNull("This should return the first message", received1); + pollableFileSource.onSend(received1); + Message received2 = pollableFileSource.receive(); + assertNotNull(received2); + pollableFileSource.onSend(received2); + Message 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 received4 = pollableFileSource.receive(); + assertNull(received4); + this.context.destroy(); + } + +} diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/filters/PersistentAcceptOnceFileListFilterTests.java new file mode 100644 index 0000000000..0fe1e51dbd --- /dev/null +++ b/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(); + } + +} diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/filters/FtpPersistentAcceptOnceFileListFilter.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/filters/FtpPersistentAcceptOnceFileListFilter.java new file mode 100644 index 0000000000..91ae1bd3f3 --- /dev/null +++ b/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 { + + 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(); + } + +} diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java index 7106219858..51e24f9cac 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java +++ b/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> filters = new ArrayList>(); + filters.add(persistFilter); + filters.add(patternFilter); + CompositeFileListFilter filter = new CompositeFileListFilter(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 { + private final Collection ftpFiles = new ArrayList(); + + 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 ftpFiles = new ArrayList(); 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[]{})); diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/filters/SftpPersistentAcceptOnceFileListFilter.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/filters/SftpPersistentAcceptOnceFileListFilter.java new file mode 100644 index 0000000000..0140c2d8e2 --- /dev/null +++ b/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 { + + 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(); + } + + +} diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java index c8fa674f48..516cadee84 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java +++ b/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> filters = new ArrayList>(); + filters.add(persistFilter); + filters.add(patternFilter); + CompositeFileListFilter filter = new CompositeFileListFilter(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 sftpEntries = new Vector(); + + 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 getSession() { + if (this.sftpEntries.size() == 0) { + this.init(); + } + try { ChannelSftp channel = mock(ChannelSftp.class); String[] files = new File("remote-test-dir").list(); - Vector sftpEntries = new Vector(); 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 { } } } + } diff --git a/src/reference/docbook/file.xml b/src/reference/docbook/file.xml index b1f2c46897..9cc835c8e2 100644 --- a/src/reference/docbook/file.xml +++ b/src/reference/docbook/file.xml @@ -35,6 +35,14 @@ FileListFilter. By default, an AcceptOnceFileListFilter is used. This filter ensures files are picked up only once from the directory. + + The AcceptOnceFileListFilter stores its state in memory. If you wish the + state to survive a system restart, consider using the + FileSystemPersistentAcceptOnceFileListFilter instead. This filter stores + the accepted file names in a MetadataStore. The framework supplies + several store implementations (such as Redis), or you can provide your own. This filter matches on + the filename and modified time. + filename-regex=".*\.test$"). And of course if you need complete control you can use filter attribute and provide a reference to any custom implementation of the org.springframework.integration.file.filters.FileListFilter, 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 AcceptOnceFileListFilter to avoid synchronizing files that + have previously been fetched, by using a CompositeFileListFilter. + + The AcceptOnceFileListFilter stores its state in memory. If you wish the + state to survive a system restart, consider using the + FtpPersistentAcceptOnceFileListFilter instead. This filter stores + the accepted file names in a MetadataStore. 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. + - 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 AcceptOnceFileListFilter which prevents processing - files with the same name multiple times in the same JVM execution; this can now be overridden - (for example with an AcceptAllFileListFilter), using the local-filter attribute. - Previously, the default AcceptOnceFileListFilter could not be overridden. + + Beginning with version 3.0, you can also specify a filter used to filter the files locally, once they have + been retrieved. The default filter is an AcceptOnceFileListFilter which prevents processing + files with the same name multiple times in the same JVM execution; this can now be overridden + (for example with an AcceptAllFileListFilter), using the local-filter attribute. + Previously, the default AcceptOnceFileListFilter could not be overridden. + + + The AcceptOnceFileListFilter stores its state in memory. If you wish the + state to survive a system restart, consider using the + FileSystemPersistentAcceptOnceFileListFilter as a local filter instead. This filter stores + the accepted file names in a MetadataStore. The framework supplies + several store implementations (such as Redis), or you can provide your own. + + 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 preserve-timestamp attribute discussed above. + + The 'remote-file-separator' attribute allows you to configure a diff --git a/src/reference/docbook/sftp.xml b/src/reference/docbook/sftp.xml index 5111c75d5f..0d8c56a677 100644 --- a/src/reference/docbook/sftp.xml +++ b/src/reference/docbook/sftp.xml @@ -297,14 +297,37 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/sftp (e.g. filename-regex=".*\.test$"). And of course if you need complete control you can use the filter attribute to provide a reference to a custom implementation of the org.springframework.integration.file.filters.FileListFilter - 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 AcceptOnceFileListFilter to avoid synchronizing files that + have previously been fetched, by using a CompositeFileListFilter. + + The AcceptOnceFileListFilter stores its state in memory. If you wish the + state to survive a system restart, consider using the + SftpPersistentAcceptOnceFileListFilter instead. This filter stores + the accepted file names in a MetadataStore. 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. + - 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 AcceptOnceFileListFilter which prevents processing - files with the same name multiple times in the same JVM execution; this can now be overridden - (for example with an AcceptAllFileListFilter), using the local-filter attribute. - Previously, the default AcceptOnceFileListFilter could not be overridden. + + Beginning with version 3.0, you can also specify a filter used to filter the files locally, once they have + been retrieved. The default filter is an AcceptOnceFileListFilter which prevents processing + files with the same name multiple times in the same JVM execution; this can now be overridden + (for example with an AcceptAllFileListFilter), using the local-filter attribute. + Previously, the default AcceptOnceFileListFilter could not be overridden. + + + The AcceptOnceFileListFilter stores its state in memory. If you wish the + state to survive a system restart, consider using the + FileSystemPersistentAcceptOnceFileListFilter as a local filter instead. This filter stores + the accepted file names in a MetadataStore. The framework supplies + several store implementations (such as Redis), or you can provide your own. + + 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 preserve-timestamp attribute discussed above. + + Please refer to the schema for more detail on these attributes. diff --git a/src/reference/docbook/whats-new.xml b/src/reference/docbook/whats-new.xml index c1271ca497..c10f243959 100644 --- a/src/reference/docbook/whats-new.xml +++ b/src/reference/docbook/whats-new.xml @@ -604,5 +604,13 @@ For more information, see . +
+ Persistend File List Filters (file, (S)FTP) + + New FileListFilters that use a persistent MetadataStore are + now available. These can be used to prevent duplicate files after a system restart. See + , , and for more information. + +