INT-4305: (S)FTP: Remove Local File Before Rename

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

The `File.renameTo()` operation may fail, therefore the content of the
local file isn't changed, but since we change `setLastModified()` anyway,
this file might be eligible for local polling.
So, we end up with the same content from local file in a new message,
meanwhile we expect a new content from the remote file

* Check the `File.renameTo()` result and attempt to `delete()`
for existing local file
* When file isn't renames remove the remote file from the `filter` to let
it be transferred one more time on the next poll.
The local file might be opened for processing, so this way we postpone a fresh
remote file for the future poll rounds
* Modify `copyFileToLocalDirectory()` to return `boolean` to reflect the fact of copy.
This way we check the real number of transferred files

**Cherry-pick to 4.3.x**

Do not transfer remote file if we can't remove local one

Polishing Log Messages

spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java

@@ -39,6 +39,7 @@ import org.springframework.expression.Expression;
 import org.springframework.expression.common.LiteralExpression;
 import org.springframework.integration.expression.ExpressionUtils;
 import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.file.filters.ResettableFileListFilter;
 import org.springframework.integration.file.filters.ReversibleFileListFilter;
 import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.file.remote.session.Session;
@@ -257,24 +258,28 @@ public abstract class AbstractInboundFileSynchronizer<F>
 						}
 						filteredFiles = newList;
 					}
+
+					int copied = filteredFiles.size();
+
 					for (F file : filteredFiles) {
 						try {
 							if (file != null) {
-								copyFileToLocalDirectory(
-										remoteDirectory, file, localDirectory,
-										session);
+								if (!copyFileToLocalDirectory(remoteDirectory, file, localDirectory, session)) {
+									copied--;
+								}
 							}
 						}
 						catch (RuntimeException e1) {
 							rollbackFromFileToListEnd(filteredFiles, file);
 							throw e1;
 						}
-						catch (IOException e2) {
+						catch (IOException e1) {
 							rollbackFromFileToListEnd(filteredFiles, file);
-							throw e2;
+							throw e1;
 						}
 					}
-					return filteredFiles.size();
+
+					return copied;
 				}
 				else {
 					return 0;
@@ -296,21 +301,21 @@ public abstract class AbstractInboundFileSynchronizer<F>
 		}
 	}
 
-	protected void copyFileToLocalDirectory(String remoteDirectoryPath, F remoteFile, File localDirectory,
+	protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, F remoteFile, File localDirectory,
 			Session<F> session) throws IOException {
 		String remoteFileName = this.getFilename(remoteFile);
 		String localFileName = this.generateLocalFileName(remoteFileName);
 		String remoteFilePath = remoteDirectoryPath != null
 				? (remoteDirectoryPath + this.remoteFileSeparator + remoteFileName)
 				: remoteFileName;
+
 		if (!this.isFile(remoteFile)) {
 			if (this.logger.isDebugEnabled()) {
 				this.logger.debug("cannot copy, not a file: " + remoteFilePath);
 			}
-			return;
+			return false;
 		}
 
-
 		long modified = getModified(remoteFile);
 
 		File localFile = new File(localDirectory, localFileName);
@@ -320,40 +325,90 @@ public abstract class AbstractInboundFileSynchronizer<F>
 					localFileName.replaceAll("/", Matcher.quoteReplacement(File.separator)).contains(File.separator)) {
 				localFile.getParentFile().mkdirs(); //NOSONAR - will fail on the writing below
 			}
