(S)FTP MaxFetchSize Doc Polishing and DSL

Also polish programming tips and tricks docs.

* Fix typos in docs

spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileInboundChannelAdapterSpec.java

@@ -232,6 +232,19 @@ public abstract class RemoteFileInboundChannelAdapterSpec<F, S extends RemoteFil
 		return _this();
 	}
 
+	/**
+	 * Specify the maximum number of remote files that will be fetched on each fetch
+	 * attempt. A small number is recommended when multiple application instances are
+	 * running, to avoid one instance from "grabbing" all the files.
+	 * @param maxFetchSize the max fetch size.
+	 * @return the spec.
+	 * @see org.springframework.integration.endpoint.MessageSourceManagement#setMaxFetchSize(int)
+	 */
+	public S maxFetchSize(int maxFetchSize) {
+		this.target.setMaxFetchSize(maxFetchSize);
+		return _this();
+	}
+
 	@Override
 	public Map<Object, String> getComponentsToRegister() {
 		Map<Object, String> componentsToRegister = new LinkedHashMap<>();

spring-integration-file/src/main/java/org/springframework/integration/file/dsl/RemoteFileStreamingInboundChannelAdapterSpec.java

@@ -126,6 +126,19 @@ public abstract class RemoteFileStreamingInboundChannelAdapterSpec<F,
 		return filter(this.expressionFileListFilter);
 	}
 
