From 72bd4e5bcf4b40bd4537092b7fb026e6751a1aae Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Tue, 27 Jun 2017 12:23:26 -0400 Subject: [PATCH] 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 --- .../AbstractInboundFileSynchronizer.java | 125 +++++++++++++----- .../file/remote/RemoteFileTestSupport.java | 12 +- .../AbstractRemoteFileSynchronizerTests.java | 10 +- .../integration/ftp/dsl/FtpTests.java | 11 +- 4 files changed, 112 insertions(+), 46 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java index a8c80db46e..7f42d5b456 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java +++ b/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 } 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 } } - protected void copyFileToLocalDirectory(String remoteDirectoryPath, F remoteFile, File localDirectory, + protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, F remoteFile, File localDirectory, Session 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 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) 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) { diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileTestSupport.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileTestSupport.java index 659988292c..9eb1464c0a 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/RemoteFileTestSupport.java +++ b/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; } /** diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java index 92ec8a150f..2e69dedfde 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java +++ b/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 session) throws IOException { + protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, String remoteFile, + File localDirectory, Session 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 session) throws IOException { + protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, String remoteFile, + File localDirectory, Session session) throws IOException { count.incrementAndGet(); + return true; } }; diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/dsl/FtpTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/dsl/FtpTests.java index 4d2f516065..71be2c4292 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/dsl/FtpTests.java +++ b/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));