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

Taught the SFTP adapter how to do file name filtering and predicates INT-1357

Browse Source
This commit is contained in:
Josh Long
2010-08-18 04:26:26 +00:00
parent 8307772dd0
commit a3ad837468
11 changed files with 308 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-ftp/src/test/resources/inbound-ftp-context.xml
View File

@@ -8,11 +8,11 @@
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/ftp http://www.springframework.org/schema/integration/ftp/spring-integration-ftp.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<context:property-placeholder
location="file://${user.home}/Desktop/ftp.properties"
ignore-unresolvable="true"/>
<bean class="org.springframework.integration.ftp.PatternMatchingFTPFileListFilter" id="patternMatchingFTPFileListFilter">
<property name="patternExpression" value=".*?jpg"/>
</bean>

45
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/AbstractSFTPFileListFilter.java Normal file
View File

@@ -0,0 +1,45 @@
/*
* Copyright 2010 the original author or authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* 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 java.util.ArrayList;
import java.util.List;
/**
* Convenience implementation patterned off {@link org.springframework.integration.file.FileListFilter}
*
* @author Josh Long
*/
public abstract class AbstractSFTPFileListFilter implements SFTPFileListFilter {
abstract public boolean accept(ChannelSftp.LsEntry lsEntry);
public List<ChannelSftp.LsEntry> filterFiles(ChannelSftp.LsEntry[] files) {
List<ChannelSftp.LsEntry> accepted = new ArrayList<ChannelSftp.LsEntry>();
if (files != null) {
for (ChannelSftp.LsEntry lsEntry : files)
if (this.accept(lsEntry)) {
accepted.add(lsEntry);
}
}
return accepted;
}
}

54
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/CompositeFTPFileListFilter.java Normal file
View File

@@ -0,0 +1,54 @@
/*
* Copyright 2010 the original author or authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* 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.util.Assert;
import java.util.*;
/**
* Patterned very much on the {@link org.springframework.integration.file.CompositeFileListFilter}
*
* @author Josh Long
*/
public class CompositeFTPFileListFilter implements SFTPFileListFilter {
private Set<SFTPFileListFilter> filters;
public CompositeFTPFileListFilter(SFTPFileListFilter... ftpFileListFilter) {
this.filters = new LinkedHashSet<SFTPFileListFilter>(Arrays.asList(ftpFileListFilter));
}
public CompositeFTPFileListFilter(Collection<SFTPFileListFilter> ftpFileListFilter) {
this.filters = new LinkedHashSet<SFTPFileListFilter>(ftpFileListFilter);
}
public void addFilter(SFTPFileListFilter ftpFileListFilter) {
this.filters.add(ftpFileListFilter);
}
public List<ChannelSftp.LsEntry> filterFiles(ChannelSftp.LsEntry[] files) {
Assert.notNull(files, "files[] can't be null!");
List<ChannelSftp.LsEntry> leftOver = Arrays.asList(files);
for (SFTPFileListFilter ff : this.filters)
leftOver = ff.filterFiles(leftOver.toArray(new ChannelSftp.LsEntry[leftOver.size()]));
return leftOver;
}
}

66
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/PatternMatchingSFTPFileListFilter.java Normal file
View File

@@ -0,0 +1,66 @@
/*
* Copyright 2010 the original author or authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* 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.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
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}.
*
* @author Josh Long
*/
public class PatternMatchingSFTPFileListFilter extends AbstractSFTPFileListFilter implements InitializingBean {
private Log logger = LogFactory.getLog(getClass());
private Pattern pattern;
private String patternExpression;
public void setPattern(Pattern pattern) {
this.pattern = pattern;
}
public void setPatternExpression(String patternExpression) {
this.patternExpression = patternExpression;
}
@Override
public boolean accept(ChannelSftp.LsEntry lsEntry) {
if (logger.isDebugEnabled()) {
logger.debug("testing: " + ToStringBuilder.reflectionToString(lsEntry));
}
return (lsEntry != null) && this.pattern.matcher(lsEntry.getFilename()).matches();
}
public void afterPropertiesSet() throws Exception {
if (StringUtils.hasText(this.patternExpression) && (this.pattern == null)) {
this.pattern = Pattern.compile(this.patternExpression);
}
Assert.notNull(this.pattern, "the pattern must not be null");
}
}

30
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPFileListFilter.java Normal file
View File

