INT-4095: Support Limiting (S)FTP Files Fetched

JIRA: https://jira.spring.io/browse/INT-4095

Limit the number of remote files fetched on each poll (when it is necessary to fetch files).

Polishing - PR Comments

Polishing - Decouple MaxFetchSize from Poller

Polishing - PR Comments

Schemas and Docs

More Polishing

* Polishing according PR comments

spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractFetchLimitingMessageSource.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2016 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.endpoint;
+
+/**
+ * A message source that can limit the number of remote objects it fetches.
+ *
+ * @author Gary Russell
+ * @since 5.0
+ *
+ */
+public abstract class AbstractFetchLimitingMessageSource<T> extends AbstractMessageSource<T>
+		implements MessageSourceManagement {
+
+	private volatile int maxFetchSize = Integer.MIN_VALUE;
+
+	@Override
+	public void setMaxFetchSize(int maxFetchSize) {
+		this.maxFetchSize = maxFetchSize;
+	}
+
+	@Override
+	public int getMaxFetchSize() {
+		return this.maxFetchSize;
+	}
+
+	@Override
+	protected Object doReceive() {
+		return doReceive(this.maxFetchSize);
+	}
+
+	/**
+	 * Subclasses must implement this method. Typically the returned value will be the
+	 * payload of type T, but the returned value may also be a Message instance whose
+	 * payload is of type T.
+	 * @param maxFetchSize the maximum number of messages to fetch if a fetch is
+	 * necessary.
+	 * @return The value returned.
+	 */
+	protected abstract Object doReceive(int maxFetchSize);
+
+}

spring-integration-core/src/main/java/org/springframework/integration/endpoint/AbstractMessageSource.java

@@ -128,10 +128,13 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
 	}
 
 	@Override
