INT-4232: (S)FTP inbound: Fix lastModified Logic

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

Currently to implement the synchronization logic when remote file is modified there is only one way: remove local file and provide a `localFilter` which can be clear from the removed file as well.

The Reference Manual claims that `FileSystemPersistentAcceptOnceFileListFilter` can let us pick up a fresh version of the remote file if we configure `preserveTimestamp = true`, but actually `AbstractInboundFileSynchronizer` just bypass that logic because it check a local file for existence.

* Modify `AbstractInboundFileSynchronizer` to compare `lastModified` of the remote file with local version if `preserveTimestamp == true`
* Also replace `AcceptOnceFileListFilter` to the `FileSystemPersistentAcceptOnceFileListFilter` since the first one doesn't care about `lastModified`.
If `preserveTimestamp` isn't in use the `FileSystemPersistentAcceptOnceFileListFilter` works fully similar to the `AcceptOnceFileListFilter`

**Cherry-pick to 4.3.x except test-case**

Make `SftpStreamingMessageSourceTests` compatible with Windows

Looks like SFTP embedded server provides different permissions for files on different OSs

Doc Polishing

spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.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.
@@ -59,6 +59,7 @@ import org.springframework.util.ObjectUtils;
  * @author Oleg Zhurakousky
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 2.0
  */
 public abstract class AbstractInboundFileSynchronizer<F>
@@ -301,8 +302,11 @@ public abstract class AbstractInboundFileSynchronizer<F>
 			return;
 		}
 
+
+		long modified = getModified(remoteFile);
+
 		File localFile = new File(localDirectory, localFileName);
-		if (!localFile.exists()) {
+		if (!localFile.exists() || (this.preserveTimestamp && modified != localFile.lastModified())) {
 			String tempFileName = localFile.getAbsolutePath() + this.temporaryFileSuffix;
 			File tempFile = new File(tempFileName);
 			OutputStream outputStream = new BufferedOutputStream(new FileOutputStream(tempFile));
@@ -334,7 +338,7 @@ public abstract class AbstractInboundFileSynchronizer<F>
 				}
 			}
 			if (this.preserveTimestamp) {
-				localFile.setLastModified(getModified(remoteFile));
+				localFile.setLastModified(modified);
 			}
 		}
 	}

spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.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.
@@ -30,7 +30,9 @@ import org.springframework.integration.file.FileReadingMessageSource;
 import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
 import org.springframework.integration.file.filters.CompositeFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.file.filters.FileSystemPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.file.filters.RegexPatternFileListFilter;
+import org.springframework.integration.metadata.SimpleMetadataStore;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
 
@@ -55,6 +57,7 @@ import org.springframework.util.Assert;
  * @author Josh Long
  * @author Oleg Zhurakousky
  * @author Gary Russell
+ * @author Artem Bilan
  */
 public abstract class AbstractInboundFileSynchronizingMessageSource<F>
 		extends AbstractFetchLimitingMessageSource<File> implements Lifecycle {
@@ -82,7 +85,7 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F>
 	 */
 	private final FileReadingMessageSource fileSource;
 
-	private volatile FileListFilter<File> localFileListFilter = new AcceptOnceFileListFilter<File>();
+	private volatile FileListFilter<File> localFileListFilter;
 
 
 	public AbstractInboundFileSynchronizingMessageSource(AbstractInboundFileSynchronizer<F> synchronizer) {
@@ -142,6 +145,10 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F>
 				}
 			}
 			this.fileSource.setDirectory(this.localDirectory);
