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

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

Browse Source
This commit is contained in:
Josh Long
2010-08-18 03:17:49 +00:00
parent 6dec375d36
commit 8307772dd0
12 changed files with 215 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
View File

@@ -149,7 +149,7 @@ public class FileReadingMessageSource implements MessageSource<File>,
}
/**
* Optional. Sets a {@link org.springframework.integration.file.locking.FileLocker} to be used to guard files
* Optional. Sets a {@link FileLocker} to be used to guard files
* against duplicate processing.
* <p/>
* <b>The supplied FileLocker must be thread safe</b>

33
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/AbstractFTPFileListFilter.java Normal file
View File

@@ -0,0 +1,33 @@
package org.springframework.integration.ftp;
import org.apache.commons.net.ftp.FTPFile;
import java.util.ArrayList;
import java.util.List;
/**
* Convenience implementation patterned off {@link org.springframework.integration.file.FileListFilter}
*
* @author Josh Long
*/
public abstract class AbstractFTPFileListFilter implements FTPFileListFilter {
/**
* {@inheritDoc}
*/
abstract public boolean accept(FTPFile ftpFile);
public List<FTPFile> filterFiles(FTPFile[] files) {
List<FTPFile> accepted = new ArrayList<FTPFile>();
if (files != null) {
for (FTPFile f : files) {
if (this.accept(f)) {
accepted.add(f);
}
}
}
return accepted;
}
}

37
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/CompositeFTPFileListFilter.java Normal file
View File

@@ -0,0 +1,37 @@
package org.springframework.integration.ftp;
import org.apache.commons.net.ftp.FTPFile;
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 FTPFileListFilter {
private Set<FTPFileListFilter> filters;
public CompositeFTPFileListFilter(FTPFileListFilter... ftpFileListFilter) {
this.filters = new LinkedHashSet<FTPFileListFilter>(Arrays.asList(ftpFileListFilter));
}
public CompositeFTPFileListFilter(Collection<FTPFileListFilter> ftpFileListFilter) {
this.filters = new LinkedHashSet<FTPFileListFilter>(ftpFileListFilter);
}
public void addFilter( FTPFileListFilter ftpFileListFilter ) {
this.filters.add(ftpFileListFilter);
}
public List<FTPFile> filterFiles(FTPFile[] files) {
Assert.notNull(files, "files[] can't be null!");
List<FTPFile> leftOver = Arrays.asList(files);
for (FTPFileListFilter ff : this.filters)
leftOver = ff.filterFiles(leftOver.toArray(new FTPFile[leftOver.size()]));
return leftOver;
}
}

14
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPFileListFilter.java Normal file
View File

@@ -0,0 +1,14 @@
package org.springframework.integration.ftp;
import org.apache.commons.net.ftp.FTPFile;
import java.util.List;
/**
* Filters out all the FTPFiles taken in a scan of the remote mount o
*
* @author Josh Long
*/
public interface FTPFileListFilter {
List<FTPFile> filterFiles (FTPFile [] files);
}

2
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPFileSource.java
View File