@@ -0,0 +1,30 @@
/*
* Copyright 2010 the original author or authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* 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 java.util.List;
/**
* Filters out all the {@link com.jcraft.jsch.ChannelSftp.LsEntry} taken in a scan of the remote mount
* and returns the balance. These are then sync'd to the local directory.
*
* @author Josh Long
*/
public interface SFTPFileListFilter {
List<ChannelSftp.LsEntry> filterFiles (ChannelSftp.LsEntry [] files);
}

73
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPInboundSynchronizer.java
View File

@@ -17,34 +17,31 @@ package org.springframework.integration.sftp;
import com.jcraft.jsch.ChannelSftp;
import com.jcraft.jsch.SftpATTRS;
import org.apache.commons.io.IOUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.io.Resource;
import org.springframework.integration.MessagingException;
import org.springframework.scheduling.TaskScheduler;
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.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ScheduledFuture;
/**
*
* This handles keeping the {@link #localDirectory} in sync with the contents of the remote mount. From there, files are deposited
* into a folder where the {@link org.springframework.integration.file.FileReadingMessageSource} will eventually deliver them as events
*
*
* @author Josh Long
* @author Mario Gray
*/
@@ -53,6 +50,7 @@ public class SFTPInboundSynchronizer implements InitializingBean {
// a lot of the approach for this (including the use of a FileReadingMessageSource and the regex / mask approach were lifted from FtpInboundSynchronizer
static final String INCOMPLETE_EXTENSION = ".INCOMPLETE";
private Log logger = LogFactory.getLog(getClass());
private volatile Resource localDirectory;
private volatile SFTPSessionPool pool;
private volatile ScheduledFuture<?> scheduledFuture;
@@ -61,11 +59,23 @@ public class SFTPInboundSynchronizer implements InitializingBean {
private volatile Trigger trigger = new PeriodicTrigger(DEFAULT_REFRESH_RATE);
private volatile boolean autoCreatePath;
private volatile boolean running;
private SFTPFileListFilter filter;
public void setFilter(SFTPFileListFilter filter) {
this.filter = filter;
}
private volatile boolean shouldDeleteDownloadedRemoteFiles; //.. this is false
private SFTPFileListFilter acceptAllFilteListFilter = new SFTPFileListFilter(){
public List<ChannelSftp.LsEntry> filterFiles(ChannelSftp.LsEntry[] files) {
return Arrays.asList( files);
}
} ;
public void afterPropertiesSet() throws Exception {
assert (taskScheduler != null) : "taskScheduler can't be null!";
assert (localDirectory != null) : "the localDirectory property must not be null!";
Assert.state(taskScheduler != null, "taskScheduler can't be null!");
Assert.state(localDirectory != null, "the localDirectory property must not be null!");
File localDir = localDirectory.getFile();
@@ -76,18 +86,10 @@ public class SFTPInboundSynchronizer implements InitializingBean {
}
}
}
}
public ScheduledFuture<?> getScheduledFuture() {
return scheduledFuture;
}
if(this.filter == null)
this.filter = acceptAllFilteListFilter;
public TaskScheduler getTaskScheduler() {
return taskScheduler;
}
public Trigger getTrigger() {
return trigger;
}
public boolean isRunning() {
@@ -134,8 +136,9 @@ public class SFTPInboundSynchronizer implements InitializingBean {
if (running) {
return;
}
assert checkThatRemotePathExists(remotePath) : "the remotePath had better exist!";
assert taskScheduler != null : "'taskScheduler' is required";
Assert.state(checkThatRemotePathExists(remotePath), "the remotePath should exist before we can sync with it!");
Assert.state(taskScheduler != null, "'taskScheduler' is required");
scheduledFuture = taskScheduler.schedule(new SynchronizeTask(), trigger);
@@ -146,7 +149,8 @@ public class SFTPInboundSynchronizer implements InitializingBean {
if (!running) {
return;
}
assert scheduledFuture != null : "scheduledFuture is null!";
Assert.state(scheduledFuture != null, "scheduledFuture is null!");
this.scheduledFuture.cancel(true);
this.running = false;
}
@@ -160,7 +164,10 @@ public class SFTPInboundSynchronizer implements InitializingBean {
session.start();
ChannelSftp channelSftp = session.getChannel();
Collection<ChannelSftp.LsEntry> files = channelSftp.ls(remotePath);
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.filterFiles( entries );
for (ChannelSftp.LsEntry lsEntry : files) {
if ((lsEntry != null) && !lsEntry.getAttrs().isDir() && !lsEntry.getAttrs().isLink()) {
@@ -178,24 +185,24 @@ public class SFTPInboundSynchronizer implements InitializingBean {
/**
* there be dragons this way ... This method will check to ensure that the remote directory exists. If the directory
* doesnt exist, and autoCreatePath is configured to be true, then this method makes a few reasonably sane attempts
* doesnt exist, and autoCreatePath is 'true,' then this method makes a few reasonably sane attempts
* to create it. Otherwise, it fails fast.
*
* @param rPath the path on the remote SSH / SFTP server to create.
* @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 rPath) {
private boolean checkThatRemotePathExists(String remotePath) {
SFTPSession session = null;
ChannelSftp channelSftp = null;
try {
session = pool.getSession();
assert session != null : "session's not null";
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(rPath);
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;
@@ -203,9 +210,9 @@ public class SFTPInboundSynchronizer implements InitializingBean {
if (this.autoCreatePath && (pool != null) && (session != null)) {
try {
if (channelSftp != null) {
channelSftp.mkdir(rPath);
channelSftp.mkdir(remotePath);
if (channelSftp.stat(rPath).isDir()) {
if (channelSftp.stat(remotePath).isDir()) {
return true;
}
}
@@ -261,10 +268,6 @@ public class SFTPInboundSynchronizer implements InitializingBean {
return false;
}
private boolean foo() {
return false;
}
class SynchronizeTask implements Runnable {
public void run() {
try {

49
spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPMessageSourceFactoryBean.java
View File

@@ -15,43 +15,32 @@
*/
package org.springframework.integration.sftp.config;
import org.apache.commons.lang.StringUtils;
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.sftp.QueuedSFTPSessionPool;
import org.springframework.integration.sftp.SFTPInboundSynchronizer;
import org.springframework.integration.sftp.SFTPMessageSource;
import org.springframework.integration.sftp.SFTPSessionFactory;
import org.springframework.integration.sftp.*;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.ErrorHandler;
import org.springframework.util.StringUtils;
import java.io.File;
import java.util.Map;
/**
*
* Building a {@link org.springframework.integration.sftp.SFTPMessageSource} is a complicated because we also
* use a {@link org.springframework.integration.file.FileReadingMessageSource} to handle the "receipt" of files in
* a {@link #localWorkingDirectory}.
*
*
* @author Josh Long
*/
public class SFTPMessageSourceFactoryBean extends AbstractFactoryBean<SFTPMessageSource> implements ApplicationContextAware, ResourceLoaderAware {
@@ -72,6 +61,8 @@ public class SFTPMessageSourceFactoryBean extends AbstractFactoryBean<SFTPMessag
private boolean autoCreateDirectories;
private boolean autoDeleteRemoteFilesOnSync;
private int port = 22;
private SFTPFileListFilter filter;
private String filenamePattern;
public FileReadingMessageSource getFileReadingMessageSource() {
return fileReadingMessageSource;
@@ -199,6 +190,14 @@ public class SFTPMessageSourceFactoryBean extends AbstractFactoryBean<SFTPMessag
this.trigger = trigger;
}
public void setFilenamePattern(String filenamePattern) {
this.filenamePattern = filenamePattern;
}
public void setFilter(SFTPFileListFilter filter) {
this.filter = filter;
}
public void setUsername(final String username) {
this.username = username;
}
@@ -206,12 +205,11 @@ public class SFTPMessageSourceFactoryBean extends AbstractFactoryBean<SFTPMessag
@Override
protected SFTPMessageSource createInstance() throws Exception {
try {
if ((localWorkingDirectory == null) || StringUtils.isEmpty(localWorkingDirectory)) {
if ((localWorkingDirectory == null) || !StringUtils.hasText(localWorkingDirectory)) {
File tmp = SystemUtils.getJavaIoTmpDir();
File sftpTmp = new File(tmp, "sftpInbound");
this.localWorkingDirectory = "file://" + sftpTmp.getAbsolutePath();
}
assert !StringUtils.isEmpty(this.localWorkingDirectory) : "the local working directory mustn't be null!";
// resource for local directory
ResourceEditor editor = new ResourceEditor(this.resourceLoader);
@@ -222,6 +220,21 @@ public class SFTPMessageSourceFactoryBean extends AbstractFactoryBean<SFTPMessag
synchronizer = new SFTPInboundSynchronizer();
CompositeFTPFileListFilter compositeFTPFileListFilter = new CompositeFTPFileListFilter();
if (StringUtils.hasText(this.filenamePattern)) {
PatternMatchingSFTPFileListFilter flp = new PatternMatchingSFTPFileListFilter();
flp.setPatternExpression(this.filenamePattern);
flp.afterPropertiesSet();
compositeFTPFileListFilter.addFilter(flp);
}
if (this.filter != null) {
compositeFTPFileListFilter.addFilter(this.filter);
}
synchronizer.setFilter(compositeFTPFileListFilter);
if (null == taskScheduler) {
Map<String, TaskScheduler> tss = null;
@@ -245,8 +258,8 @@ public class SFTPMessageSourceFactoryBean extends AbstractFactoryBean<SFTPMessag
this.taskScheduler = ts;
}
SFTPSessionFactory sessionFactory = SFTPSessionUtils.buildSftpSessionFactory(this.getHost(), this.getPassword(), this.getUsername(), this.getKeyFile(), this.getKeyFilePassword(),
this.getPort());
SFTPSessionFactory sessionFactory = SFTPSessionUtils.buildSftpSessionFactory(
this.getHost(), this.getPassword(), this.getUsername(), this.getKeyFile(), this.getKeyFilePassword(), this.getPort());
QueuedSFTPSessionPool pool = new QueuedSFTPSessionPool(15, sessionFactory);
pool.afterPropertiesSet();

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

@@ -66,7 +66,9 @@ public class SFTPNamespaceHandler extends NamespaceHandlerSupport {
protected String parseSource(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition( SFTPMessageSourceFactoryBean.class.getName());
for (String p : "auto-create-directories,username,password,host,key-file,key-file-password,remote-directory,local-working-directory,auto-delete-remote-files-on-sync".split(",")) {
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "filter");
for (String p : "filename-pattern,auto-create-directories,username,password,host,key-file,key-file-password,remote-directory,local-working-directory,auto-delete-remote-files-on-sync".split(",")) {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, p);
}

15
spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-2.0.xsd
View File

@@ -74,7 +74,7 @@
There is support for automatically deleting remote files upon synchornization. This adapter supports two connectivity options:
<ol>
<li> Password authentication: using this opton, authenticatio is done using a username and a password.</li>
<li> Password authentication: using this opton, authentication is done using a username and a password.</li>
<li> Key-based authentication: using this option, you may specify a key that will be used to authenticate. If they key itself is encrypted and requires a password, you may specify that, as well.</li>
</ol>
]]></xsd:documentation>
@@ -94,6 +94,17 @@
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="filter" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.ftp.FTPFileListFilter"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="filename-pattern" type="xsd:string"/>
<xsd:attribute name="username" type="xsd:string" use="required"/>
<xsd:attribute name="remote-directory" type="xsd:string" use="required"/>
@@ -114,4 +125,4 @@
</xsd:element>
</xsd:schema>
</xsd:schema>

14
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/TestInboundSFTP.java Normal file
View File

@@ -0,0 +1,14 @@
package org.springframework.integration.sftp;
import org.springframework.context.support.ClassPathXmlApplicationContext;
/**
* @author Josh Long
*/
public class TestInboundSFTP {
static public void main(String [] args) throws Throwable {
ClassPathXmlApplicationContext classPathXmlApplicationContext = new ClassPathXmlApplicationContext("TestInboundSFTP.xml");
classPathXmlApplicationContext.start();
}
}

23
spring-integration-sftp/src/test/resources/TestInboundSFTP.xml
View File

@@ -35,24 +35,27 @@
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"
ignore-unresolvable="true"/>
<channel id="inboundFilesChannel"/>
<sftp:inbound-channel-adapter username="user" remote-directory="/home/user/Desktop/in" password="password"
host="host" channel="inboundFilesChannel">
<sftp:inbound-channel-adapter
key-file="${sftp.key}"
remote-directory="${sftp.remote-dir}"
channel="inboundFilesChannel"
filename-pattern=".*?jpg"
username="${sftp.username}"
host="${sftp.host}">
<poller>
<interval-trigger interval="10"/>
<interval-trigger interval="1000" time-unit="MILLISECONDS"/>
</poller>
</sftp:inbound-channel-adapter>
<sftp:outbound-channel-adapter
key-file="/home/user/user.pem"
remote-directory="remote_mount_key"
channel="inboundFilesChannel"
username="ubuntu"
host="siteonec2usingubuntuami.com"/>
<service-activator ref="sftpAnnouncer" input-channel="inboundFilesChannel"/>
</beans:beans>
</beans:beans>