+			if (this.localFileListFilter == null) {
+				this.localFileListFilter = new FileSystemPersistentAcceptOnceFileListFilter(
+						new SimpleMetadataStore(), getComponentName());
+			}
 			this.fileSource.setFilter(this.buildFilter());
 			if (this.getBeanFactory() != null) {
 				this.fileSource.setBeanFactory(this.getBeanFactory());

spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2015 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.
@@ -41,6 +41,7 @@ import org.springframework.messaging.MessagingException;
 /**
  * @author Gary Russell
  * @author Artem Bilan
+ *
  * @since 4.0.4
  *
  */
@@ -128,6 +129,7 @@ public class AbstractRemoteFileSynchronizerTests {
 		source.setAutoCreateLocalDirectory(true);
 		source.setBeanFactory(mock(BeanFactory.class));
 		source.setMaxFetchSize(1);
+		source.setBeanName("maxFetchSizeSource");
 		source.afterPropertiesSet();
 
 		source.receive();

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/dsl/FtpTests.java

@@ -21,6 +21,7 @@ import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.isOneOf;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertThat;
 
 import java.io.File;
@@ -95,6 +96,18 @@ public class FtpTests extends FtpTestSupport {
 		assertThat(file.getName(), isOneOf(" FTPSOURCE1.TXT.a", "FTPSOURCE2.TXT.a"));
 		assertThat(file.getAbsolutePath(), containsString("localTarget"));
 
+		assertNull(out.receive(10));
+
+		File remoteFile = new File(this.sourceRemoteDirectory, " " + prefix() + "Source1.txt");
+		remoteFile.setLastModified(System.currentTimeMillis() - 1000 * 60 * 60 * 24);
+
+		message = out.receive(10_000);
+		assertNotNull(message);
+		payload = message.getPayload();
+		assertThat(payload, instanceOf(File.class));
+		file = (File) payload;
+		assertEquals(" FTPSOURCE1.TXT.a", file.getName());
+
 		registration.destroy();
 	}
 

spring-integration-ftp/src/test/resources/log4j.properties

@@ -5,5 +5,5 @@ log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
 log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m%n
 
 log4j.category.org.springframework=WARN
-log4j.category.org.springframework.integration=DEBUG
-log4j.category.org.springframework.integration.file=DEBUG
+log4j.category.org.springframework.integration=WARN
+log4j.category.org.springframework.integration.file=WARN

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

@@ -77,7 +77,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
 		assertThat(new String(received.getPayload()), equalTo("source1"));
 		String fileInfo = (String) received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO);
 		assertThat(fileInfo, containsString("remoteDirectory\":\"sftpSource"));
-		assertThat(fileInfo, containsString("permissions\":\"-rw-r--r--"));
+		assertThat(fileInfo, containsString("permissions\":"));
 		assertThat(fileInfo, containsString("size\":7"));
 		assertThat(fileInfo, containsString("directory\":false"));
 		assertThat(fileInfo, containsString("filename\":\" sftpSource1.txt"));
@@ -87,7 +87,7 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
 		assertNotNull(received);
 		fileInfo = (String) received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO);
 		assertThat(fileInfo, containsString("remoteDirectory\":\"sftpSource"));
-		assertThat(fileInfo, containsString("permissions\":\"-rw-r--r--"));
+		assertThat(fileInfo, containsString("permissions\":"));
 		assertThat(fileInfo, containsString("size\":7"));
 		assertThat(fileInfo, containsString("directory\":false"));
 		assertThat(fileInfo, containsString("filename\":\"sftpSource2.txt"));

src/reference/asciidoc/ftp.adoc

@@ -291,8 +291,9 @@ Unless your application removes files after processing, the adapter will re-proc
 Also, if you configure the `filter` to use a `FtpPersistentAcceptOnceFileListFilter`, and the remote file timestamp changes (causing it to be re-fetched), the default local filter will not allow this new file to be processed.
 
 Use the `local-filter` attribute to configure the behavior of the local file system filter.
-To solve these particular use cases, you can use a `FileSystemPersistentAcceptOnceFileListFilter` as a local filter instead.
-This filter also stores the accepted file names and modified timestamp in an instance of the`MetadataStore` strategy (<<metadata-store>>), and will detect the change in the local file modified time.
+Starting with _verion 4.3.8_, a `FileSystemPersistentAcceptOnceFileListFilter` is configured by default.
+This filter stores the accepted file names and modified timestamp in an instance of the `MetadataStore` strategy (<<metadata-store>>), and will detect changes to the local file modified time.
+The default `MetadataStore` is a `SimpleMetadataStore` which stores state in memory.
 
 Since __version 4.1.5__, these filters have a new property `flushOnUpdate` which will cause them to flush the
 metadata store on every update (if the store implements `Flushable`).

src/reference/asciidoc/sftp.adoc

@@ -353,8 +353,9 @@ Unless your application removes files after processing, the adapter will re-proc
 Also, if you configure the `filter` to use a `FtpPersistentAcceptOnceFileListFilter`, and the remote file timestamp changes (causing it to be re-fetched), the default local filter will not allow this new file to be processed.
 
 Use the `local-filter` attribute to configure the behavior of the local file system filter.
-To solve these particular use cases, you can use a `FileSystemPersistentAcceptOnceFileListFilter` as a local filter instead.
-This filter also stores the accepted file names and modified timestamp in an instance of the`MetadataStore` strategy (<<metadata-store>>), and will detect the change in the local file modified time.
+Starting with _verion 4.3.8_, a `FileSystemPersistentAcceptOnceFileListFilter` is configured by default.
+This filter stores the accepted file names and modified timestamp in an instance of the `MetadataStore` strategy (<<metadata-store>>), and will detect changes to the local file modified time.
+The default `MetadataStore` is a `SimpleMetadataStore` which stores state in memory.
 
 Since __version 4.1.5__, these filters have a new property `flushOnUpdate` which will cause them to flush the
 metadata store on every update (if the store implements `Flushable`).

src/reference/asciidoc/whats-new.adoc

@@ -77,6 +77,8 @@ The file outbound channel adapter (`FileWritingMessageHandler`) now supports the
 ==== (S)FTP Changes
 
 The inbound channel adapters now have a property `max-fetch-size` which is used to limit the number of files fetched during a poll when there are no files currently in the local directory.
+They also are configured with a `FileSystemPersistentAcceptOnceFileListFilter` in the `local-filter` by default.
+See <<ftp-inbound>> and <<sftp-inbound>> for more information.
 
 The regex and pattern filters can now be configured to always pass directories.
 This can be useful when using recursion in the outbound gateways.