@@ -29,7 +29,6 @@ import org.springframework.scheduling.Trigger;
import java.io.File;
import java.io.IOException;
import java.util.logging.Logger;
import java.util.regex.Pattern;
@@ -39,7 +38,6 @@ import java.util.regex.Pattern;
* @author Iwein Fuld
*/
public class FTPFileSource implements MessageSource<File>, InitializingBean, Lifecycle {
private static final Logger logger = Logger.getLogger(FTPFileSource.class.getName());
private FileReadingMessageSource fileSource;
private FTPInboundSynchronizer synchronizer;

30
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPInboundSynchronizer.java
View File

@@ -32,6 +32,9 @@ import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Arrays;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.ScheduledFuture;
@@ -40,7 +43,6 @@ import java.util.concurrent.ScheduledFuture;
* It will NOT move new files put into the local directory to the remote server.
*
* @author Iwein Fuld
*
*/
public class FTPInboundSynchronizer implements InitializingBean, Lifecycle {
private static final Log logger = LogFactory.getLog(FTPInboundSynchronizer.class);
@@ -52,6 +54,16 @@ public class FTPInboundSynchronizer implements InitializingBean, Lifecycle {
private volatile Resource localDirectory;
private boolean running = false;
private ScheduledFuture<?> scheduledFuture;
private FTPFileListFilter filter;
private FTPFileListFilter acceptAllFTPFileListFilter = new FTPFileListFilter() {
public List<FTPFile> filterFiles(FTPFile[] files) {
return Arrays.asList(files);
}
};
public void setFilter(FTPFileListFilter filter) {
this.filter = filter;
}
public void setTaskScheduler(TaskScheduler scheduler) {
this.taskScheduler = scheduler;
@@ -71,6 +83,10 @@ public class FTPInboundSynchronizer implements InitializingBean, Lifecycle {
public void afterPropertiesSet() throws Exception {
Assert.notNull(localDirectory, "'localDirectory' is required.");
if (this.filter == null) {
this.filter = acceptAllFTPFileListFilter;
}
}
private void synchronize() {
@@ -78,22 +94,20 @@ public class FTPInboundSynchronizer implements InitializingBean, Lifecycle {
FTPClient client = this.clientPool.getClient();
Assert.state(client != null, FTPClientPool.class.getSimpleName() + " returned 'null' client this most likely a bug in the pool implementation.");
FTPFile[] fileList = client.listFiles();
Collection<FTPFile> fileList = this.filter.filterFiles(client.listFiles());
try {
for (FTPFile ftpFile : fileList) {
/*
* according to the FTPFile javadoc the list can contain
* nulls if files couldn't be parsed
*/
* according to the FTPFile javadoc the list can contain
* nulls if files couldn't be parsed
*/
if ((ftpFile != null) && ftpFile.isFile()) {
copyFileToLocalDirectory(client, ftpFile, this.localDirectory);
}
}
} finally {
if (client != null) {
this.clientPool.releaseClient(client);
}
this.clientPool.releaseClient(client);
}
} catch (IOException e) {
throw new MessagingException("Problem occurred while synchronizing remote to local directory", e);

36
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPMessageSourceFactoryBean.java
View File

@@ -1,6 +1,5 @@
package org.springframework.integration.ftp;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.SystemUtils;
import org.apache.commons.net.ftp.FTPClient;
import org.springframework.beans.BeansException;
@@ -16,9 +15,11 @@ import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.util.Assert;
import org.springframework.util.ErrorHandler;
import org.springframework.util.StringUtils;
import java.io.File;
import java.util.Map;
import java.util.regex.Pattern;
/**
@@ -29,6 +30,7 @@ import java.util.Map;
public class FTPMessageSourceFactoryBean extends AbstractFactoryBean<FTPFileSource> implements ResourceLoaderAware, ApplicationContextAware {
private int port;
private boolean autoCreateDirectories;
private String filenamePattern;
private String username;
private String password;
private String host;
@@ -42,6 +44,15 @@ public class FTPMessageSourceFactoryBean extends AbstractFactoryBean<FTPFileSour
private FileReadingMessageSource fileReadingMessageSource;
private int clientMode = FTPClient.ACTIVE_LOCAL_DATA_CONNECTION_MODE;
/**
* Used to teach the FTP adapter what files you are interested in receiving
*/
private FTPFileListFilter filter;
public void setFilenamePattern(String filenamePattern) {
this.filenamePattern = filenamePattern;
}
public void setApplicationContext(ApplicationContext applicationContext)
throws BeansException {
this.applicationContext = applicationContext;
@@ -87,15 +98,20 @@ public class FTPMessageSourceFactoryBean extends AbstractFactoryBean<FTPFileSour
this.clientMode = clientMode;
}
public void setFilter(FTPFileListFilter filter) {
this.filter = filter;
}
@Override
protected FTPFileSource createInstance() throws Exception {
// setup local dir
if (StringUtils.isEmpty(this.localWorkingDirectory)) {
if (!StringUtils.hasText(this.localWorkingDirectory)) {
File tmp = SystemUtils.getJavaIoTmpDir();
File ftpTmp = new File(tmp, "ftpInbound");
this.localWorkingDirectory = "file://" + ftpTmp.getAbsolutePath();
}
Assert.hasText( this.localWorkingDirectory , "the local working directory can't be null!" );
Assert.hasText(this.localWorkingDirectory, "the local working directory can't be null!");
ResourceEditor resourceEditor = new ResourceEditor(this.resourceLoader);
resourceEditor.setAsText(this.localWorkingDirectory);
@@ -104,6 +120,20 @@ public class FTPMessageSourceFactoryBean extends AbstractFactoryBean<FTPFileSour
this.ftpInboundSynchronizer = new FTPInboundSynchronizer();
CompositeFTPFileListFilter compositeFTPFileListFilter = new CompositeFTPFileListFilter();
if (StringUtils.hasText(this.filenamePattern)) {
PatternMatchingFTPFileListFilter patternMatchingFTPFileListFilter = new PatternMatchingFTPFileListFilter();
patternMatchingFTPFileListFilter.setPattern(Pattern.compile(this.filenamePattern));
compositeFTPFileListFilter.addFilter(patternMatchingFTPFileListFilter);
}
if (this.filter != null) {
compositeFTPFileListFilter.addFilter(this.filter);
}
this.ftpInboundSynchronizer.setFilter(compositeFTPFileListFilter);
if (this.taskScheduler == null) {
Map<String, TaskScheduler> tss = null;

51
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/PatternMatchingFTPFileListFilter.java Normal file
View File

@@ -0,0 +1,51 @@
package org.springframework.integration.ftp;
import org.apache.commons.lang.builder.ToStringBuilder;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.net.ftp.FTPFile;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import java.util.regex.Pattern;
/**
* Validates {@link org.apache.commons.net.ftp.FTPFile}s against a {@link java.util.regex.Pattern}.
* Patterned very much like {@link org.springframework.integration.file.PatternMatchingFileListFilter}.
*
* @author Josh Long
*/
public class PatternMatchingFTPFileListFilter extends AbstractFTPFileListFilter 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(FTPFile ftpFile) {
if (logger.isDebugEnabled()) {
logger.debug("testing: " + ToStringBuilder.reflectionToString(ftpFile));
}
return (ftpFile != null) && this.pattern.matcher(ftpFile.getName()).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");
}
}

6
spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FTPNamespaceHandler.java
View File

@@ -81,8 +81,10 @@ public class FTPNamespaceHandler extends NamespaceHandlerSupport {
protected String parseSource(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(PACKAGE_NAME + ".FTPMessageSourceFactoryBean");
for (String p : ("auto-create-directories,username,password,host,port," + "remote-directory,local-working-directory").split(",")) {
//auto-delete-remote-files-on-sync
// reference
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element,"filter");
for (String p : ("filename-pattern,auto-create-directories,username,password,host,port," + "remote-directory,local-working-directory").split(",")) {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, p);
}

13
spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.0.xsd
View File

@@ -134,6 +134,17 @@
</xsd:appinfo>
</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="local-working-directory" type="xsd:string"/>
<xsd:attribute name="auto-create-directories" type="xsd:boolean"/>
@@ -202,4 +213,4 @@
</xsd:element>
</xsd:schema>
</xsd:schema>

4
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/InboundFTPFileServiceActivator.java
View File

@@ -1,14 +1,10 @@
package org.springframework.integration.ftp;
import org.apache.commons.lang.StringUtils;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.stereotype.Component;
import java.io.File;

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

@@ -13,10 +13,15 @@
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>
<ftp:inbound-channel-adapter remote-directory="${ftp.remotedir}" channel="ftpIn" auto-create-directories="true"
host="${ftp.host}" auto-delete-remote-files-on-sync="false"
username="${ftp.username}" password="${ftp.password}" port="2222"
client-mode="passive-local-data-connection-mode"
username="${ftp.username}" password="${ftp.password}" port="2222"
client-mode="passive-local-data-connection-mode"
filename-pattern=".*?jpg"
>
<int:poller>
<int:interval-trigger interval="10000" time-unit="MILLISECONDS"/>
@@ -30,4 +35,4 @@
<int:service-activator input-channel="ftpIn" ref="inboundFTPFileServiceActivator"/>
</beans>
</beans>