From a3ad837468480e53703a015a1c963a1d44392ddf Mon Sep 17 00:00:00 2001 From: Josh Long Date: Wed, 18 Aug 2010 04:26:26 +0000 Subject: [PATCH] Taught the SFTP adapter how to do file name filtering and predicates INT-1357 --- .../test/resources/inbound-ftp-context.xml | 2 +- .../sftp/AbstractSFTPFileListFilter.java | 45 ++++++++++++ .../sftp/CompositeFTPFileListFilter.java | 54 ++++++++++++++ .../PatternMatchingSFTPFileListFilter.java | 66 +++++++++++++++++ .../integration/sftp/SFTPFileListFilter.java | 30 ++++++++ .../sftp/SFTPInboundSynchronizer.java | 73 ++++++++++--------- .../config/SFTPMessageSourceFactoryBean.java | 49 ++++++++----- .../sftp/config/SFTPNamespaceHandler.java | 4 +- .../config/spring-integration-sftp-2.0.xsd | 15 +++- .../integration/sftp/TestInboundSFTP.java | 14 ++++ .../src/test/resources/TestInboundSFTP.xml | 23 +++--- 11 files changed, 308 insertions(+), 67 deletions(-) create mode 100644 spring-integration-sftp/src/main/java/org/springframework/integration/sftp/AbstractSFTPFileListFilter.java create mode 100644 spring-integration-sftp/src/main/java/org/springframework/integration/sftp/CompositeFTPFileListFilter.java create mode 100644 spring-integration-sftp/src/main/java/org/springframework/integration/sftp/PatternMatchingSFTPFileListFilter.java create mode 100644 spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPFileListFilter.java create mode 100644 spring-integration-sftp/src/test/java/org/springframework/integration/sftp/TestInboundSFTP.java diff --git a/spring-integration-ftp/src/test/resources/inbound-ftp-context.xml b/spring-integration-ftp/src/test/resources/inbound-ftp-context.xml index a5193c5af6..b30314ffd5 100644 --- a/spring-integration-ftp/src/test/resources/inbound-ftp-context.xml +++ b/spring-integration-ftp/src/test/resources/inbound-ftp-context.xml @@ -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"> - + diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/AbstractSFTPFileListFilter.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/AbstractSFTPFileListFilter.java new file mode 100644 index 0000000000..f8a91a9107 --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/AbstractSFTPFileListFilter.java @@ -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 filterFiles(ChannelSftp.LsEntry[] files) { + List accepted = new ArrayList(); + + if (files != null) { + for (ChannelSftp.LsEntry lsEntry : files) + if (this.accept(lsEntry)) { + accepted.add(lsEntry); + } + } + + return accepted; + } +} diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/CompositeFTPFileListFilter.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/CompositeFTPFileListFilter.java new file mode 100644 index 0000000000..a906d90533 --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/CompositeFTPFileListFilter.java @@ -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 filters; + + public CompositeFTPFileListFilter(SFTPFileListFilter... ftpFileListFilter) { + this.filters = new LinkedHashSet(Arrays.asList(ftpFileListFilter)); + } + + public CompositeFTPFileListFilter(Collection ftpFileListFilter) { + this.filters = new LinkedHashSet(ftpFileListFilter); + } + + public void addFilter(SFTPFileListFilter ftpFileListFilter) { + this.filters.add(ftpFileListFilter); + } + + public List filterFiles(ChannelSftp.LsEntry[] files) { + Assert.notNull(files, "files[] can't be null!"); + + List leftOver = Arrays.asList(files); + + for (SFTPFileListFilter ff : this.filters) + leftOver = ff.filterFiles(leftOver.toArray(new ChannelSftp.LsEntry[leftOver.size()])); + + return leftOver; + } +} diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/PatternMatchingSFTPFileListFilter.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/PatternMatchingSFTPFileListFilter.java new file mode 100644 index 0000000000..a6ca0b9216 --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/PatternMatchingSFTPFileListFilter.java @@ -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"); + } +} diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPFileListFilter.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPFileListFilter.java new file mode 100644 index 0000000000..8d15dabd04 --- /dev/null +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPFileListFilter.java @@ -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 filterFiles (ChannelSftp.LsEntry [] files); +} diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPInboundSynchronizer.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPInboundSynchronizer.java index f050b2d800..d11ec51cfa 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPInboundSynchronizer.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/SFTPInboundSynchronizer.java @@ -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 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 files = channelSftp.ls(remotePath); + Collection beforeFilter = channelSftp.ls(remotePath); + ChannelSftp.LsEntry [] entries = beforeFilter == null? new ChannelSftp.LsEntry[0] : + beforeFilter.toArray(new ChannelSftp.LsEntry[ beforeFilter.size()]) ; + Collection 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 { diff --git a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPMessageSourceFactoryBean.java b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPMessageSourceFactoryBean.java index 274db9bfea..d6ef0cd39b 100644 --- a/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPMessageSourceFactoryBean.java +++ b/spring-integration-sftp/src/main/java/org/springframework/integration/sftp/config/SFTPMessageSourceFactoryBean.java @@ -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 implements ApplicationContextAware, ResourceLoaderAware { @@ -72,6 +61,8 @@ public class SFTPMessageSourceFactoryBean extends AbstractFactoryBean tss = null; @@ -245,8 +258,8 @@ public class SFTPMessageSourceFactoryBean extends AbstractFactoryBean -
  • Password authentication: using this opton, authenticatio is done using a username and a password.
    • +
  • Password authentication: using this opton, authentication is done using a username and a password.
  • 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.
    • ]]> @@ -94,6 +94,17 @@ + + + + + + + + + + + @@ -114,4 +125,4 @@ - \ No newline at end of file + diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/TestInboundSFTP.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/TestInboundSFTP.java new file mode 100644 index 0000000000..a2f4271322 --- /dev/null +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/TestInboundSFTP.java @@ -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(); + } +} diff --git a/spring-integration-sftp/src/test/resources/TestInboundSFTP.xml b/spring-integration-sftp/src/test/resources/TestInboundSFTP.xml index eee6bc37f5..6ef7076255 100644 --- a/spring-integration-sftp/src/test/resources/TestInboundSFTP.xml +++ b/spring-integration-sftp/src/test/resources/TestInboundSFTP.xml @@ -35,24 +35,27 @@ http://www.springframework.org/schema/integration/sftp http://www.springframework.org/schema/integration/sftp/spring-integration-sftp-2.0.xsd"> + - + - + - + - \ No newline at end of file +