-			String tempFileName = localFile.getAbsolutePath() + this.temporaryFileSuffix;
-			File tempFile = new File(tempFileName);
-			OutputStream outputStream = new BufferedOutputStream(new FileOutputStream(tempFile));
-			try {
-				session.read(remoteFilePath, outputStream);
-			}
-			catch (Exception e) {
-				if (e instanceof RuntimeException) {
-					throw (RuntimeException) e;
-				}
-				else {
-					throw new MessagingException("Failure occurred while copying from remote to local directory", e);
-				}
-			}
-			finally {
-				try {
-					outputStream.close();
-				}
-				catch (Exception ignored2) {
+
+			boolean transfer = true;
+
+			if (exists && !localFile.delete()) {
+				transfer = false;
+				if (this.logger.isInfoEnabled()) {
+					this.logger.info("Cannot delete local file '" + localFile +
+							"' in order to transfer modified remote file '" + remoteFile + "'. " +
+							"The local file may be busy in some other process.");
 				}
 			}
 
-			if (tempFile.renameTo(localFile)) {
-				if (this.deleteRemoteFiles) {
-					session.remove(remoteFilePath);
-					if (this.logger.isDebugEnabled()) {
-						this.logger.debug("deleted " + remoteFilePath);
+			boolean renamed = false;
+
+			if (transfer) {
+				String tempFileName = localFile.getAbsolutePath() + this.temporaryFileSuffix;
+				File tempFile = new File(tempFileName);
+
+				OutputStream outputStream = new BufferedOutputStream(new FileOutputStream(tempFile));
+				try {
+					session.read(remoteFilePath, outputStream);
+				}
+				catch (Exception e) {
+					if (e instanceof RuntimeException) {
+						throw (RuntimeException) e;
+					}
+					else {
+						throw new MessagingException("Failure occurred while copying '" + remoteFilePath
+								+ "' from the remote to the local directory", e);
+					}
+				}
+				finally {
+					try {
+						outputStream.close();
+					}
+					catch (Exception ignored2) {
+					}
+				}
+
+				renamed = tempFile.renameTo(localFile);
+
+				if (!renamed) {
+					if (localFile.delete()) {
+						renamed = tempFile.renameTo(localFile);
+						if (!renamed && this.logger.isInfoEnabled()) {
+							this.logger.info("Cannot rename '"
+									+ tempFileName
+									+ "' to local file '" + localFile + "' after deleting. " +
+									"The local file may be busy in some other process.");
+						}
+					}
+					else if (this.logger.isInfoEnabled()) {
+						this.logger.info("Cannot delete local file '" + localFile +
+								"'. The local file may be busy in some other process.");
 					}
 				}
 			}
-			if (this.preserveTimestamp) {
-				localFile.setLastModified(modified);
+
+			if (renamed) {
+				if (this.deleteRemoteFiles) {
+					session.remove(remoteFilePath);
+					if (this.logger.isDebugEnabled()) {
+						this.logger.debug("deleted remote file: " + remoteFilePath);
+					}
+				}
+				if (this.preserveTimestamp) {
+					localFile.setLastModified(modified);
+				}
+				return true;
+			}
+			else if (this.filter instanceof ResettableFileListFilter) {
+				if (this.logger.isInfoEnabled()) {
+					this.logger.info("Reverting the remote file '" + remoteFile +
+							"' from the filter for a subsequent transfer attempt");
+				}
+				((ResettableFileListFilter<F>) this.filter).remove(remoteFile);
 			}
 		}
+		else if (this.logger.isWarnEnabled()) {
+			this.logger.warn("The remote file '" + remoteFile + "' has not been transferred " +
+					"to the existing local file '" + localFile + "'. Consider removing the local file.");
+		}
+
+		return false;
 	}
 
 	private String generateLocalFileName(String remoteFileName) {

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

@@ -60,27 +60,27 @@ public abstract class RemoteFileTestSupport {
 	protected volatile File targetLocalDirectory;
 
 	public File getSourceRemoteDirectory() {
-		return sourceRemoteDirectory;
+		return this.sourceRemoteDirectory;
 	}
 
 	public File getTargetRemoteDirectory() {
-		return targetRemoteDirectory;
+		return this.targetRemoteDirectory;
 	}
 
 	public String getTargetRemoteDirectoryName() {
-		return targetRemoteDirectory.getAbsolutePath() + File.separator;
+		return this.targetRemoteDirectory.getAbsolutePath() + File.separator;
 	}
 
 	public File getSourceLocalDirectory() {
-		return sourceLocalDirectory;
+		return this.sourceLocalDirectory;
 	}
 
 	public File getTargetLocalDirectory() {
-		return targetLocalDirectory;
+		return this.targetLocalDirectory;
 	}
 
 	public String getTargetLocalDirectoryName() {
-		return targetLocalDirectory.getAbsolutePath() + File.separator;
+		return this.targetLocalDirectory.getAbsolutePath() + File.separator;
 	}
 
 	/**

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

@@ -70,12 +70,13 @@ public class AbstractRemoteFileSynchronizerTests {
 			}
 
 			@Override
-			protected void copyFileToLocalDirectory(String remoteDirectoryPath, String remoteFile, File localDirectory,
-					Session<String> session) throws IOException {
+			protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, String remoteFile,
+					File localDirectory, Session<String> session) throws IOException {
 				if ("bar".equals(remoteFile) && failWhenCopyingBar.getAndSet(false)) {
 					throw new IOException("fail");
 				}
 				count.incrementAndGet();
+				return true;
 			}
 
 		};
@@ -162,9 +163,10 @@ public class AbstractRemoteFileSynchronizerTests {
 			}
 
 			@Override
-			protected void copyFileToLocalDirectory(String remoteDirectoryPath, String remoteFile, File localDirectory,
-					Session<String> session) throws IOException {
+			protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, String remoteFile,
+					File localDirectory, Session<String> session) throws IOException {
 				count.incrementAndGet();
+				return true;
 			}
 
 		};

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

@@ -27,6 +27,9 @@ import static org.junit.Assert.assertThat;
 
 import java.io.ByteArrayInputStream;
 import java.io.File;
+import java.io.FileOutputStream;
+import java.io.FileReader;
+import java.io.IOException;
 import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
 import java.util.List;
@@ -64,6 +67,7 @@ import org.springframework.messaging.Message;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
 import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.util.FileCopyUtils;
 
 /**
  * @author Artem Bilan
@@ -82,7 +86,7 @@ public class FtpTests extends FtpTestSupport {
 	private ApplicationContext context;
 
 	@Test
-	public void testFtpInboundFlow() {
+	public void testFtpInboundFlow() throws IOException {
 		QueueChannel out = new QueueChannel();
 		IntegrationFlow flow = IntegrationFlows.from(Ftp.inboundAdapter(sessionFactory())
 						.preserveTimestamp(true)
@@ -112,6 +116,10 @@ public class FtpTests extends FtpTestSupport {
 		assertNull(out.receive(10));
 
 		File remoteFile = new File(this.sourceRemoteDirectory, " " + prefix() + "Source1.txt");
+
+		FileOutputStream fos = new FileOutputStream(remoteFile);
+		fos.write("New content".getBytes());
+		fos.close();
 		remoteFile.setLastModified(System.currentTimeMillis() - 1000 * 60 * 60 * 24);
 
 		message = out.receive(10_000);
@@ -120,6 +128,7 @@ public class FtpTests extends FtpTestSupport {
 		assertThat(payload, instanceOf(File.class));
 		file = (File) payload;
 		assertEquals(" FTPSOURCE1.TXT.a", file.getName());
+		assertEquals("New content", FileCopyUtils.copyToString(new FileReader(file)));
 
 		MessageSource<?> source = context.getBean(FtpInboundFileSynchronizingMessageSource.class);
 		assertThat(TestUtils.getPropertyValue(source, "maxFetchSize"), equalTo(10));