Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

initial commit of my work to make building filesystem based adapters more consistent, cleaner.

Browse Source
This commit is contained in:
Josh Long
2010-08-20 10:06:13 +00:00
parent 59e4b98d39
commit 25b5afa528
151 changed files with 3351 additions and 1767 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/AbstractSftpFileListFilter.java
View File

@@ -22,10 +22,10 @@ import java.util.List;
/**
* Convenience implementation patterned off {@link org.springframework.integration.file.FileListFilter}
* Convenience implementation patterned off {@link org.springframework.integration.file.filters.FileListFilter}
*
* @author Josh Long
*/
*/ @Deprecated
public abstract class AbstractSftpFileListFilter implements SftpFileListFilter {
abstract public boolean accept(ChannelSftp.LsEntry lsEntry);

4
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/CompositeFtpFileListFilter.java
View File

@@ -22,10 +22,10 @@ import java.util.*;
/**
* Patterned very much on the {@link org.springframework.integration.file.CompositeFileListFilter}
* Patterned very much on the {@link org.springframework.integration.file.filters.CompositeFileListFilter}
*
* @author Josh Long
*/
*/ @Deprecated
public class CompositeFtpFileListFilter implements SftpFileListFilter {
private Set<SftpFileListFilter> filters;

4
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/PatternMatchingSftpFileListFilter.java
View File

@@ -28,10 +28,10 @@ import java.util.regex.Pattern;
/**
* Validates {@link com.jcraft.jsch.ChannelSftp.LsEntry}s against a {@link java.util.regex.Pattern}.
* Patterned very much like {@link org.springframework.integration.file.PatternMatchingFileListFilter}.
* Patterned very much like {@link org.springframework.integration.file.filters.PatternMatchingFileListFilter}.
*
* @author Josh Long
*/
*/ @Deprecated
public class PatternMatchingSftpFileListFilter extends AbstractSftpFileListFilter implements InitializingBean {
private Log logger = LogFactory.getLog(getClass());

16
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpEntryNamer.java Normal file
View File

@@ -0,0 +1,16 @@
package org.springframework.integration.sftp;
import com.jcraft.jsch.ChannelSftp;
import org.springframework.integration.file.entries.EntryNamer;
/**
* Knows how to name a {@link com.jcraft.jsch.ChannelSftp.LsEntry} instance
*
* @author Josh Long
*/
public class SftpEntryNamer implements EntryNamer<ChannelSftp.LsEntry>{
public String nameOf(ChannelSftp.LsEntry entry) {
return entry.getFilename() ;
}
}

2
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpFileListFilter.java
View File

@@ -24,7 +24,7 @@ import java.util.List;
* and returns the balance. These are then sync'd to the local directory.
*
* @author Josh Long
*/
*/ @Deprecated
public interface SftpFileListFilter {
List<ChannelSftp.LsEntry> filterFiles (ChannelSftp.LsEntry [] files);
}

2
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpInboundSynchronizer.java
View File

@@ -44,7 +44,7 @@ import java.util.concurrent.ScheduledFuture;
*
* @author Josh Long
* @author Mario Gray
*/
*/ @Deprecated
public class SftpInboundSynchronizer implements InitializingBean {
private static final long DEFAULT_REFRESH_RATE = 10 * 1000; // 10 seconds

26
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SftpMessageSource.java
View File

@@ -13,23 +13,28 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.sftp;
import com.jcraft.jsch.ChannelSftp;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.Lifecycle;
import org.springframework.core.io.Resource;
import org.springframework.integration.Message;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.file.AcceptOnceFileListFilter;
import org.springframework.integration.file.CompositeFileListFilter;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.integration.file.PatternMatchingFileListFilter;
import org.springframework.integration.file.entries.CompositeEntryListFilter;
import org.springframework.integration.file.entries.PatternMatchingEntryListFilter;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.Trigger;
import java.io.File;
import java.io.IOException;
import java.util.regex.Pattern;
@@ -46,13 +51,21 @@ public class SftpMessageSource implements MessageSource<File>, InitializingBean,
private SftpInboundSynchronizer synchronizer;
private TaskScheduler taskScheduler;
private Trigger trigger;
private SftpEntryNamer lsEntryEntryNamer = new SftpEntryNamer();
public SftpMessageSource(FileReadingMessageSource fileSource, SftpInboundSynchronizer synchronizer) {
this.fileReadingMessageSource = fileSource;
this.synchronizer = synchronizer;
Pattern completePattern = Pattern.compile("^.*(?<!" + SftpInboundSynchronizer.INCOMPLETE_EXTENSION + ")$");
fileReadingMessageSource.setFilter(new CompositeFileListFilter(new AcceptOnceFileListFilter(), new PatternMatchingFileListFilter(completePattern)));
PatternMatchingEntryListFilter<ChannelSftp.LsEntry> filePatternMatchingEntryListFilter = new PatternMatchingEntryListFilter<ChannelSftp.LsEntry>(lsEntryEntryNamer, completePattern);
PatternMatchingEntryListFilter<ChannelSftp.LsEntry> lsEntryPatternMatchingEntryListFilter = new PatternMatchingEntryListFilter<ChannelSftp.LsEntry>(this.lsEntryEntryNamer, completePattern);
CompositeEntryListFilter<ChannelSftp.LsEntry> fileCompositeEntryListFilter = new CompositeEntryListFilter<ChannelSftp.LsEntry>(filePatternMatchingEntryListFilter,
lsEntryPatternMatchingEntryListFilter);
// todo this.fileReadingMessageSource.setFilter( fileCompositeEntryListFilter);
//fileReadingMessageSource.setFilter( (completePattern)));
}
public void afterPropertiesSet() throws Exception {
@@ -94,11 +107,12 @@ public class SftpMessageSource implements MessageSource<File>, InitializingBean,
public void setLocalDirectory(final Resource localDirectory) {
this.localDirectory = localDirectory;
try {
this.fileReadingMessageSource.setDirectory(localDirectory.getFile());
} catch (IOException e) {
}
this.synchronizer.setLocalDirectory(localDirectory);
}

5
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SftpNamespaceHandler.java
View File

@@ -23,6 +23,8 @@ import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.sftp.impl.SftpInboundRemoteFileSystemSynchronizingMessageSource;
import org.springframework.integration.sftp.impl.SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean;
import org.w3c.dom.Element;
@@ -63,7 +65,8 @@ public class SftpNamespaceHandler extends NamespaceHandlerSupport {
private static class SFTPMessageSourceBeanDefinitionParser extends AbstractPollingInboundChannelAdapterParser {
@Override
protected String parseSource(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( SftpMessageSourceFactoryBean.class.getName());
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(
SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean.class.getName());
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "filter");

141
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizer.java Normal file
View File

@@ -0,0 +1,141 @@
package org.springframework.integration.sftp.impl;
import com.jcraft.jsch.ChannelSftp;
import org.apache.commons.io.IOUtils;
import org.springframework.beans.factory.annotation.Required;
import org.springframework.core.io.Resource;
import org.springframework.integration.MessagingException;
import org.springframework.integration.file.AbstractInboundRemoteFileSystemSychronizer;
import org.springframework.integration.file.AbstractInboundRemoteFileSystemSynchronizingMessageSource;
import org.springframework.integration.sftp.SftpSession;
import org.springframework.integration.sftp.SftpSessionPool;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.Assert;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.Collection;
/**
* This handles the synchronization between a remote SFTP endpoint and a local mount
*
* @author Josh Long
*/
public class SftpInboundRemoteFileSystemSynchronizer extends AbstractInboundRemoteFileSystemSychronizer<ChannelSftp.LsEntry> {
/**
* the path on te remote mount
*/
private volatile String remotePath;
/**
* the pool of {@link org.springframework.integration.sftp.SftpSessionPool} SFTP sessions
*/
private volatile SftpSessionPool clientPool;
public void setRemotePath(String remotePath) {
this.remotePath = remotePath;
}
@Override
protected void onInit() throws Exception {
Assert.notNull(this.clientPool, "clientPool can't be null");
if (this.shouldDeleteSourceFile) {
this.entryAcknowledgmentStrategy = new DeletionEntryAcknowledgmentStrategy();
}
}
@Required
public void setClientPool(SftpSessionPool clientPool) {
this.clientPool = clientPool;
}
@SuppressWarnings("ignored")
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);
IOUtils.copy(in, fileOutputStream);
if (tmpLocalTarget.renameTo(localFile)) {
// last step
this.acknowledge(sftpSession, entry);
}
return true;
} catch (Throwable th) {
IOUtils.closeQuietly(in);
IOUtils.closeQuietly(fileOutputStream);
}
} else {
return true;
}
return false;
}
@Override
@SuppressWarnings("unchecked")
protected void syncRemoteToLocalFileSystem() throws Exception {
SftpSession session = null;
try {
session = clientPool.getSession();
session.start();
ChannelSftp channelSftp = session.getChannel();
Collection<ChannelSftp.LsEntry> beforeFilter = channelSftp.ls(remotePath);
ChannelSftp.LsEntry[] entries = (beforeFilter == null) ? new ChannelSftp.LsEntry[0] : beforeFilter.toArray(new ChannelSftp.LsEntry[beforeFilter.size()]);
Collection<ChannelSftp.LsEntry> files = this.filter.filterEntries(entries);
for (ChannelSftp.LsEntry lsEntry : files) {
if ((lsEntry != null) && !lsEntry.getAttrs().isDir() && !lsEntry.getAttrs().isLink()) {
copyFromRemoteToLocalDirectory(session, lsEntry, this.localDirectory);
}
}
} catch (IOException e) {
throw new MessagingException("couldn't synchronize remote to local directory", e);
} finally {
if ((session != null) && (clientPool != null)) {
clientPool.release(session);
}
}
}
@Override
protected Trigger getTrigger() {
return new PeriodicTrigger(10 * 1000);
}
class DeletionEntryAcknowledgmentStrategy implements AbstractInboundRemoteFileSystemSychronizer.EntryAcknowledgmentStrategy<ChannelSftp.LsEntry> {
public void acknowledge(Object useful, ChannelSftp.LsEntry msg)
throws Exception {
SftpSession sftpSession = (SftpSession) useful;
String remoteFqPath = remotePath + "/" + msg.getFilename();
sftpSession.getChannel().rm(remoteFqPath);
if (logger.isDebugEnabled()) {
logger.debug("deleted " + msg.getFilename());
}
}
}
}

101
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/impl/SftpInboundRemoteFileSystemSynchronizingMessageSource.java Normal file
View File

@@ -0,0 +1,101 @@
package org.springframework.integration.sftp.impl;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.SftpATTRS;
import org.springframework.integration.file.AbstractInboundRemoteFileSystemSynchronizingMessageSource;
import org.springframework.integration.sftp.SftpSession;
import org.springframework.integration.sftp.SftpSessionPool;
import org.springframework.util.Assert;
/**
* a {@link org.springframework.integration.core.MessageSource} implementation for SFTP
*
* @author Josh Long
*/
public class SftpInboundRemoteFileSystemSynchronizingMessageSource extends AbstractInboundRemoteFileSystemSynchronizingMessageSource<ChannelSftp.LsEntry, SftpInboundRemoteFileSystemSynchronizer> {
/**
* the pool of sessions
*/
private volatile SftpSessionPool clientPool;
/**
* the remote path on teh server
*/
private volatile String remotePath;
public void setClientPool(SftpSessionPool clientPool) {
this.clientPool = clientPool;
}
public void setRemotePath(String remotePath) {
this.remotePath = remotePath;
}
@Override
protected void doStart() {
this.synchronizer.start();
}
@Override
protected void doStop() {
this.synchronizer.stop();
}
/**
* there be dragons this way ... This method will check to ensure that the remote directory exists. If the directory
* doesnt exist, and autoCreatePath is 'true,' then this method makes a few reasonably sane attempts
* to create it. Otherwise, it fails fast.
*
* @param remotePath the path on the remote SSH / SFTP server to create.
* @return whether or not the directory is there (regardless of whether we created it in this method or it already
* existed.)
*/
private boolean checkThatRemotePathExists(String remotePath) {
SftpSession session = null;
ChannelSftp channelSftp = null;
try {
session = this.clientPool.getSession();
Assert.state(session != null, "session as returned from the pool should not be null. " + "If it is, it is most likely an error in the pool implementation. ");
session.start();
channelSftp = session.getChannel();
SftpATTRS attrs = channelSftp.stat(remotePath);
assert (attrs != null) && attrs.isDir() : "attrs can't be null, and should indicate that it's a directory!";
return true;
} catch (Throwable th) {
if (this.autoCreateDirectories && (this.clientPool != null) && (session != null)) {
try {
if (channelSftp != null) {
channelSftp.mkdir(remotePath);
if (channelSftp.stat(remotePath).isDir()) {
return true;
}
}
} catch (Throwable t) {
return false;
}
}
} finally {
if ((clientPool != null) && (session != null)) {
clientPool.release(session);
}
}
return false;
}
@Override
protected void onInit() throws Exception {
super.onInit();
this.checkThatRemotePathExists(this.remotePath);
this.synchronizer.setClientPool(this.clientPool);
}
}

195
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/impl/SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean.java Normal file
View File

@@ -0,0 +1,195 @@
package org.springframework.integration.sftp.impl;
import com.jcraft.jsch.ChannelSftp;
import org.apache.commons.lang.SystemUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.AbstractFactoryBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.ResourceLoaderAware;
import org.springframework.core.io.Resource;
import org.springframework.core.io.ResourceEditor;
import org.springframework.core.io.ResourceLoader;
import org.springframework.integration.file.FileReadingMessageSource;
import org.springframework.integration.file.entries.CompositeEntryListFilter;
import org.springframework.integration.file.entries.EntryListFilter;
import org.springframework.integration.file.entries.PatternMatchingEntryListFilter;
import org.springframework.integration.sftp.*;
import org.springframework.integration.sftp.config.SftpSessionUtils;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.ErrorHandler;
import org.springframework.util.StringUtils;
import java.io.File;
import java.util.Map;
/**
* a factory bean to hide the fairly complex configuration possibilities for an SFTP endpoint
*
* @author Josh Long
*/
public class SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean extends
AbstractFactoryBean<SftpInboundRemoteFileSystemSynchronizingMessageSource> implements ResourceLoaderAware {
/**
* injected by the container
*/
private volatile ResourceLoader resourceLoader;
private volatile Resource localDirectoryResource;
private volatile String localDirectoryPath;
private volatile String autoCreateDirectories;
private volatile String autoDeleteRemoteFilesOnSync;
private volatile String filenamePattern;
private volatile EntryListFilter<ChannelSftp.LsEntry> filter;
private int port = 22;
public void setLocalDirectoryResource(Resource localDirectoryResource) {
this.localDirectoryResource = localDirectoryResource;
}
public void setLocalDirectoryPath(String localDirectoryPath) {
this.localDirectoryPath = localDirectoryPath;
}
public void setAutoCreateDirectories(String autoCreateDirectories) {
this.autoCreateDirectories = autoCreateDirectories;
}
public void setAutoDeleteRemoteFilesOnSync(String autoDeleteRemoteFilesOnSync) {
this.autoDeleteRemoteFilesOnSync = autoDeleteRemoteFilesOnSync;
}
public void setFilenamePattern(String filenamePattern) {
this.filenamePattern = filenamePattern;
}
public void setFilter(EntryListFilter<ChannelSftp.LsEntry> filter) {
this.filter = filter;
}
public void setPort(int port) {
this.port = port;
}
public void setHost(String host) {
this.host = host;
}
public void setKeyFile(String keyFile) {
this.keyFile = keyFile;
}
public void setKeyFilePassword(String keyFilePassword) {
this.keyFilePassword = keyFilePassword;
}
public void setLocalWorkingDirectory(String localWorkingDirectory) {
this.localWorkingDirectory = localWorkingDirectory;
}
public void setPassword(String password) {
this.password = password;
}
public void setRemoteDirectory(String remoteDirectory) {
this.remoteDirectory = remoteDirectory;
}
public void setUsername(String username) {
this.username = username;
}
private String host;
private String keyFile;
private String keyFilePassword;
private String localWorkingDirectory;
private String password;
private String remoteDirectory;
private String username;
public void setResourceLoader(ResourceLoader resourceLoader) {
this.resourceLoader = resourceLoader;
}
@Override
public Class<?> getObjectType() {
return SftpRemoteFileSystemSynchronizingMessageSourceFactoryBean.class;
}
@Override
protected SftpInboundRemoteFileSystemSynchronizingMessageSource createInstance()
throws Exception {
boolean autoCreatDirs = Boolean.parseBoolean(this.autoCreateDirectories);
boolean ackRemoteDir = Boolean.parseBoolean(this.autoDeleteRemoteFilesOnSync);
SftpInboundRemoteFileSystemSynchronizingMessageSource sftpMsgSrc = new SftpInboundRemoteFileSystemSynchronizingMessageSource();
sftpMsgSrc.setAutoCreateDirectories(autoCreatDirs);
// local directories
if ((this.localDirectoryResource == null) || !StringUtils.hasText(this.localDirectoryPath)) {
File tmp = SystemUtils.getJavaIoTmpDir();
File sftpTmp = new File(tmp, "sftpInbound");
this.localDirectoryPath = "file://" + sftpTmp.getAbsolutePath();
}
this.localDirectoryResource = this.fromText(localDirectoryPath);
// remote predicates
SftpEntryNamer sftpEntryNamer = new SftpEntryNamer();
CompositeEntryListFilter<ChannelSftp.LsEntry> compositeFtpFileListFilter = new CompositeEntryListFilter<ChannelSftp.LsEntry>();
if (StringUtils.hasText(this.filenamePattern)) {
PatternMatchingEntryListFilter<ChannelSftp.LsEntry> ftpFilePatternMatchingEntryListFilter = new PatternMatchingEntryListFilter<ChannelSftp.LsEntry>(sftpEntryNamer, filenamePattern);
compositeFtpFileListFilter.addFilter(ftpFilePatternMatchingEntryListFilter);
}
if (this.filter != null) {
compositeFtpFileListFilter.addFilter(this.filter);
}
this.filter = compositeFtpFileListFilter;
// pools
SftpSessionFactory sessionFactory = SftpSessionUtils.buildSftpSessionFactory(
this.host, this.password, this.username, this.keyFile, this.keyFilePassword, this.port);
QueuedSftpSessionPool pool = new QueuedSftpSessionPool(15, sessionFactory);
pool.afterPropertiesSet();
SftpInboundRemoteFileSystemSynchronizer sftpSync = new SftpInboundRemoteFileSystemSynchronizer();
sftpSync.setClientPool(pool);
sftpSync.setLocalDirectory(this.localDirectoryResource);
sftpSync.setShouldDeleteSourceFile(ackRemoteDir);
sftpSync.setFilter(compositeFtpFileListFilter);
sftpSync.afterPropertiesSet();// todo is this correct ?
sftpSync.start();//todo
sftpMsgSrc.setRemotePredicate(compositeFtpFileListFilter);
sftpMsgSrc.setSynchronizer(sftpSync);
sftpMsgSrc.setClientPool( pool);
sftpMsgSrc.setRemotePath( this.remoteDirectory);
sftpMsgSrc.setLocalDirectory(this.localDirectoryResource);
sftpMsgSrc.setBeanFactory(this.getBeanFactory());
sftpMsgSrc.setAutoStartup(true);
sftpMsgSrc.afterPropertiesSet();
sftpMsgSrc.start();
return sftpMsgSrc;
}
private Resource fromText(String path) {
ResourceEditor resourceEditor = new ResourceEditor(this.resourceLoader);
resourceEditor.setAsText(path);
return (Resource) resourceEditor.getValue();
}
}

4
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/TestSftpReceipt.java
View File

@@ -19,7 +19,7 @@ import java.util.logging.Logger;
* @author Josh Long
*/
public class TestSftpReceipt {
/*
private static final Logger logger = Logger.getLogger(TestSftpReceipt.class.getName());
private SftpSessionFactory sftpSessionFactory;
private String host;
@@ -104,5 +104,5 @@ public class TestSftpReceipt {
sftpSessionFactory.afterPropertiesSet();
return sftpSessionFactory;
}
}*/
}

3
spring-integration-sftp/src/test/resources/TestInboundSftp.xml
View File

@@ -34,6 +34,7 @@
http://www.springframework.org/schema/lang http://www.springframework.org/schema/lang/spring-lang-3.0.xsd
http://www.springframework.org/schema/integration/sftp http://www.springframework.org/schema/integration/sftp/spring-integration-sftp-2.0.xsd">
<context:component-scan base-package="org.springframework.integration.sftp"/>
<context:property-placeholder
location="file://${user.home}/Desktop/sftp.properties"
@@ -49,6 +50,8 @@
channel="inboundFilesChannel"
filename-pattern=".*?jpg"
username="${sftp.username}"
auto-create-directories="true"
auto-delete-remote-files-on-sync="true"
host="${sftp.host}">
<poller>
<interval-trigger interval="1000" time-unit="MILLISECONDS"/>