diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
index ef78598160..85d71be275 100644
--- a/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
+++ b/spring-integration-file/src/main/java/org/springframework/integration/file/FileReadingMessageSource.java
@@ -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>
diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/AbstractFTPFileListFilter.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/AbstractFTPFileListFilter.java
new file mode 100644
index 0000000000..87144af198
--- /dev/null
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/AbstractFTPFileListFilter.java
@@ -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;
+    }
+}
diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/CompositeFTPFileListFilter.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/CompositeFTPFileListFilter.java
new file mode 100644
index 0000000000..2923f5e390
--- /dev/null
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/CompositeFTPFileListFilter.java
@@ -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;
+    }
+}
diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPFileListFilter.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPFileListFilter.java
new file mode 100644
index 0000000000..77d013846a
--- /dev/null
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPFileListFilter.java
@@ -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);
+}
diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPFileSource.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPFileSource.java
index 2efaa37d72..3d2a961d15 100644
--- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPFileSource.java
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPFileSource.java
@@ -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;
 
diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPInboundSynchronizer.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPInboundSynchronizer.java
index 7e2b931057..139a049b52 100644
--- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPInboundSynchronizer.java
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPInboundSynchronizer.java
@@ -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);
diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPMessageSourceFactoryBean.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPMessageSourceFactoryBean.java
index 92894e8ad6..43c7cbda4a 100644
--- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPMessageSourceFactoryBean.java
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/FTPMessageSourceFactoryBean.java
@@ -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;
 
diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/PatternMatchingFTPFileListFilter.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/PatternMatchingFTPFileListFilter.java
new file mode 100644
index 0000000000..ccecb7de49
--- /dev/null
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/PatternMatchingFTPFileListFilter.java
@@ -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");
+    }
+}
diff --git a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FTPNamespaceHandler.java b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FTPNamespaceHandler.java
index e6ea25a9f6..f90560c756 100644
--- a/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FTPNamespaceHandler.java
+++ b/spring-integration-ftp/src/main/java/org/springframework/integration/ftp/config/FTPNamespaceHandler.java
@@ -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);
             }
 
diff --git a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.0.xsd b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.0.xsd
index 6dd332feff..cf013408ef 100644
--- a/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.0.xsd
+++ b/spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-2.0.xsd
@@ -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>
\ No newline at end of file
+</xsd:schema>
diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/InboundFTPFileServiceActivator.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/InboundFTPFileServiceActivator.java
index 8c1691c5d4..4f436d52cd 100644
--- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/InboundFTPFileServiceActivator.java
+++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/InboundFTPFileServiceActivator.java
@@ -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;
 
 
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 f0986409a8..a5193c5af6 100644
--- a/spring-integration-ftp/src/test/resources/inbound-ftp-context.xml
+++ b/spring-integration-ftp/src/test/resources/inbound-ftp-context.xml
@@ -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>
\ No newline at end of file
+</beans>