From 4d05b9c559fb77aac58be1a3b14b49d3cdfa6496 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Fri, 19 Nov 2010 08:24:00 -0500 Subject: [PATCH] INT-1614 polishing --- ...actInboundRemoteFileSystemSychronizer.java | 20 ++--- ...tpInboundRemoteFileSystemSynchronizer.java | 73 +++++++++--------- .../sftp/inbound/SftpInboundSynchronizer.java | 76 ++++++++++--------- 3 files changed, 89 insertions(+), 80 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/synchronization/AbstractInboundRemoteFileSystemSychronizer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/synchronization/AbstractInboundRemoteFileSystemSychronizer.java index cd2de61b2d..a78bea992f 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/synchronization/AbstractInboundRemoteFileSystemSychronizer.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/synchronization/AbstractInboundRemoteFileSystemSychronizer.java @@ -61,14 +61,6 @@ public abstract class AbstractInboundRemoteFileSystemSychronizer implements I protected EntryAcknowledgmentStrategy entryAcknowledgmentStrategy; - public void setEntryAcknowledgmentStrategy(EntryAcknowledgmentStrategy entryAcknowledgmentStrategy) { - this.entryAcknowledgmentStrategy = entryAcknowledgmentStrategy; - } - - public void setShouldDeleteSourceFile(boolean shouldDeleteSourceFile) { - this.shouldDeleteSourceFile = shouldDeleteSourceFile; - } - public void setLocalDirectory(Resource localDirectory) { this.localDirectory = localDirectory; } @@ -77,6 +69,14 @@ public abstract class AbstractInboundRemoteFileSystemSychronizer implements I this.filter = filter; } + public void setEntryAcknowledgmentStrategy(EntryAcknowledgmentStrategy entryAcknowledgmentStrategy) { + this.entryAcknowledgmentStrategy = entryAcknowledgmentStrategy; + } + + public void setShouldDeleteSourceFile(boolean shouldDeleteSourceFile) { + this.shouldDeleteSourceFile = shouldDeleteSourceFile; + } + /** * @param usefulContextOrClientData * this is context information to be passed to the individual {@link EntryAcknowledgmentStrategy}. @@ -88,10 +88,10 @@ public abstract class AbstractInboundRemoteFileSystemSychronizer implements I * behavior. It's a hook to the file entry after it's been * successfully downloaded. Conceptually, you might delete the * remote one or rename it, etc. - * @throws Throwable + * @throws Exception * escape hatch exception, let the adapter deal with it. */ - protected void acknowledge(Object usefulContextOrClientData, F file) throws Throwable { + protected void acknowledge(Object usefulContextOrClientData, F file) throws Exception { if (this.entryAcknowledgmentStrategy != null) { this.entryAcknowledgmentStrategy.acknowledge(usefulContextOrClientData, file); } diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizer.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizer.java index 7795260500..3ddc018f75 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizer.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizer.java @@ -22,6 +22,8 @@ import java.io.FileOutputStream; import java.io.IOException; import java.util.Collection; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.apache.commons.net.ftp.FTPClient; import org.apache.commons.net.ftp.FTPFile; @@ -40,7 +42,7 @@ import org.springframework.util.Assert; */ public class FtpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemoteFileSystemSychronizer { - protected volatile FtpClientPool clientPool; + private volatile FtpClientPool clientPool; /** @@ -59,39 +61,6 @@ public class FtpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemot } } - private boolean copyFileToLocalDirectory(FTPClient client, FTPFile ftpFile, Resource localDirectory) - throws IOException, FileNotFoundException { - - String remoteFileName = ftpFile.getName(); - String localFileName = localDirectory.getFile().getPath() + "/" + remoteFileName; - File localFile = new File(localFileName); - if (!localFile.exists()) { - String tempFileName = localFileName + - AbstractInboundRemoteFileSystemSynchronizingMessageSource.INCOMPLETE_EXTENSION; - File file = new File(tempFileName); - FileOutputStream fos = new FileOutputStream(file); - try { - client.retrieveFile(remoteFileName, fos); - // Perhaps we have some dispatch of the source file to do? - acknowledge(client, ftpFile); - } - catch (Throwable th) { - if (th instanceof RuntimeException){ - throw (RuntimeException)th; - } - else { - throw new MessagingException("Failed to copy file", th); - } - } - finally { - fos.close(); - } - file.renameTo(localFile); - return true; - } - return false; - } - @Override protected void syncRemoteToLocalFileSystem() { try { @@ -117,11 +86,45 @@ public class FtpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemot } } + private boolean copyFileToLocalDirectory(FTPClient client, FTPFile ftpFile, Resource localDirectory) + throws IOException, FileNotFoundException { + + String remoteFileName = ftpFile.getName(); + String localFileName = localDirectory.getFile().getPath() + "/" + remoteFileName; + File localFile = new File(localFileName); + if (!localFile.exists()) { + String tempFileName = localFileName + AbstractInboundRemoteFileSystemSynchronizingMessageSource.INCOMPLETE_EXTENSION; + File file = new File(tempFileName); + FileOutputStream fos = new FileOutputStream(file); + try { + client.retrieveFile(remoteFileName, fos); + // Perhaps we have some dispatch of the source file to do? + acknowledge(client, ftpFile); + } + catch (Exception e) { + if (e instanceof RuntimeException){ + throw (RuntimeException) e; + } + else { + throw new MessagingException("Failed to copy file", e); + } + } + finally { + fos.close(); + } + file.renameTo(localFile); + return true; + } + return false; + } + /** * An acknowledgment strategy that deletes the file. */ - private class DeletionEntryAcknowledgmentStrategy implements AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy { + private static class DeletionEntryAcknowledgmentStrategy implements AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy { + + private final Log logger = LogFactory.getLog(this.getClass()); public void acknowledge(Object useful, FTPFile fptFile) throws Exception { FTPClient ftpClient = (FTPClient) useful; diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java index 2be02b4311..63eb36730a 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundSynchronizer.java @@ -47,7 +47,7 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych * the path on the remote mount */ private volatile String remotePath; - + private volatile boolean autoCreateDirectories; /** @@ -116,39 +116,6 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych return false; } - private boolean copyFromRemoteToLocalDirectory(SftpSession sftpSession, ChannelSftp.LsEntry entry, Resource localDir) throws Exception { - File fileForLocalDir = localDir.getFile(); - File localFile = new File(fileForLocalDir, entry.getFilename()); - if (!localFile.exists()) { - InputStream in = null; - FileOutputStream fileOutputStream = null; - try { - File tmpLocalTarget = new File(localFile.getAbsolutePath() + - AbstractInboundRemoteFileSystemSynchronizingMessageSource.INCOMPLETE_EXTENSION); - fileOutputStream = new FileOutputStream(tmpLocalTarget); - String remoteFqPath = this.remotePath + "/" + entry.getFilename(); - in = sftpSession.getChannel().get(remoteFqPath); - try { - IOUtils.copy(in, fileOutputStream); - } - finally { - IOUtils.closeQuietly(in); - IOUtils.closeQuietly(fileOutputStream); - } - if (tmpLocalTarget.renameTo(localFile) && this.entryAcknowledgmentStrategy != null) { - this.acknowledge(sftpSession, entry); - } - return true; - } - catch (Throwable th) { - throw new MessagingException("Failure occurred while copying from remote to local directory", th); - } - } - else { - return true; - } - } - @Override @SuppressWarnings("unchecked") protected void syncRemoteToLocalFileSystem() { @@ -178,8 +145,47 @@ public class SftpInboundSynchronizer extends AbstractInboundRemoteFileSystemSych } } + private boolean copyFromRemoteToLocalDirectory(SftpSession sftpSession, ChannelSftp.LsEntry entry, Resource localDir) throws Exception { + File fileForLocalDir = localDir.getFile(); + File localFile = new File(fileForLocalDir, entry.getFilename()); + if (!localFile.exists()) { + InputStream in = null; + FileOutputStream fileOutputStream = null; + try { + File tmpLocalTarget = new File(localFile.getAbsolutePath() + + AbstractInboundRemoteFileSystemSynchronizingMessageSource.INCOMPLETE_EXTENSION); + fileOutputStream = new FileOutputStream(tmpLocalTarget); + String remoteFqPath = this.remotePath + "/" + entry.getFilename(); + in = sftpSession.getChannel().get(remoteFqPath); + try { + IOUtils.copy(in, fileOutputStream); + } + finally { + IOUtils.closeQuietly(in); + IOUtils.closeQuietly(fileOutputStream); + } + if (tmpLocalTarget.renameTo(localFile) && this.entryAcknowledgmentStrategy != null) { + this.acknowledge(sftpSession, entry); + } + return true; + } + catch (Exception e) { + if (e instanceof RuntimeException){ + throw (RuntimeException) e; + } + else { + throw new MessagingException("Failure occurred while copying from remote to local directory", e); + } + } + } + else { + return true; + } + } + + private class DeletionEntryAcknowledgmentStrategy implements AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy { - + public void acknowledge(Object useful, ChannelSftp.LsEntry msg) throws Exception { SftpSession sftpSession = (SftpSession) useful; String remoteFqPath = remotePath + "/" + msg.getFilename();