-	@SuppressWarnings("unchecked")
 	public final Message<T> receive() {
+		return buildMessage(doReceive());
+	}
+
+	@SuppressWarnings("unchecked")
+	protected Message<T> buildMessage(Object result) {
 		Message<T> message = null;
-		Object result = this.doReceive();
 		Map<String, Object> headers = this.evaluateHeaders();
 		if (result instanceof Message<?>) {
 			try {

spring-integration-core/src/main/java/org/springframework/integration/endpoint/MessageSourceManagement.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2016 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.endpoint;
+
+import org.springframework.integration.support.management.IntegrationManagedResource;
+import org.springframework.integration.support.management.MessageSourceMetrics;
+import org.springframework.jmx.export.annotation.ManagedAttribute;
+
+/**
+ * Message sources implementing this interface have additional properties that
+ * can be set or examined using JMX.
+ *
+ * @author Gary Russell
+ * @since 5.0
+ *
+ */
+@IntegrationManagedResource
+public interface MessageSourceManagement extends MessageSourceMetrics {
+
+	/**
+	 * Set the maximum number of objects the source should fetch if it is necessary to
+	 * fetch objects. Setting the
+	 * maxFetchSize to 0 disables remote fetching, a negative value indicates no limit.
+	 * @param maxFetchSize the max fetch size; a negative value means unlimited.
+	 */
+	@ManagedAttribute(description = "Maximum objects to fetch")
+	void setMaxFetchSize(int maxFetchSize);
+
+	/**
+	 * Return the max fetch size.
+	 * @return the max fetch size.
+	 * @see #setMaxFetchSize(int)
+	 */
+	@ManagedAttribute
+	int getMaxFetchSize();
+
+}

spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleMessageSourceManagement.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2016 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.support.management;
+
+import org.springframework.context.Lifecycle;
+import org.springframework.integration.endpoint.MessageSourceManagement;
+
+/**
+ * An extension to {@link LifecycleMessageSourceMetrics} for sources that implement {@link MessageSourceManagement}.
+ *
+ * @author Gary Russell
+ * @since 5.0
+ *
+ */
+public class LifecycleMessageSourceManagement extends LifecycleMessageSourceMetrics implements MessageSourceManagement {
+
+	public LifecycleMessageSourceManagement(Lifecycle lifecycle, MessageSourceManagement delegate) {
+		super(lifecycle, delegate);
+	}
+
+	@Override
+	public void setMaxFetchSize(int maxFetchSize) {
+		((MessageSourceManagement) this.delegate).setMaxFetchSize(maxFetchSize);
+	}
+
+	@Override
+	public int getMaxFetchSize() {
+		return ((MessageSourceManagement) this.delegate).getMaxFetchSize();
+	}
+
+}

spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleMessageSourceMetrics.java

@@ -33,7 +33,7 @@ public class LifecycleMessageSourceMetrics implements MessageSourceMetrics, Life
 
 	private final Lifecycle lifecycle;
 
-	private final MessageSourceMetrics delegate;
+	protected final MessageSourceMetrics delegate;
 
 
 	public LifecycleMessageSourceMetrics(Lifecycle lifecycle, MessageSourceMetrics delegate) {

spring-integration-core/src/main/java/org/springframework/integration/support/management/LifecycleTrackableMessageSourceManagement.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2016 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.support.management;
+
+import org.springframework.context.Lifecycle;
+import org.springframework.integration.endpoint.MessageSourceManagement;
+
+/**
+ * An extension to {@link LifecycleTrackableMessageSourceMetrics} for sources
+ * that implement {@link MessageSourceManagement}.
+ *
+ * @author Gary Russell
+ * @since 5.0
+ *
+ */
+public class LifecycleTrackableMessageSourceManagement extends LifecycleTrackableMessageSourceMetrics
+		implements MessageSourceManagement {
+
+	public LifecycleTrackableMessageSourceManagement(Lifecycle lifecycle, MessageSourceManagement delegate) {
+		super(lifecycle, delegate);
+	}
+
+	@Override
+	public void setMaxFetchSize(int maxFetchSize) {
+		((MessageSourceManagement) this.delegate).setMaxFetchSize(maxFetchSize);
+	}
+
+	@Override
+	public int getMaxFetchSize() {
+		return ((MessageSourceManagement) this.delegate).getMaxFetchSize();
+	}
+
+
+}

spring-integration-core/src/main/resources/org/springframework/integration/config/spring-integration-5.0.xsd

@@ -5009,6 +5009,18 @@ See 'SmartLifecycleRoleController'.
 		</xsd:attribute>
 	</xsd:attributeGroup>
 
+	<xsd:attributeGroup name="maxFetchGroup">
+		<xsd:attribute name="max-fetch-size" type="xsd:string">
+			<xsd:annotation>
+				<xsd:documentation>
+When fetching objects from some external resource, limit the number of such objects that
+will be retrieved on each fetch. A negative value (default) indicates no limit; a value
+of zero effectively disables fetching remote objects.
+				</xsd:documentation>
+			</xsd:annotation>
+		</xsd:attribute>
+	</xsd:attributeGroup>
+
 	<xsd:simpleType name="loggingLevel">
 		<xsd:restriction base="xsd:token">
 			<xsd:enumeration value="FATAL" />

spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileInboundChannelAdapterParser.java

@@ -82,6 +82,7 @@ public abstract class AbstractRemoteFileInboundChannelAdapterParser extends Abst
 			synchronizerBuilder.addPropertyValue("localFilenameGeneratorExpression",
 					localFileGeneratorExpressionBuilder.getBeanDefinition());
 		}
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(messageSourceBuilder, element, "max-fetch-size");
 		return messageSourceBuilder.getBeanDefinition();
 	}
 

spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java

@@ -22,6 +22,7 @@ import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.OutputStream;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
@@ -227,6 +228,17 @@ public abstract class AbstractInboundFileSynchronizer<F>
 
 	@Override
 	public void synchronizeToLocalDirectory(final File localDirectory) {
+		synchronizeToLocalDirectory(localDirectory, Integer.MIN_VALUE);
+	}
+
+	@Override
+	public void synchronizeToLocalDirectory(final File localDirectory, final int maxFetchSize) {
+		if (maxFetchSize == 0) {
+			if (this.logger.isDebugEnabled()) {
+				this.logger.debug("Max Fetch Size is zero - fetch to " + localDirectory.getAbsolutePath() + " ignored");
+			}
+			return;
+		}
 		final String remoteDirectory = this.remoteDirectoryExpression.getValue(this.evaluationContext, String.class);
 		try {
 			int transferred = this.remoteFileTemplate.execute(new SessionCallback<F, Integer>() {
@@ -236,6 +248,14 @@ public abstract class AbstractInboundFileSynchronizer<F>
 					F[] files = session.list(remoteDirectory);
 					if (!ObjectUtils.isEmpty(files)) {
 						List<F> filteredFiles = filterFiles(files);
+						if (maxFetchSize >= 0 && filteredFiles.size() > maxFetchSize) {
+							rollbackFromFileToListEnd(filteredFiles, filteredFiles.get(maxFetchSize));
+							List<F> newList = new ArrayList<>(maxFetchSize);
+							for (int i = 0; i < maxFetchSize; i++) {
+								newList.add(filteredFiles.get(i));
+							}
+							filteredFiles = newList;
+						}
 						for (F file : filteredFiles) {
 							try {
 								if (file != null) {
@@ -245,17 +265,11 @@ public abstract class AbstractInboundFileSynchronizer<F>
 								}
 							}
 							catch (RuntimeException e) {
-								if (AbstractInboundFileSynchronizer.this.filter instanceof ReversibleFileListFilter) {
-									((ReversibleFileListFilter<F>) AbstractInboundFileSynchronizer.this.filter)
-											.rollback(file, filteredFiles);
-								}
+								rollbackFromFileToListEnd(filteredFiles, file);
 								throw e;
 							}
 							catch (IOException e) {
-								if (AbstractInboundFileSynchronizer.this.filter instanceof ReversibleFileListFilter) {
-									((ReversibleFileListFilter<F>) AbstractInboundFileSynchronizer.this.filter)
-											.rollback(file, filteredFiles);
-								}
+								rollbackFromFileToListEnd(filteredFiles, file);
 								throw e;
 							}
 						}
@@ -265,6 +279,14 @@ public abstract class AbstractInboundFileSynchronizer<F>
 						return 0;
 					}
 				}
+
+				public void rollbackFromFileToListEnd(List<F> filteredFiles, F file) {
+					if (AbstractInboundFileSynchronizer.this.filter instanceof ReversibleFileListFilter) {
+						((ReversibleFileListFilter<F>) AbstractInboundFileSynchronizer.this.filter)
+								.rollback(file, filteredFiles);
+					}
+				}
+
 			});
 			if (this.logger.isDebugEnabled()) {
 				this.logger.debug(transferred + " files transferred");

spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizingMessageSource.java

@@ -25,7 +25,7 @@ import java.util.regex.Pattern;
 
 import org.springframework.beans.factory.BeanInitializationException;
 import org.springframework.context.Lifecycle;
-import org.springframework.integration.endpoint.AbstractMessageSource;
+import org.springframework.integration.endpoint.AbstractFetchLimitingMessageSource;
 import org.springframework.integration.file.FileReadingMessageSource;
 import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
 import org.springframework.integration.file.filters.CompositeFileListFilter;
@@ -57,7 +57,7 @@ import org.springframework.util.Assert;
  * @author Gary Russell
  */
 public abstract class AbstractInboundFileSynchronizingMessageSource<F>
-		extends AbstractMessageSource<File> implements Lifecycle {
+		extends AbstractFetchLimitingMessageSource<File> implements Lifecycle {
 
 	private volatile boolean running;
 
@@ -182,21 +182,20 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F>
 	/**
 	 * Polls from the file source. If the result is not null, it will be returned.
 	 * If the result is null, it attempts to sync up with the remote directory to populate the file source.
+	 * At most, maxFetchSize files will be fetched.
 	 * Then, it polls the file source again and returns the result, whether or not it is null.
+	 * @param maxFetchSize the maximum files to fetch.
 	 */
 	@Override
-	public final Message<File> doReceive() {
-		Assert.state(this.fileSource != null, "fileSource must not be null");
-		Assert.state(this.synchronizer != null, "synchronizer must not be null");
+	public final Message<File> doReceive(int maxFetchSize) {
 		Message<File> message = this.fileSource.receive();
 		if (message == null) {
-			this.synchronizer.synchronizeToLocalDirectory(this.localDirectory);
+			this.synchronizer.synchronizeToLocalDirectory(this.localDirectory, maxFetchSize);
 			message = this.fileSource.receive();
 		}
 		return message;
 	}
 
-	@SuppressWarnings("unchecked")
 	private FileListFilter<File> buildFilter() {
 		Pattern completePattern = Pattern.compile("^.*(?<!" + this.synchronizer.getTemporaryFileSuffix() + ")$");
 		return new CompositeFileListFilter<File>(Arrays.asList(

spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/InboundFileSynchronizer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2016 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.
@@ -22,10 +22,25 @@ import java.io.File;
  * Strategy for synchronizing from a remote File system to a local directory.
  *
  * @author Mark Fisher
+ * @author Gary Russell
  * @since 2.0
  */
+@FunctionalInterface
 public interface InboundFileSynchronizer {
 
+	/**
+	 * Synchronize all available files to the local directory;
+	 * @param localDirectory the directory.
+	 */
 	void synchronizeToLocalDirectory(File localDirectory);
 
+	/**
+	 * Synchronize up to maxFetchSize files to the local directory;
+	 * @param localDirectory the directory.
+	 * @param maxFetchSize the maximum files to fetch.
+	 */
+	default void synchronizeToLocalDirectory(File localDirectory, int maxFetchSize) {
+		synchronizeToLocalDirectory(localDirectory);
+	}
+
 }

spring-integration-file/src/test/java/org/springframework/integration/file/remote/synchronizer/AbstractRemoteFileSynchronizerTests.java

@@ -26,11 +26,13 @@ import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
+import java.util.UUID;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.junit.Test;
 
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.integration.file.filters.AcceptOnceFileListFilter;
 import org.springframework.integration.file.remote.session.Session;
 import org.springframework.integration.file.remote.session.SessionFactory;
@@ -94,6 +96,81 @@ public class AbstractRemoteFileSynchronizerTests {
 		sync.close();
 	}
 
+	@Test
+	public void testMaxFetchSizeSynchronizer() throws Exception {
+		final AtomicInteger count = new AtomicInteger();
+		AbstractInboundFileSynchronizer<String> sync = createLimitingSynchronizer(count);
+
+		sync.synchronizeToLocalDirectory(mock(File.class), 1);
+		assertEquals(1, count.get());
+		sync.synchronizeToLocalDirectory(mock(File.class), 1);
+		assertEquals(2, count.get());
+		sync.synchronizeToLocalDirectory(mock(File.class), 1);
+		assertEquals(3, count.get());
+		sync.close();
+	}
+
+	@Test
+	public void testMaxFetchSizeSource() throws Exception {
+		final AtomicInteger count = new AtomicInteger();
+		AbstractInboundFileSynchronizer<String> sync = createLimitingSynchronizer(count);
+
+		AbstractInboundFileSynchronizingMessageSource<String> source =
+				new AbstractInboundFileSynchronizingMessageSource<String>(sync) {
+
+			@Override
+			public String getComponentType() {
+				return "foo";
+			}
+
+		};
+		source.setLocalDirectory(new File(System.getProperty("java.io.tmpdir") + File.separator + UUID.randomUUID()));
+		source.setAutoCreateLocalDirectory(true);
+		source.setBeanFactory(mock(BeanFactory.class));
+		source.setMaxFetchSize(1);
+		source.afterPropertiesSet();
+
+		source.receive();
+		assertEquals(1, count.get());
+		sync.synchronizeToLocalDirectory(mock(File.class), 1);
+		source.receive();
+		sync.synchronizeToLocalDirectory(mock(File.class), 1);
+		source.receive();
+		sync.close();
+	}
+
+	private AbstractInboundFileSynchronizer<String> createLimitingSynchronizer(final AtomicInteger count) {
+		SessionFactory<String> sf = new StringSessionFactory();
+		AbstractInboundFileSynchronizer<String> sync = new AbstractInboundFileSynchronizer<String>(sf) {
+
+			@Override
+			protected boolean isFile(String file) {
+				return true;
+			}
+
+			@Override
+			protected String getFilename(String file) {
+				return file;
+			}
+
+			@Override
+			protected long getModified(String file) {
+				return 0;
+			}
+
+			@Override
+			protected void copyFileToLocalDirectory(String remoteDirectoryPath, String remoteFile, File localDirectory,
+					Session<String> session) throws IOException {
+				count.incrementAndGet();
+			}
+
+		};
+		sync.setFilter(new AcceptOnceFileListFilter<String>());
+		sync.setRemoteDirectory("foo");
+		sync.setBeanFactory(mock(BeanFactory.class));
+		return sync;
+	}
+
 	private class StringSessionFactory implements SessionFactory<String> {
 
 		@Override

spring-integration-ftp/src/main/resources/org/springframework/integration/ftp/config/spring-integration-ftp-5.0.xsd

@@ -164,6 +164,7 @@
 						</xsd:annotation>
 					</xsd:attribute>
 					<xsd:attributeGroup ref="tempSuffixGroup" />
+					<xsd:attributeGroup ref="integration:maxFetchGroup" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests-context.xml

@@ -27,6 +27,7 @@
 				local-filename-generator-expression="#this.toUpperCase() + '.a' + @fooString"
 				comparator="comparator"
 				temporary-file-suffix=".foo"
+				max-fetch-size="42"
 				local-filter="acceptAllFilter"
 				remote-directory-expression="'foo/bar'">
 			<int:poller fixed-rate="1000">

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java

@@ -122,6 +122,7 @@ public class FtpInboundChannelAdapterParserTests {
 			}
 		});
 		assertEquals("FOO.afoo", genMethod.get().invoke(fisync, "foo"));
+		assertEquals(42, inbound.getMaxFetchSize());
 	}
 
 	@Test
@@ -135,6 +136,7 @@ public class FtpInboundChannelAdapterParserTests {
 		assertEquals("/", remoteFileSeparator);
 		assertEquals("foo/bar", TestUtils.getPropertyValue(fisync, "remoteDirectoryExpression", Expression.class)
 				.getExpressionString());
+		assertEquals(Integer.MIN_VALUE, TestUtils.getPropertyValue(simpleAdapterWithCachedSessions, "source.maxFetchSize"));
 	}
 
 	@Test

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/inbound/FtpInboundRemoteFileSystemSynchronizerTests.java

@@ -35,6 +35,14 @@ import java.util.Calendar;
 import java.util.Collection;
 import java.util.List;
 
+import org.apache.commons.net.ftp.FTPClient;
+import org.apache.commons.net.ftp.FTPFile;
+import org.hamcrest.Matchers;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.expression.Expression;
 import org.springframework.expression.ExpressionParser;
@@ -51,14 +59,6 @@ import org.springframework.integration.metadata.PropertiesPersistingMetadataStor
 import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 
-import org.apache.commons.net.ftp.FTPClient;
-import org.apache.commons.net.ftp.FTPFile;
-import org.hamcrest.Matchers;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.mockito.Mockito;
-
 /**
  * @author Oleg Zhurakousky
  * @author Gunnar Hillert
@@ -142,7 +142,7 @@ public class FtpInboundRemoteFileSystemSynchronizerTests {
 		assertNull(nothing);
 
 		// two times because on the third receive (above) the internal queue will be empty, so it will attempt
-		verify(synchronizer, times(2)).synchronizeToLocalDirectory(localDirectoy);
+		verify(synchronizer, times(2)).synchronizeToLocalDirectory(localDirectoy, Integer.MIN_VALUE);
 
 		assertTrue(new File("test/A.TEST.a").exists());
 		assertTrue(new File("test/B.TEST.a").exists());

spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java

@@ -48,14 +48,17 @@ import org.springframework.integration.context.IntegrationContextUtils;
 import org.springframework.integration.context.OrderlyShutdownCapable;
 import org.springframework.integration.core.MessageProducer;
 import org.springframework.integration.endpoint.AbstractEndpoint;
+import org.springframework.integration.endpoint.MessageSourceManagement;
 import org.springframework.integration.gateway.MessagingGatewaySupport;
 import org.springframework.integration.handler.AbstractMessageProducingHandler;
 import org.springframework.integration.history.MessageHistoryConfigurer;
 import org.springframework.integration.support.context.NamedComponent;
 import org.springframework.integration.support.management.IntegrationManagementConfigurer;
 import org.springframework.integration.support.management.LifecycleMessageHandlerMetrics;
+import org.springframework.integration.support.management.LifecycleMessageSourceManagement;
 import org.springframework.integration.support.management.LifecycleMessageSourceMetrics;
 import org.springframework.integration.support.management.LifecycleTrackableMessageHandlerMetrics;
+import org.springframework.integration.support.management.LifecycleTrackableMessageSourceManagement;
 import org.springframework.integration.support.management.LifecycleTrackableMessageSourceMetrics;
 import org.springframework.integration.support.management.MappingMessageRouterManagement;
 import org.springframework.integration.support.management.MessageChannelMetrics;
@@ -993,6 +996,9 @@ public class IntegrationMBeanExporter extends MBeanExporter implements Applicati
 				break;
 			}
 		}
+		if (endpointName == null) {
+			endpoint = null;
+		}
 		if (name != null && endpoint != null && name.startsWith("_org.springframework.integration")) {
 			name = getInternalComponentName(name);
 			source = "internal";
@@ -1040,16 +1046,28 @@ public class IntegrationMBeanExporter extends MBeanExporter implements Applicati
 		if (endpoint instanceof Lifecycle) {
 			// Wrap the monitor in a lifecycle so it exposes the start/stop operations
 			if (endpoint instanceof TrackableComponent) {
-				result = new LifecycleTrackableMessageSourceMetrics((Lifecycle) endpoint, monitor);
+				if (monitor instanceof MessageSourceManagement) {
+					result = new LifecycleTrackableMessageSourceManagement((Lifecycle) endpoint,
+							(MessageSourceManagement) monitor);
+				}
+				else {
+					result = new LifecycleTrackableMessageSourceMetrics((Lifecycle) endpoint, monitor);
+				}
 			}
 			else {
-				result = new LifecycleMessageSourceMetrics((Lifecycle) endpoint, monitor);
+				if (monitor instanceof MessageSourceManagement) {
+					result = new LifecycleMessageSourceManagement((Lifecycle) endpoint,
+							(MessageSourceManagement) monitor);
+				}
+				else {
+					result = new LifecycleMessageSourceMetrics((Lifecycle) endpoint, monitor);
+				}
 			}
 		}
 
 		if (name == null) {
 			name = monitor.toString();
-			source = "handler";
+			source = "source";
 		}
 
 		if (endpointName != null) {

spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageSourceTests.java

@@ -0,0 +1,128 @@
+/*
+ * Copyright 2016 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.monitor;
+
+import static org.hamcrest.Matchers.equalTo;
+import static org.junit.Assert.assertThat;
+
+import java.util.Set;
+
+import javax.management.Attribute;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.annotation.InboundChannelAdapter;
+import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.endpoint.AbstractFetchLimitingMessageSource;
+import org.springframework.integration.jmx.config.EnableIntegrationMBeanExport;
+import org.springframework.integration.scheduling.PollerMetadata;
+import org.springframework.jmx.support.MBeanServerFactoryBean;
+import org.springframework.messaging.PollableChannel;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * @author Gary Russell
+ * @since 5.0
+ *
+ */
+@RunWith(SpringRunner.class)
+public class MessageSourceTests {
+
+	@Autowired
+	private MBeanServer server;
+
+	@Autowired
+	private MaxFetchSource source1;
+
+	@Autowired
+	private MaxFetchSource source2;
+
+	@Test
+	public void testMaxFetch() throws Exception {
+		Set<ObjectName> query = this.server.queryNames(new ObjectName("foo:type=MessageSource,*"), null);
+		for (ObjectName instance : query) {
+			assertThat(this.server.getAttribute(instance, "MaxFetchSize"), equalTo(123));
+			this.server.setAttribute(instance, new Attribute("MaxFetchSize", 456));
+			assertThat(this.server.getAttribute(instance, "MaxFetchSize"), equalTo(456));
+		}
+		assertThat(this.source1.getMaxFetchSize(), equalTo(456));
+		assertThat(this.source2.getMaxFetchSize(), equalTo(456));
+	}
+
+	@Configuration
+	@EnableIntegrationMBeanExport(defaultDomain = "foo")
+	@EnableIntegration
+	public static class Config {
+
+		@Bean
+		public MBeanServerFactoryBean fb() {
+			MBeanServerFactoryBean fb = new MBeanServerFactoryBean();
+			fb.setLocateExistingServerIfPossible(true);
+			return fb;
+		}
+
+		@Bean
+		@InboundChannelAdapter(channel = "out")
+		public MaxFetchSource source1() {
+			return new MaxFetchSource();
+		}
+
+		@Bean
+		public MaxFetchSource source2() { // raw source, no consumer
+			return new MaxFetchSource();
+		}
+
+		@Bean
+		public PollableChannel out() {
+			return new QueueChannel();
+		}
+
+		@Bean(name = PollerMetadata.DEFAULT_POLLER)
+		public PollerMetadata defaultPoller() {
+			PollerMetadata pollerMetadata = new PollerMetadata();
+			pollerMetadata.setTrigger(t -> null);
+			return pollerMetadata;
+		}
+
+	}
+
+	public static class MaxFetchSource extends AbstractFetchLimitingMessageSource<String> {
+
+		{
+			setMaxFetchSize(123);
+		}
+
+		@Override
+		public String getComponentType() {
+			return null;
+		}
+
+		@Override
+		protected Object doReceive(int maxFetchSize) {
+			return null;
+		}
+
+	}
+
+}

spring-integration-sftp/src/main/resources/org/springframework/integration/sftp/config/spring-integration-sftp-5.0.xsd

@@ -168,6 +168,7 @@
 						</xsd:annotation>
 					</xsd:attribute>
 					<xsd:attributeGroup ref="tempSuffixGroup" />
+					<xsd:attributeGroup ref="integration:maxFetchGroup" />
 				</xsd:extension>
 			</xsd:complexContent>
 		</xsd:complexType>

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests-context.xml

@@ -42,6 +42,7 @@
 								  temporary-file-suffix=".bar"
 								  comparator="comparator"
 								  local-filter="acceptAllFilter"
+								  max-fetch-size="42"
 								  delete-remote-files="${delete.remote.files}"
 								  preserve-timestamp="true">
 		<poller fixed-rate="1000">

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserTests.java

@@ -102,6 +102,7 @@ public class InboundChannelAdapterParserTests {
 		Collection<FileListFilter<?>> filters =
 				TestUtils.getPropertyValue(source, "fileSource.scanner.filter.fileFilters", Collection.class);
 		assertThat(filters, hasItem(acceptAllFilter));
+		assertEquals(42, source.getMaxFetchSize());
 		context.close();
 	}
 
@@ -117,6 +118,7 @@ public class InboundChannelAdapterParserTests {
 				.getPropertyValue(autoChannelAdapter, "source.synchronizer.remoteDirectoryExpression", Expression.class)
 				.getExpressionString());
 		assertSame(autoChannel, TestUtils.getPropertyValue(autoChannelAdapter, "outputChannel"));
+		assertEquals(Integer.MIN_VALUE, TestUtils.getPropertyValue(autoChannelAdapter, "source.maxFetchSize"));
 		context.close();
 	}
 

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpInboundRemoteFileSystemSynchronizerTests.java

@@ -137,7 +137,7 @@ public class SftpInboundRemoteFileSystemSynchronizerTests {
 		assertNull(nothing);
 
 		// two times because on the third receive (above) the internal queue will be empty, so it will attempt
-		verify(synchronizer, times(2)).synchronizeToLocalDirectory(localDirectory);
+		verify(synchronizer, times(2)).synchronizeToLocalDirectory(localDirectory, Integer.MIN_VALUE);
 
 		assertTrue(new File("test/a.test").exists());
 		assertTrue(new File("test/b.test").exists());

src/reference/asciidoc/ftp.adoc

@@ -172,6 +172,7 @@ The _FTP Inbound Channel Adapter_ is a special listener that will connect to the
     local-filename-generator-expression="#this.toUpperCase() + '.a'"
     local-filter="myFilter"
     temporary-file-suffix=".writing"
+    max-fetch-size="-1"
     local-directory=".">
     <int:poller fixed-rate="1000"/>
 </int-ftp:inbound-channel-adapter>
@@ -184,6 +185,11 @@ If you want to override this behavior you can set the `local-filename-generator-
 Unlike outbound gateways and adapters where the root object of the SpEL Evaluation Context is a `Message`, this inbound adapter does not yet have the Message at the time of evaluation since that's what it ultimately generates with the transferred file as its payload.
 So, the root object of the SpEL Evaluation Context is the original name of the remote file (String).
 
+The inbound channel adapter first retrieves the file to a local directory and then emits each file according to the poller configuration.
+Starting with _version 5.0_ you can now limit the number of files fetched from the FTP server when new file retrievals are needed.
+This can be beneficial when the target files are very large and/or when running in a clustered system with a persistent file list filter discussed below.
+Use `max-fetch-size` for this purpose; a negative value (default) means no limit and all matching files will be retrieved.
+
 Starting with _Spring Integration 3.0_, you can specify the `preserve-timestamp` attribute (default `false`); when `true`, the local file's modified timestamp will be set to the value retrieved from the server; otherwise it will be set to the current time.
 
 Starting with _version 4.2_, you can specify `remote-directory-expression` instead of `remote-directory`, allowing
@@ -209,7 +215,7 @@ When used with a shared data store (such as `Redis` with the `RedisMetadataStore
 
 The above discussion refers to filtering the files before retrieving them.
 Once the files have been retrieved, an additional filter is applied to the files on the file system.
-By default, this is an`AcceptOnceFileListFilter` which, as discussed, retains state in memory and does not consider the file's modified time.
+By default, this is an `AcceptOnceFileListFilter` which, as discussed, retains state in memory and does not consider the file's modified time.
 Unless your application removes files after processing, the adapter will re-process the files on disk by default after an application restart.
 
 Also, if you configure the `filter` to use a `FtpPersistentAcceptOnceFileListFilter`, and the remote file timestamp changes (causing it to be re-fetched), the default local filter will not allow this new file to be processed.
@@ -367,6 +373,7 @@ public class FtpJavaApplication {
         source.setLocalDirectory(new File("ftp-inbound"));
         source.setAutoCreateLocalDirectory(true);
         source.setLocalFilter(new AcceptOnceFileListFilter<File>());
+        source.setMaxFetchSize(1);
         return source;
     }
 

src/reference/asciidoc/sftp.adoc

@@ -306,6 +306,7 @@ The _SFTP Inbound Channel Adapter_ is a special listener that will connect to th
             local-filename-generator-expression="#this.toUpperCase() + '.a'"
             local-filter="myFilter"
             temporary-file-suffix=".writing"
+            max-fetch-size="-1"
             delete-remote-files="false">
         <int:poller fixed-rate="1000"/>
 </int-sftp:inbound-channel-adapter>
@@ -318,6 +319,11 @@ If you want to override this behavior you can set the `local-filename-generator-
 Unlike outbound gateways and adapters where the root object of the SpEL Evaluation Context is a `Message`, this inbound adapter does not yet have the Message at the time of evaluation since that's what it ultimately generates with the transferred file as its payload.
 So, the root object of the SpEL Evaluation Context is the original name of the remote file (String).
 
+The inbound channel adapter first retrieves the file to a local directory and then emits each file according to the poller configuration.
+Starting with _version 5.0_ you can now limit the number of files fetched from the FTP server when new file retrievals are needed.
+This can be beneficial when the target files are very large and/or when running in a clustered system with a persistent file list filter discussed below.
+Use `max-fetch-size` for this purpose; a negative value (default) means no limit and all matching files will be retrieved.
+
 Starting with _Spring Integration 3.0_, you can specify the `preserve-timestamp` attribute (default `false`); when `true`, the local file's modified timestamp will be set to the value retrieved from the server; otherwise it will be set to the current time.
 
 Starting with _version 4.2_, you can specify `remote-directory-expression` instead of `remote-directory`, allowing
@@ -476,6 +482,7 @@ public class SftpJavaApplication {
         source.setLocalDirectory(new File("ftp-inbound"));
         source.setAutoCreateLocalDirectory(true);
         source.setLocalFilter(new AcceptOnceFileListFilter<File>());
+        source.setMaxFetchSize(1);
         return source;
     }
 

src/reference/asciidoc/whats-new.adoc

@@ -33,3 +33,7 @@ See <<gateway-error-handling>> for more information.
 
 Some inconsistencies with rendering IMAP mail content have been resolved.
 See <<imap-format-important, the note in the Mail-Receiving Channel Adapter Section>> for more information.
+
+==== (S)FTP Changes
+
+The inbound channel adapters now have a property `max-fetch-size` which is used to limit the number of files fetched during a poll when there are no files currently in the local directory.