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 9a078003d2..8e24d72d87 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 @@ -16,6 +16,8 @@ package org.springframework.integration.file.remote.synchronizer; +import java.io.File; +import java.io.IOException; import java.util.Arrays; import java.util.List; @@ -23,7 +25,11 @@ import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.springframework.beans.factory.InitializingBean; +import org.springframework.integration.MessagingException; import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.remote.session.Session; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.util.Assert; /** * Base class charged with knowing how to connect to a remote file system, @@ -41,6 +47,11 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS protected final Log logger = LogFactory.getLog(this.getClass()); + /** + * the {@link SessionFactory} for acquiring remote file Sessions. + */ + private volatile SessionFactory sessionFactory; + /** * An {@link FileListFilter} that runs against the remote file system view. */ @@ -53,6 +64,15 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS protected boolean shouldDeleteSourceFile; + /** + * Create a synchronizer with the {@link SessionFactory} used to acquire {@link Session} instances. + */ + public AbstractInboundFileSynchronizer(SessionFactory sessionFactory) { + Assert.notNull(sessionFactory, "sessionFactory must not be null"); + this.sessionFactory = sessionFactory; + } + + public void setFilter(FileListFilter filter) { this.filter = filter; } @@ -61,8 +81,38 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS this.shouldDeleteSourceFile = shouldDeleteSourceFile; } + public void afterPropertiesSet() { + Assert.notNull(this.sessionFactory, "sessionFactory must not be null"); + } + protected final List filterFiles(F[] files) { return (this.filter != null) ? this.filter.filterFiles(files) : Arrays.asList(files); } + public void synchronizeToLocalDirectory(File localDirectory) { + Session session = null; + try { + session = this.sessionFactory.getSession(); + Assert.state(session != null, "failed to acquire a Session"); + this.synchronizeToLocalDirectory(localDirectory, session); + } + catch (IOException e) { + throw new MessagingException("Problem occurred while synchronizing remote to local directory", e); + } + finally { + if (session != null) { + try { + session.close(); + } + catch (Exception ignored) { + if (logger.isDebugEnabled()) { + logger.debug("failed to close Session", ignored); + } + } + } + } + } + + protected abstract void synchronizeToLocalDirectory(File localDirectory, Session session) throws IOException; + } 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 3afc2ed6b1..335d260019 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 @@ -30,7 +30,6 @@ import org.springframework.integration.file.remote.session.Session; import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizer; import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizingMessageSource; -import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; import org.springframework.util.FileCopyUtils; @@ -44,15 +43,12 @@ public class FtpInboundFileSynchronizer extends AbstractInboundFileSynchronizer< private volatile String remotePath; - private volatile SessionFactory sessionFactory; - /** * Create a synchronizer with the {@link SessionFactory} used to acquire {@link Session} instances. */ public FtpInboundFileSynchronizer(SessionFactory sessionFactory) { - Assert.notNull(sessionFactory, "sessionFactory must not be null"); - this.sessionFactory = sessionFactory; + super(sessionFactory); } @@ -60,34 +56,14 @@ public class FtpInboundFileSynchronizer extends AbstractInboundFileSynchronizer< this.remotePath = remotePath; } - public void afterPropertiesSet() { - Assert.notNull(this.sessionFactory, "sessionFactory must not be null"); - } - - public void synchronizeToLocalDirectory(File localDirectory) { - Session session = null; - try { - session = this.sessionFactory.getSession(); - Assert.state(session != null, "failed to acquire an FTP Session"); - Collection files = session.ls(this.remotePath); - if (!CollectionUtils.isEmpty(files)) { - Collection filteredFiles = this.filterFiles(files.toArray(new FTPFile[]{})); - for (FTPFile ftpFile : filteredFiles) { - if ((ftpFile != null) && ftpFile.isFile()) { - copyFileToLocalDirectory(session, ftpFile, localDirectory); - } - } - } - } - catch (IOException e) { - throw new MessagingException("Problem occurred while synchronizing remote to local directory", e); - } - finally { - if (session != null) { - try { - session.close(); - } - catch (Exception ignored) { + @Override + protected void synchronizeToLocalDirectory(File localDirectory, Session session) throws IOException { + Collection files = session.ls(this.remotePath); + if (!CollectionUtils.isEmpty(files)) { + Collection filteredFiles = this.filterFiles(files.toArray(new FTPFile[]{})); + for (FTPFile ftpFile : filteredFiles) { + if ((ftpFile != null) && ftpFile.isFile()) { + copyFileToLocalDirectory(session, ftpFile, localDirectory); } } } 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 750bc33b35..c3ec29b4d5 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 @@ -18,6 +18,7 @@ package org.springframework.integration.sftp.inbound; import java.io.File; import java.io.FileOutputStream; +import java.io.IOException; import java.io.InputStream; import java.util.Collection; @@ -45,15 +46,9 @@ public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer */ private volatile String remotePath; - /** - * the {@link SessionFactory} for acquiring SFTP Sessions. - */ - private final SessionFactory sessionFactory; - public SftpInboundFileSynchronizer(SessionFactory sessionFactory) { - Assert.notNull(sessionFactory, "sessionFactory must not be null"); - this.sessionFactory = sessionFactory; + super(sessionFactory); } @@ -61,46 +56,25 @@ public class SftpInboundFileSynchronizer extends AbstractInboundFileSynchronizer this.remotePath = remotePath; } - public void afterPropertiesSet() throws Exception { + public void afterPropertiesSet() { + super.afterPropertiesSet(); Assert.notNull(this.remotePath, "'remotePath' must not be null"); } - public void synchronizeToLocalDirectory(File localDirectory) { - Session session = null; - try { - session = this.sessionFactory.getSession(); - if (logger.isTraceEnabled()) { - logger.trace("Pooled SftpSession " + session + " from the pool"); - } - Collection beforeFilter = session.ls(remotePath); - ChannelSftp.LsEntry[] entries = (beforeFilter == null) ? new ChannelSftp.LsEntry[0] : + @Override + protected void synchronizeToLocalDirectory(File localDirectory, Session session) throws IOException { + Collection beforeFilter = session.ls(remotePath); + ChannelSftp.LsEntry[] entries = (beforeFilter == null) ? new ChannelSftp.LsEntry[0] : beforeFilter.toArray(new ChannelSftp.LsEntry[beforeFilter.size()]); - Collection files = this.filterFiles(entries); - for (ChannelSftp.LsEntry lsEntry : files) { - if ((lsEntry != null) && !lsEntry.getAttrs().isDir() && !lsEntry.getAttrs().isLink()) { - copyFromRemoteToLocalDirectory(session, lsEntry, localDirectory); - } - } - } - catch (Exception e) { - throw new MessagingException("couldn't synchronize remote to local directory", e); - } - finally { - try { - session.close(); - } - catch (Exception ignored) { - if (logger.isDebugEnabled()) { - logger.debug("failed to close Session", ignored); - } - } - if (logger.isTraceEnabled()) { - logger.trace("Putting SftpSession " + session + " back into the pool"); + Collection files = this.filterFiles(entries); + for (ChannelSftp.LsEntry lsEntry : files) { + if ((lsEntry != null) && !lsEntry.getAttrs().isDir() && !lsEntry.getAttrs().isLink()) { + copyFromRemoteToLocalDirectory(session, lsEntry, localDirectory); } } } - private boolean copyFromRemoteToLocalDirectory(Session session, ChannelSftp.LsEntry entry, File localDirectory) throws Exception { + private boolean copyFromRemoteToLocalDirectory(Session session, ChannelSftp.LsEntry entry, File localDirectory) throws IOException { File localFile = new File(localDirectory, entry.getFilename()); if (!localFile.exists()) { InputStream in = null;