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 f0298e03ad..9a078003d2 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 @@ -24,7 +24,6 @@ import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.file.filters.FileListFilter; -import org.springframework.integration.file.remote.session.Session; /** * Base class charged with knowing how to connect to a remote file system, @@ -35,89 +34,35 @@ import org.springframework.integration.file.remote.session.Session; * ensure the file entry is acceptable. * * @author Josh Long + * @author Mark Fisher + * @since 2.0 */ public abstract class AbstractInboundFileSynchronizer implements InboundFileSynchronizer, InitializingBean { protected final Log logger = LogFactory.getLog(this.getClass()); - - /** - * Should we delete the source file? For an FTP - * server, for example, this would delete the original FTPFile instance. - */ - protected boolean shouldDeleteSourceFile; - /** * An {@link FileListFilter} that runs against the remote file system view. */ private volatile FileListFilter filter; /** - * The {@link AcknowledgmentStrategy} implementation. + * Should we delete the source file? For an FTP + * server, for example, this would delete the original FTPFile instance. */ - private AcknowledgmentStrategy acknowledgmentStrategy; + protected boolean shouldDeleteSourceFile; public void setFilter(FileListFilter filter) { this.filter = filter; } - public void setAcknowledgmentStrategy(AcknowledgmentStrategy acknowledgmentStrategy) { - this.acknowledgmentStrategy = acknowledgmentStrategy; - } - public void setShouldDeleteSourceFile(boolean shouldDeleteSourceFile) { this.shouldDeleteSourceFile = shouldDeleteSourceFile; } - /** - * @param session - * session that was used to retrieve the file. Will be passed to the {@link EntryAcknowledgmentStrategy}. - * The {@link EntryAcknowledgmentStrategy#acknowledge(Object, Object)} will be called in line with the - * {@link org.springframework.integration.core.MessageSource#receive()} call so this could conceivably - * be a 'live' stateful client (a connection?) that is inappropriate to cache as it has per-request state. - * @param file - * leverages strategy implementations to enable different - * 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 Exception - * escape hatch exception, let the adapter deal with it. - */ - protected final void acknowledge(Session session, F file) throws Exception { - if (this.acknowledgmentStrategy != null) { - this.acknowledgmentStrategy.acknowledge(session, file); - } - } - protected final List filterFiles(F[] files) { return (this.filter != null) ? this.filter.filterFiles(files) : Arrays.asList(files); } - - /** - * Strategy interface to expose a hook for dispatching, moving, or deleting - * the file once it has been delivered. Adapters should (for consistency) - * expose an attribute dictating whether the adapter will delete the - * source entry on the remote file system. This is the - * file-system version of an ack-mode. - * - * @param the file entry type (file, sftp, ftp, ...) - */ - public static interface AcknowledgmentStrategy { - - /** - * Semantics are simple. You get a pointer to the file just processed - * and the FTP Session that processed it. - * - * @param session - * the FTP session - * @param file - * the file that has been processed - * @throws Exception in case of an error while acknowledging - */ - void acknowledge(Session session, F file) throws Exception; - - } - } diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java index e671718eb7..2a872a5be2 100644 --- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java +++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/inbound/FtpInboundFileSynchronizer.java @@ -23,8 +23,6 @@ import java.io.IOException; import java.io.InputStream; import java.util.Collection; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; import org.apache.commons.net.ftp.FTPFile; import org.springframework.integration.MessagingException; @@ -64,9 +62,6 @@ public class FtpInboundFileSynchronizer extends AbstractInboundFileSynchronizer< public void afterPropertiesSet() { Assert.notNull(this.sessionFactory, "sessionFactory must not be null"); - if (this.shouldDeleteSourceFile) { - this.setAcknowledgmentStrategy(new DeletionAcknowledgmentStrategy()); - } } public void synchronizeToLocalDirectory(File localDirectory) { @@ -114,7 +109,9 @@ public class FtpInboundFileSynchronizer extends AbstractInboundFileSynchronizer< return false; } FileCopyUtils.copy(inputStream, fileOutputStream); - acknowledge(session, ftpFile); + if (this.shouldDeleteSourceFile) { + this.deleteRemoteFile(session, ftpFile); + } } catch (Exception e) { if (e instanceof RuntimeException){ @@ -133,19 +130,11 @@ public class FtpInboundFileSynchronizer extends AbstractInboundFileSynchronizer< return false; } - - /** - * An acknowledgment strategy that deletes the file. - */ - private static class DeletionAcknowledgmentStrategy implements AcknowledgmentStrategy { - - private final Log logger = LogFactory.getLog(this.getClass()); - - public void acknowledge(Session session, FTPFile ftpFile) throws Exception { - if ((ftpFile != null) && session.rm(ftpFile.getName())) { - if (logger.isDebugEnabled()) { - logger.debug("deleted " + ftpFile.getName()); - } + // TODO: make this an abstract method in the base class once the code that calls this is refactored upward + private void deleteRemoteFile(Session session, FTPFile ftpFile) { + if ((ftpFile != null) && session.rm(ftpFile.getName())) { + if (logger.isDebugEnabled()) { + logger.debug("deleted " + ftpFile.getName()); } } } diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java index 3620f3ecde..750bc33b35 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/inbound/SftpInboundFileSynchronizer.java @@ -63,9 +63,6 @@ public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer public void afterPropertiesSet() throws Exception { Assert.notNull(this.remotePath, "'remotePath' must not be null"); - if (this.shouldDeleteSourceFile) { - this.setAcknowledgmentStrategy(new DeletionAcknowledgmentStrategy()); - } } public void synchronizeToLocalDirectory(File localDirectory) { @@ -122,7 +119,9 @@ public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer fileOutputStream.close(); } if (tmpLocalTarget.renameTo(localFile)) { - this.acknowledge(session, entry); + if (this.shouldDeleteSourceFile) { + this.deleteRemoteFile(session, entry); + } } return true; } @@ -140,15 +139,11 @@ public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer } } - - private class DeletionAcknowledgmentStrategy implements AcknowledgmentStrategy { - - public void acknowledge(Session session, ChannelSftp.LsEntry msg) throws Exception { - String remoteFqPath = remotePath + "/" + msg.getFilename(); - session.rm(remoteFqPath); - if (logger.isDebugEnabled()) { - logger.debug("deleted " + msg.getFilename()); - } + private void deleteRemoteFile(Session session, ChannelSftp.LsEntry msg) { + String remoteFqPath = remotePath + "/" + msg.getFilename(); + session.rm(remoteFqPath); + if (logger.isDebugEnabled()) { + logger.debug("deleted " + msg.getFilename()); } } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizerTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizerTests.java index d9fce38e98..87faec5356 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizerTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizerTests.java @@ -29,7 +29,6 @@ import org.springframework.core.io.FileSystemResource; import org.springframework.core.io.Resource; import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; -import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer.AcknowledgmentStrategy; import org.springframework.integration.sftp.inbound.SftpInboundFileSynchronizer; import org.springframework.util.ReflectionUtils; @@ -88,7 +87,6 @@ public class SftpInboundRemoteFileSystemSynchronizerTests { @Test public void testCopyAndRenameWhenLocalFileDoesntExist() throws Exception { SftpInboundFileSynchronizer synchronizer = new SftpInboundFileSynchronizer(mock(SessionFactory.class)); - synchronizer.setAcknowledgmentStrategy(mock(AcknowledgmentStrategy.class)); Method method = ReflectionUtils.findMethod(synchronizer.getClass(), "copyFromRemoteToLocalDirectory", Session.class, LsEntry.class, Resource.class); method.setAccessible(true);