+	/**
+	 * Specify the maximum number of remote files that will be fetched on each fetch
+	 * attempt. A small number is recommended when multiple application instances are
+	 * running, to avoid one instance from "grabbing" all the files.
+	 * @param maxFetchSize the max fetch size.
+	 * @return the spec.
+	 * @see org.springframework.integration.endpoint.MessageSourceManagement#setMaxFetchSize(int)
+	 */
+	public S maxFetchSize(int maxFetchSize) {
+		this.target.setMaxFetchSize(maxFetchSize);
+		return _this();
+	}
+
 	@Override
 	public Map<Object, String> getComponentsToRegister() {
 		if (this.expressionFileListFilter != null) {

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/dsl/FtpTests.java

@@ -17,6 +17,7 @@
 package org.springframework.integration.ftp.dsl;
 
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.instanceOf;
 import static org.hamcrest.Matchers.isOneOf;
 import static org.junit.Assert.assertEquals;
@@ -37,10 +38,12 @@ import org.junit.Test;
 import org.junit.runner.RunWith;
 
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.IntegrationMessageHeaderAccessor;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.config.EnableIntegration;
+import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.dsl.Pollers;
@@ -52,8 +55,11 @@ import org.springframework.integration.file.remote.RemoteFileTemplate;
 import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
 import org.springframework.integration.file.support.FileExistsMode;
 import org.springframework.integration.ftp.FtpTestSupport;
+import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizingMessageSource;
+import org.springframework.integration.ftp.inbound.FtpStreamingMessageSource;
 import org.springframework.integration.ftp.session.FtpRemoteFileTemplate;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.test.annotation.DirtiesContext;
@@ -72,12 +78,16 @@ public class FtpTests extends FtpTestSupport {
 	@Autowired
 	private IntegrationFlowContext flowContext;
 
+	@Autowired
+	private ApplicationContext context;
+
 	@Test
 	public void testFtpInboundFlow() {
 		QueueChannel out = new QueueChannel();
 		IntegrationFlow flow = IntegrationFlows.from(Ftp.inboundAdapter(sessionFactory())
 						.preserveTimestamp(true)
 						.remoteDirectory("ftpSource")
+						.maxFetchSize(10)
 						.regexFilter(".*\\.txt$")
 						.localFilename(f -> f.toUpperCase() + ".a")
 						.localDirectory(getTargetLocalDirectory()),
@@ -111,6 +121,8 @@ public class FtpTests extends FtpTestSupport {
 		file = (File) payload;
 		assertEquals(" FTPSOURCE1.TXT.a", file.getName());
 
+		MessageSource<?> source = context.getBean(FtpInboundFileSynchronizingMessageSource.class);
+		assertThat(TestUtils.getPropertyValue(source, "maxFetchSize"), equalTo(10));
 		registration.destroy();
 	}
 
@@ -120,6 +132,7 @@ public class FtpTests extends FtpTestSupport {
 		StandardIntegrationFlow flow = IntegrationFlows.from(
 				Ftp.inboundStreamingAdapter(new FtpRemoteFileTemplate(sessionFactory()))
 						.remoteDirectory("ftpSource")
+						.maxFetchSize(11)
 						.regexFilter(".*\\.txt$"),
 				e -> e.id("ftpInboundAdapter").poller(Pollers.fixedDelay(100)))
 				.channel(out)
@@ -137,6 +150,8 @@ public class FtpTests extends FtpTestSupport {
 		assertThat(message.getHeaders().get(FileHeaders.REMOTE_FILE), isOneOf(" ftpSource1.txt", "ftpSource2.txt"));
 		new IntegrationMessageHeaderAccessor(message).getCloseableResource().close();
 
+		MessageSource<?> source = context.getBean(FtpStreamingMessageSource.class);
+		assertThat(TestUtils.getPropertyValue(source, "maxFetchSize"), equalTo(11));
 		registration.destroy();
 	}
 

src/reference/asciidoc/aggregator.adoc

@@ -33,7 +33,7 @@ This default strategy may be overridden by providing a reference to a custom `Re
 
 The Aggregation API consists of a number of classes:
 
-* The interface `MessageGroupProcessor`, and its subclasses:`MethodInvokingAggregatingMessageGroupProcessor` and `ExpressionEvaluatingMessageGroupProcessor`
+* The interface `MessageGroupProcessor`, and its subclasses: `MethodInvokingAggregatingMessageGroupProcessor` and `ExpressionEvaluatingMessageGroupProcessor`
 
 * The `ReleaseStrategy` interface and its default implementation `SimpleSequenceSizeReleaseStrategy`
 

src/reference/asciidoc/amqp.adoc

@@ -454,7 +454,7 @@ public class AmqpJavaApplication {
 
 By default the inbound endpoints use acknowledge mode `AUTO`, which means the container automatically _acks_ the message when the downstream integration flow completes (or a message is handed off to another thread using a `QueueChannel` or `ExecutorChannel`).
 Setting the mode to `NONE` configures the consumer such that acks are not used at all (the broker automatically acks the message as soon as it is sent).
-Setting the mode to`MANUAL` allows user code to ack the message at some other point during processing.
+Setting the mode to `MANUAL` allows user code to ack the message at some other point during processing.
 To support this, with this mode, the endpoints provide the `Channel` and `deliveryTag` in the `amqp_channel` and `amqp_deliveryTag` headers respectively.
 
 You can perform any valid rabbit command on the `Channel` but, generally, only `basicAck` and `basicNack` (or `basicReject`) would be used.

src/reference/asciidoc/changes-2.1-2.2.adoc

@@ -153,7 +153,7 @@ This allows a Spring Integration application to be shut down in an orderly manne
 [[x2.2-jms-og]]
 ===== JMS Oubound Gateway Improvements
 
-The JMS Outbound Gateway can now be configured to use a`MessageListener` container to receive replies.
+The JMS Outbound Gateway can now be configured to use a `MessageListener` container to receive replies.
 This can improve performance of the gateway.
 
 [[x2.2-o-t-j-t]]

src/reference/asciidoc/changes-2.2-3.0.adoc

@@ -71,7 +71,7 @@ See <<mongodb>> for more information.
 ===== Syslog Support
 
 Building on the 2.2 `SyslogToMapTransformer` Spring Integration 3.0 now introduces `UDP` and `TCP` inbound channel adapters especially tailored for receiving SYSLOG messages.
-For more information, see<<syslog>>.
+For more information, see <<syslog>>.
 
 [[x3.0-tail]]
 ===== 'Tail' Support
@@ -188,7 +188,7 @@ For more information see <<aggregator-config>>.
 
 New `FileListFilter` s that use a persistent `MetadataStore` are now available.
 These can be used to prevent duplicate files after a system restart.
-See<<file-reading>>, <<ftp-inbound>>, and <<sftp-inbound>> for more information.
+See <<file-reading>>, <<ftp-inbound>>, and <<sftp-inbound>> for more information.
 
 [[x3.0-scripting-variables]]
 ===== Scripting Support: Variables Changes

src/reference/asciidoc/changes-3.0-4.0.adoc

@@ -189,7 +189,7 @@ See <<jms-outbound-channel-adapter>>.
 
 The JMS inbound channel adapter now supports the `session-transacted` attribute (default false).
 Previously, you had to inject a customized `JmsTemplate` to use transactions (the adapter allowed 'transacted' in the acknowledgeMode which was incorrect, and didn't work; this value is no longer allowed).
-See<<jms-inbound-channel-adapter>>.
+See <<jms-inbound-channel-adapter>>.
 
 [[x4.0-datatype-channel]]
 ===== Datatype Channels

src/reference/asciidoc/changes-4.0-4.1.adoc

@@ -40,7 +40,7 @@ See <<idempotent-receiver>> and their JavaDocs for more information.
 [[x4.1-BoonJsonObjectMapper]]
 ===== BoonJsonObjectMapper
 
-The _Boon_`JsonObjectMapper` is now provided for the JSON transformers.
+The _Boon_ `JsonObjectMapper` is now provided for the JSON transformers.
 See <<transformer>> for more information.
 
 [[x4.1-redis-queue-gateways]]

src/reference/asciidoc/configuration.adoc

@@ -605,7 +605,7 @@ Together with the existing Spring Container logic, the Messaging Endpoint bean,
 ==== Creating a Bridge with Annotations
 
 Starting with _version 4.0_, the Messaging Annotation and Java configuration provides `@BridgeFrom` and `@BridgeTo` `@Bean` method annotations to mark `MessageChannel` beans in `@Configuration` classes.
-This is just for completeness, providing a convenient mechanism to declare a`BridgeHandler` and its Message Endpoint configuration:
+This is just for completeness, providing a convenient mechanism to declare a `BridgeHandler` and its Message Endpoint configuration:
 [source,java]
 ----
 @Bean

src/reference/asciidoc/endpoint.adoc

@@ -3,7 +3,7 @@
 
 The first part of this chapter covers some background theory and reveals quite a bit about the underlying API that drives Spring Integration's various messaging components.
 This information can be helpful if you want to really understand what's going on behind the scenes.
-However, if you want to get up and running with the simplified namespace-based configuration of the various elements, feel free to skip ahead to<<endpoint-namespace>> for now.
+However, if you want to get up and running with the simplified namespace-based configuration of the various elements, feel free to skip ahead to <<endpoint-namespace>> for now.
 
 As mentioned in the overview, Message Endpoints are responsible for connecting the various messaging components to channels.
 Over the next several chapters, you will see a number of different components that consume Messages.
@@ -218,7 +218,8 @@ If a `PollingConsumer` is used, this atribute will default to _-1_.
 However, if a `SourcePollingChannelAdapter` is used, then the `max-messages-per-poll` attribute defaults to _1_.
 
 
-<8> Value is set on the underlying class `PollerMetadata`_Optional_.
+<8> Value is set on the underlying class `PollerMetadata`.
+_Optional_.
 If not specified it defaults to 1000 (milliseconds).
 
 
@@ -244,7 +245,7 @@ For hourly, daily, and monthly settings, consider using a `cron` trigger instead
 
 <12> Reference to any spring configured bean which implements the `org.springframework.scheduling.Trigger` interface.
 _Optional_.
-However, if this attribute is set, none of the following attributes must be specified:`fixed-delay`, `fixed-rate`, `cron`, `ref`.
+However, if this attribute is set, none of the following attributes must be specified: `fixed-delay`, `fixed-rate`, `cron`, `ref`.
 
 
 <13> Allows to specify extra AOP Advices to handle additional cross cutting concerns.

src/reference/asciidoc/filter.adoc

@@ -36,7 +36,7 @@ In combination with the namespace and SpEL, very powerful filters can be configu
 ===== Configuring a Filter with XML
 
 The <filter> element is used to create a Message-selecting endpoint.
-In addition to "`input-channel` and `output-channel` attributes, it requires a `ref`.
+In addition to `input-channel` and `output-channel` attributes, it requires a `ref`.
 The `ref` may point to a `MessageSelector` implementation:
 [source,xml]
 ----

src/reference/asciidoc/ftp.adoc

@@ -258,7 +258,7 @@ So, the root object of the SpEL Evaluation Context is the original name of the r
 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.
+Use `max-fetch-size` for this purpose; a negative value (default) means no limit and all matching files will be retrieved; see <<ftp-max-fetch>> for more information.
 
 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.
 
@@ -554,7 +554,7 @@ If there is a requirement to allow duplicates, the `AcceptAllFileListFilter` can
 Any other use-cases can be reached via `CompositeFileListFilter` (or `ChainFileListFilter`).
 The java configuration below shows one technique to remove the remote file after processing, avoiding duplicates.
 
-Use the `max-fetch-size` attribute to limit the number of files fetched on each poll when a fetch is necessary; set to 1 and use a persistent filter when running in a clustered environment.
+Use the `max-fetch-size` attribute to limit the number of files fetched on each poll when a fetch is necessary; set to 1 and use a persistent filter when running in a clustered environment; see <<ftp-max-fetch>> for more information.
 
 The adapter puts the remote directory and file name in headers `FileHeaders.REMOTE_DIRECTORY` and `FileHeaders.REMOTE_FILE` respectively.
 Starting with _version 5.0_, additional remote file information, represented in JSON by default, is provided in the `FileHeaders.REMOTE_FILE_INFO` header.
@@ -617,6 +617,27 @@ public class FtpJavaApplication {
 
 Notice that, in this example, the message handler downstream of the transformer has an advice that removes the remote file after processing.
 
+[[ftp-max-fetch]]
+=== Inbound Channel Adapters: Controlling Remote File Fetching
+
+There are two properties that should be considered when configuring inbound channel adapters.
+`max-messages-per-poll`, as with all pollers, can be used to limit the number of messages emitted on each poll (if more than the configured value are ready).
+`max-fetch-size` (since _version 5.0_) can limit the number of files retrieved from the remote server at a time.
+
+The following scenarios assume the starting state is an empty local directory.
+
+* `max-messages-per-poll=2` and `max-fetch-size=1`, the adapter will fetch one file, emit it, fetch the next file, emit it; then sleep until the next poll.
+* `max-messages-per-poll=2` and `max-fetch-size=2`), the adapter will fetch both files, then emit each one.
+* `max-messages-per-poll=2` and `max-fetch-size=4`, the adapter will fetch up to 4 files (if available) and emit the first two (if there are at least two); the next two files will be emitted on the next poll.
+* `max-messages-per-poll=2` and `max-fetch-size` not specified, the adapter will fetch all remote files and emit the first two (if there are at least two); the subsequent files will be emitted on subsequent polls (2-at-a-time); when all are consumed, the remote fetch will be attempted again, to pick up any new files.
+
+IMPORTANT: When deploying multiple instances of an application, a small `max-fetch-size` is recommended to avoid one instance "grabbing" all the files and starving other instances.
+
+Another use for `max-fetch-size` is if you want to stop fetching remote files, but continue to process files that have already been fetched.
+Setting the `maxFetchSize` property on the `MessageSource` (programmatically, via JMX, or via a <<control-bus, control bus>>) effectively stops the adapter from fetching more files, but allows the poller to continue to emit messages for files that have previously been fetched.
+If the poller is active when the property is changed, the change will take effect on the next poll.
+
+
 [[ftp-outbound]]
 === FTP Outbound Channel Adapter
 
@@ -902,7 +923,7 @@ The remote file is NOT deleted if the transfer is ignored because the `FileExist
 
 The message payload resulting from an _mget_ operation is a `List<File>` object - a List of File objects, each representing a retrieved file.
 
-IMPORTANT: Startng with _version 5.0_, if the `FileExistsMode` is `IGNORE`, the payload of the output message will no longer contain files that were not fetched due to the file already existing.
+IMPORTANT: Starting with _version 5.0_, if the `FileExistsMode` is `IGNORE`, the payload of the output message will no longer contain files that were not fetched due to the file already existing.
 Previously, the array contained all files, including those that already existed.
 
 The expression used to determine the remote path should produce a result that ends with `*` - e.g. `foo/*` will fetch the complete tree under `foo`.
@@ -927,7 +948,7 @@ The `-dirs` option is not allowed (the recursive mget uses the recursive `ls` to
 Typically, you would use the `#remoteDirectory` variable in the `local-directory-expression` so that the remote directory structure is retained locally.
 =====
 
-Starting with _version 5.0_, the `FtpSimplePatternFileListFilter` and `FtpRegexPatternFileListFilter` can be configured to always pass directories by setting the `alwaysAcceptDirectorties` to `true`.
+Starting with _version 5.0_, the `FtpSimplePatternFileListFilter` and `FtpRegexPatternFileListFilter` can be configured to always pass directories by setting the `alwaysAcceptDirectories` to `true`.
 This allows recursion for a simple pattern; examples follow:
 
 [source, xml]

src/reference/asciidoc/gateway.adoc

@@ -174,7 +174,7 @@ In the second case (or the first when the argument for parameter `foo` is a `Map
 This can generally be resolved using a `payload-expression`, a `@Payload` annotation and/or a `@Headers` annotation.
 
 Alternatively, and whenever the conventions break down, you can take the entire responsibility for mapping the method calls to messages.
-To do this, implement an`MethodArgsMessageMapper` and provide it to the `<gateway/>` using the `mapper` attribute.
+To do this, implement an `MethodArgsMessageMapper` and provide it to the `<gateway/>` using the `mapper` attribute.
 The mapper maps a `MethodArgsHolder`, which is a simple class wrapping the `java.reflect.Method` instance and an `Object[]` containing the arguments.
 When providing a custom mapper, the `default-payload-expression` attribute and `<default-header/>` elements are not allowed on the gateway; similarly, the `payload-expression` attribute and `<header/>` elements are not allowed on any `<method/>` elements.
 

src/reference/asciidoc/ip.adoc

@@ -712,7 +712,7 @@ This can be a custom `(de)Serializer`, which would typically be needed if the pe
 A `MapJsonSerializer` is provided that will convert a Map to/from JSON.
 This uses a Spring Integration `JsonObjectMapper` to perform this function.
 You can provide a custom `JsonObjectMapper` if needed.
-By default, the serializer inserts a linefeed`0x0a` character between objects.
+By default, the serializer inserts a linefeed `0x0a` character between objects.
 See the JavaDocs for more information.
 
 NOTE: At the time of writing, the `JsonObjectMapper` uses whichever version of `Jackson` is on the classpath.

src/reference/asciidoc/mail.adoc

@@ -313,7 +313,7 @@ Again, below notes are based on GMAIL.
 
 With Java Mail 1.4.1 if `mail.imaps.timeout` property is set for a relatively short period of time (e.g., ~ 5 min) then `IMAPFolder.idle()` will throw `FolderClosedException` after this timeout.
 However if this property is not set (should be indefinite) the behavior that was observed is that `IMAPFolder.idle()` method never returns nor it throws an exception.
-It will however reconnect automatically if connection was lost for a short period of time (e.g., under 10 min), but if connection was lost for a long period of time (e.g., over 10 min), then`IMAPFolder.idle()` will not throw `FolderClosedException` nor it will re-establish connection and will remain in the blocked state indefinitely, thus leaving you no possibility to reconnect without restarting the adapter.
+It will however reconnect automatically if connection was lost for a short period of time (e.g., under 10 min), but if connection was lost for a long period of time (e.g., over 10 min), then `IMAPFolder.idle()` will not throw `FolderClosedException` nor it will re-establish connection and will remain in the blocked state indefinitely, thus leaving you no possibility to reconnect without restarting the adapter.
 So the only way to make re-connect to work with Java Mail 1.4.1 is to set `mail.imaps.timeout` property explicitly to some value, but it also means that such value should be relatively short (under 10 min) and the connection should be re-established relatively quickly.
 Again, it may be different with other providers.
 With Java Mail 1.4.3 there was significant improvements to the API ensuring that there will always be a condition which will force `IMAPFolder.idle()` method to return via `StoreClosedException` or `FolderClosedException` or simply return, thus allowing us to proceed with auto-reconnect.

src/reference/asciidoc/overview.adoc

@@ -312,8 +312,8 @@ The affected modules are
 ==== Finding Class Names for Java and DSL Configuration
 
 With XML configuration and Spring Integration Namespace support, the XML Parsers hide how
-target beans are built and wired together.
-For Java & Annotation Configuration, it is important to understand the Framework API for the target end-user
+target beans are declared and wired together.
+For Java & Annotation Configuration, it is important to understand the Framework API for target end-user
 applications.
 
 The first class citizens for EIP implementation are `Message`, `Channel` and `Endpoint` (see <<overview-components>>
@@ -336,14 +336,14 @@ its main implementations are:
 Using Messaging Annotations and/or Java DSL, you shouldn't worry about these components, because the Framework produces
 them automatically via appropriate annotations and `BeanPostProcessor` s.
 When building components manually, the `ConsumerEndpointFactoryBean` should be used to help to determine the target
-`AbstractEndpoint` implementation based on the provided `inputChannel` property.
+`AbstractEndpoint` consumer implementation to create, based on the provided `inputChannel` property.
 
-On the other hand, the `ConsumerEndpointFactoryBean` exhibits an another first class citizens in the Framework -
+On the other hand, the `ConsumerEndpointFactoryBean` delegates to an another first class citizen in the Framework -
 `org.springframework.messaging.MessageHandler`.
-The goal of the implementation of this class is to _handle the message consumed by the endpoint from the channel_.
+The goal of the implementation of this interface is to _handle the message consumed by the endpoint from the channel_.
 All EIP components in Spring Integration are `MessageHandler` implementations,
 e.g. `AggregatingMessageHandler`, `MessageTransformingHandler`, `AbstractMessageSplitter` etc.; as well as the target
-protocol outbound adapters are implementations, too, e.g. `FileWritingMessageHandler`,
+protocol outbound adapters are implementations too, e.g. `FileWritingMessageHandler`,
 `HttpRequestExecutingMessageHandler`, `AbstractMqttMessageHandler` etc.
 When you develop Spring Integration applications with Java & Annotation Configuration, you should take a look into the
 Spring Integration module to find an appropriate `MessageHandler` implementation to be used for the `@ServiceActivator`
@@ -367,8 +367,8 @@ public MessageHandler sendChatMessageHandler(XMPPConnection xmppConnection) {
 
 The `MessageHandler` implementations represent the _outbound_ and _processing_ part of the message flow.
 
-The _inbound_ message flow side has its own components, which are divided to the _polling_ and _listening_ behavior.
-The listening components are pretty simple and typically requires only one target class implementation to be ready to
+The _inbound_ message flow side has its own components, which are divided to _polling_ and _listening_ behaviors.
+The listening (message-driven) components are simple and typically require only one target class implementation to be ready to
 produce messages.
 Listening components can be one-way `MessageProducerSupport` implementations,
 e.g. `AbstractMqttMessageDrivenChannelAdapter` and `ImapIdleChannelAdapter`; and request-reply -
@@ -380,13 +380,13 @@ For example any File based protocol, as an FTP, any data bases (RDBMS or NoSQL)
 
 These inbound endpoints consist with two components: the poller configuration, to initiate the polling task periodically,
 and message source class to read data from the target protocol and produce a message for the downstream integration flow.
-The first class, for poller configuration, is `SourcePollingChannelAdapter`.
-It is one more `AbstractEndpoint` implementation, but especially for the polling purpose for initiating an integration
+The first class, for the poller configuration, is a `SourcePollingChannelAdapter`.
+It is one more `AbstractEndpoint` implementation, but especially for polling to initiate an integration
 flow.
 Typically, with the Messaging Annotations or Java DSL, you shouldn't worry about this class, the Framework produces
-a bean for it, based on the `@InboundChannelAdapter` configuration or Java DSL particular Builder.
+a bean for it, based on the `@InboundChannelAdapter` configuration or a Java DSL Builder spec.
 
-The _message source_ components are more important for the target application development and they all implement
+_Message source_ components are more important for the target application development and they all implement
 the `MessageSource` interface, e.g. `MongoDbMessageSource` and `AbstractTwitterMessageSource`.
 With that in mind, our config for reading data from an RDBMS table with JDBC may look like:
 
@@ -408,9 +408,8 @@ implementation to listen frames on the socket and produce message to the channel
 - `o.s.i.websocket.outbound.WebSocketOutboundMessageHandler` - the one-way
 `AbstractMessageHandler` implementation to convert incoming messages to the appropriate frame and send over websocket.
 
-If you are familiar with Spring Integration XML configuration already, starting with _version 4.3_, we provide in the
-XSD elements definitions the description with the pointer which target classes are used to produce beans for the adapter
-or gateway, for example:
+If you are familiar with Spring Integration XML configuration, starting with _version 4.3_, we provide information in the
+XSD element definitions about which target classes are used to declare beans for the adapter or gateway, for example:
 
 [source,xml]
 ----

src/reference/asciidoc/polling-consumer.adoc

@@ -50,14 +50,14 @@ Starting with _version 4.1_ a `PollSkipAdvice` is provided.
 Pollers use triggers to determine the time of the next poll.
 The `PollSkipAdvice` can be used to suppress (skip) a poll, perhaps because there is some downstream condition that would prevent the message to be processed properly.
 To use this advice, you have to provide it with an implementation of a `PollSkipStrategy`.
-Startng with version 4.2.5, a `SimplePollSkipStrategy` is provided.
+Starting with _version 4.2.5_, a `SimplePollSkipStrategy` is provided.
 Add an instance as a bean to the application context, inject it into a `PollSkipAdvice` and add that to the poller's
 advice chain.
 To skip polling, call `skipPolls()`, to resume polling, call `reset()`.
 _Version 4.2_ added more flexibility in this area - see <<conditional-pollers>>.
 
 This chapter is meant to only give a high-level overview regarding Polling Consumers and how they fit into the concept of message channels - <<channel>> and channel adapters - <<channel-adapter>>.
-For more in-depth information regarding Messaging Endpoints in general and Polling Consumers in particular, please see<<endpoint>>.
+For more in-depth information regarding Messaging Endpoints in general and Polling Consumers in particular, please see <<endpoint>>.
 
 
 [[conditional-pollers]]

src/reference/asciidoc/redis.adoc

@@ -161,8 +161,8 @@ It receives Spring Integration messages and converts them to platform-specific m
 ----
 
 As you can see the configuration is similar to the Redis Inbound Channel Adapter.
-The adapter is implicitly injected with a `RedisConnectionFactory` which was defined with '`redisConnectionFactory`' as its bean name.
-This example also includes the optional, custom `MessageConverter` (the '`testConverter`' bean).
+The adapter is implicitly injected with a `RedisConnectionFactory` which was defined with `redisConnectionFactory` as its bean name.
+This example also includes the optional, custom `MessageConverter` (the `testConverter` bean).
 
 Since _Spring Integration 3.0_, the `<int-redis:outbound-channel-adapter>`, as an alternative to the `topic` attribute, has the `topic-expression` attribute to determine the Redis topic against the Message at runtime.
 These attributes are mutually exclusive.
@@ -339,7 +339,7 @@ As you can see it is a simple bean configuration, and it expects a `RedisConnect
 By default the `RedisMessageStore` will use Java serialization to serialize the Message.
 However if you want to use a different serialization technique (e.g., JSON), you can provide your own serializer via the `valueSerializer` property of the `RedisMessageStore`.
 
-Starting with _version 4.3.10_, the Framework provides Jackson Serializer and Deserializer implementations for `Message`s and `MessageHeaders` - `MessageHeadersJacksonSerializer` and `MessageJacksonDeserializer`, respectively.
+Starting with _version 4.3.10_, the Framework provides Jackson Serializer and Deserializer implementations for `Message` s and `MessageHeaders` - `MessageHeadersJacksonSerializer` and `MessageJacksonDeserializer`, respectively.
 They have to be configured via the `SimpleModule` options for the `ObjectMapper`.
 In addition, `enableDefaultTyping` should be configured on the `ObjectMapper` to add type information for each serialized complex object.
 That type information is then used during deserialization.

src/reference/asciidoc/samples.adoc

@@ -438,7 +438,7 @@ public class Barista {
 
 As you can see from the code excerpt above, the barista methods have different delays (the hot drinks take 5 times as long to prepare).
 This simulates work being completed at different rates.
-When the`CafeDemo` 'main' method runs, it will loop 100 times sending a single hot drink and a single cold drink each time.
+When the `CafeDemo` 'main' method runs, it will loop 100 times sending a single hot drink and a single cold drink each time.
 It actually sends the messages by invoking the 'placeOrder' method on the Cafe interface.
 Above, you will see that the <gateway> element is specified in the configuration file.
 This triggers the creation of a proxy that implements the given 'service-interface' and connects it to a channel.

src/reference/asciidoc/service-activator.adoc

@@ -47,7 +47,7 @@ To determine the reply channel, it will first check if an "output-channel" was p
 
 If the method returns a result and no "output-channel" is defined, the framework will then check the request Message's `replyChannel` header value.
 If that value is available, it will then check its type.
-If it is a`MessageChannel`, the reply message will be sent to that channel.
+If it is a `MessageChannel`, the reply message will be sent to that channel.
 If it is a `String`, then the endpoint will attempt to resolve the channel name to a channel instance.
 If the channel cannot be resolved, then a `DestinationResolutionException` will be thrown.
 It it can be resolved, the Message will be sent there.

src/reference/asciidoc/sftp.adoc

@@ -329,7 +329,7 @@ So, the root object of the SpEL Evaluation Context is the original name of the r
 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.
+Use `max-fetch-size` for this purpose; a negative value (default) means no limit and all matching files will be retrieved; see <<sftp-max-fetch>> for more information.
 
 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.
 
@@ -599,7 +599,7 @@ If there is a requirement to allow duplicates, the `AcceptAllFileListFilter` can
 Any other use-cases can be reached via `CompositeFileListFilter` (or `ChainFileListFilter`).
 The java configuration below shows one technique to remove the remote file after processing, avoiding duplicates.
 
-Use the `max-fetch-size` attribute to limit the number of files fetched on each poll when a fetch is necessary; set to 1 and use a persistent filter when running in a clustered environment.
+Use the `max-fetch-size` attribute to limit the number of files fetched on each poll when a fetch is necessary; set to 1 and use a persistent filter when running in a clustered environment; see <<sftp-max-fetch>> for more information.
 
 The adapter puts the remote directory and file name in headers `FileHeaders.REMOTE_DIRECTORY` and `FileHeaders.REMOTE_FILE` respectively.
 Starting with _version 5.0_, additional remote file information, in JSON, is provided in the `FileHeaders.REMOTE_FILE_INFO` header.
@@ -662,6 +662,27 @@ public class SftpJavaApplication {
 
 Notice that, in this example, the message handler downstream of the transformer has an advice that removes the remote file after processing.
 
+[[sftp-max-fetch]]
+=== Inbound Channel Adapters: Controlling Remote File Fetching
+
+There are two properties that should be considered when configuring inbound channel adapters.
+`max-messages-per-poll`, as with all pollers, can be used to limit the number of messages emitted on each poll (if more than the configured value are ready).
+`max-fetch-size` (since _version 5.0_) can limit the number of files retrieved from the remote server at a time.
+
+The following scenarios assume the starting state is an empty local directory.
+
+* `max-messages-per-poll=2` and `max-fetch-size=1`, the adapter will fetch one file, emit it, fetch the next file, emit it; then sleep until the next poll.
+* `max-messages-per-poll=2` and `max-fetch-size=2`), the adapter will fetch both files, then emit each one.
+* `max-messages-per-poll=2` and `max-fetch-size=4`, the adapter will fetch up to 4 files (if available) and emit the first two (if there are at least two); the next two files will be emitted on the next poll.
+* `max-messages-per-poll=2` and `max-fetch-size` not specified, the adapter will fetch all remote files and emit the first two (if there are at least two); the subsequent files will be emitted on subsequent polls (2-at-a-time); when all are consumed, the remote fetch will be attempted again, to pick up any new files.
+
+IMPORTANT: When deploying multiple instances of an application, a small `max-fetch-size` is recommended to avoid one instance "grabbing" all the files and starving other instances.
+
+Another use for `max-fetch-size` is if you want to stop fetching remote files, but continue to process files that have already been fetched.
+Setting the `maxFetchSize` property on the `MessageSource` (programmatically, via JMX, or via a <<control-bus, control bus>>) effectively stops the adapter from fetching more files, but allows the poller to continue to emit messages for files that have previously been fetched.
+If the poller is active when the property is changed, the change will take effect on the next poll.
+
+
 [[sftp-outbound]]
 === SFTP Outbound Channel Adapter
 
@@ -924,7 +945,7 @@ The remote file is NOT deleted if the transfer is ignored because the `FileExist
 
 The message payload resulting from an _mget_ operation is a `List<File>` object - a List of File objects, each representing a retrieved file.
 
-IMPORTANT: Startng with _version 5.0_, if the `FileExistsMode` is `IGNORE`, the payload of the output message will no longer contain files that were not fetched due to the file already existing.
+IMPORTANT: Starting with _version 5.0_, if the `FileExistsMode` is `IGNORE`, the payload of the output message will no longer contain files that were not fetched due to the file already existing.
 Previously, the array contained all files, including those that already existed.
 
 

src/reference/asciidoc/splitter.adoc

@@ -129,7 +129,7 @@ If you inadvertently reference the same message handler from multiple beans, you
 
 ===== Configuring a Splitter with Annotations
 
-The `@Splitter` annotation is applicable to methods that expect either the`Message` type or the message payload type, and the return values of the method should be a `Collection` of any type.
+The `@Splitter` annotation is applicable to methods that expect either the `Message` type or the message payload type, and the return values of the method should be a `Collection` of any type.
 If the returned values are not actual `Message` objects, then each item will be wrapped in a Message as its payload.
 Each message will be sent to the designated output channel for the endpoint on which the `@Splitter` is defined.
 [source,java]