INT-1614 refactoring as much as possible into base class
This commit is contained in:
@@ -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<F> 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 <emphasis>remote</emphasis> file system view.
|
||||
*/
|
||||
@@ -53,6 +64,15 @@ public abstract class AbstractInboundFileSynchronizer<F> 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<F> filter) {
|
||||
this.filter = filter;
|
||||
}
|
||||
@@ -61,8 +81,38 @@ public abstract class AbstractInboundFileSynchronizer<F> implements InboundFileS
|
||||
this.shouldDeleteSourceFile = shouldDeleteSourceFile;
|
||||
}
|
||||
|
||||
public void afterPropertiesSet() {
|
||||
Assert.notNull(this.sessionFactory, "sessionFactory must not be null");
|
||||
}
|
||||
|
||||
protected final List<F> 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;
|
||||
|
||||
}
|
||||
|
||||
@@ -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<FTPFile> files = session.ls(this.remotePath);
|
||||
if (!CollectionUtils.isEmpty(files)) {
|
||||
Collection<FTPFile> 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<FTPFile> files = session.ls(this.remotePath);
|
||||
if (!CollectionUtils.isEmpty(files)) {
|
||||
Collection<FTPFile> filteredFiles = this.filterFiles(files.toArray(new FTPFile[]{}));
|
||||
for (FTPFile ftpFile : filteredFiles) {
|
||||
if ((ftpFile != null) && ftpFile.isFile()) {
|
||||
copyFileToLocalDirectory(session, ftpFile, localDirectory);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<ChannelSftp.LsEntry> beforeFilter = session.ls(remotePath);
|
||||
ChannelSftp.LsEntry[] entries = (beforeFilter == null) ? new ChannelSftp.LsEntry[0] :
|
||||
@Override
|
||||
protected void synchronizeToLocalDirectory(File localDirectory, Session session) throws IOException {
|
||||
Collection<ChannelSftp.LsEntry> beforeFilter = session.ls(remotePath);
|
||||
ChannelSftp.LsEntry[] entries = (beforeFilter == null) ? new ChannelSftp.LsEntry[0] :
|
||||
beforeFilter.toArray(new ChannelSftp.LsEntry[beforeFilter.size()]);
|
||||
Collection<ChannelSftp.LsEntry> 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<ChannelSftp.LsEntry> 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;
|
||||
|
||||
Reference in New Issue
Block a user