INT-4491: (S)FTP inbound rotate dirs/servers

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

Add Rotating Server/Directory Polling Advice.

**cherry-pick to 5.0.x**

* Polishing - PR Comments.

* Polishing

* Polishing; revert `KeyDirectory`; WARN about `TaskExecutor` and `MessageSoureMutator`(s).

* More polishing - PR comments

* Apply stashed changes.

* Fix WARN log - the `SyncTaskExecutor` is wrapped.

# Conflicts:
#	spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java
#	src/reference/asciidoc/whats-new.adoc

spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/RotatingServerAdvice.java

@@ -0,0 +1,247 @@
+/*
+ * Copyright 2018 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.file.remote.session;
+
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.integration.aop.AbstractMessageSourceAdvice;
+import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.file.remote.AbstractRemoteFileStreamingMessageSource;
+import org.springframework.integration.file.remote.synchronizer.AbstractInboundFileSynchronizingMessageSource;
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+
+/**
+ * A smart poller advice that rotates across multiple remote servers/directories.
+ *
+ * @author Gary Russell
+ *
+ * @since 5.0.7
+ *
+ */
+public class RotatingServerAdvice extends AbstractMessageSourceAdvice {
+
+	private final RotationPolicy rotationPolicy;
+
+	/**
+	 * Create an instance that rotates to the next server/directory if no message is
+	 * received.
+	 * @param factory the {@link DelegatingSessionFactory}.
+	 * @param keyDirectories a list of {@link KeyDirectory}.
+	 */
+	public RotatingServerAdvice(DelegatingSessionFactory<?> factory, List<KeyDirectory> keyDirectories) {
+		this(factory, keyDirectories, false);
+	}
+
+	/**
+	 * Create an instance that rotates to the next server/directory depending on the fair
+	 * argument.
+	 * @param factory the {@link DelegatingSessionFactory}.
+	 * @param keyDirectories a list of {@link KeyDirectory}.
+	 * @param fair true to rotate on every poll, false to rotate when no message is received.
+	 */
+	public RotatingServerAdvice(DelegatingSessionFactory<?> factory, List<KeyDirectory> keyDirectories, boolean fair) {
+		this(new StandardRotationPolicy(factory, keyDirectories, fair));
+	}
+
+	/**
+	 * Construct an instance that rotates according to the supplied
+	 * {@link RotationPolicy}.
+	 * @param rotationPolicy the policy.
+	 */
+	public RotatingServerAdvice(RotationPolicy rotationPolicy) {
+		Assert.notNull(rotationPolicy, "'rotationPolicy' cannot be null");
+		this.rotationPolicy = rotationPolicy;
+	}
+
+	@Override
+	public boolean beforeReceive(MessageSource<?> source) {
+		this.rotationPolicy.beforeReceive(source);
+		return true;
+	}
+
+	@Override
+	public Message<?> afterReceive(Message<?> result, MessageSource<?> source) {
+		this.rotationPolicy.afterReceive(result != null, source);
+		return result;
+	}
+
+	/**
+	 * Implementations can reconfigure the message source before and/or after
+	 * a poll.
+	 */
+	public interface RotationPolicy {
+
+		/**
+		 * Invoked before the message source receive() method.
+		 * @param source the message source.
+		 */
+		void beforeReceive(MessageSource<?> source);
+
+		/**
+		 * Invoked after the message source receive() method.
+		 * @param messageReceived true if a message was received.
+		 * @param source the message source.
+		 */
+		void afterReceive(boolean messageReceived, MessageSource<?> source);
+
+	}
+
+	/**
+	 * Standard rotation policy; iterates over key/directory pairs; when the end
+	 * is reached, starts again at the beginning. If the fair option is true
+	 * the rotation occurs on every poll, regardless of result. Otherwise rotation
+	 * occurs when the current pair returns no message.
+	 */
+	public static class StandardRotationPolicy implements RotationPolicy {
+
+		protected final Log logger = LogFactory.getLog(getClass());
+
+		private final DelegatingSessionFactory<?> factory;
+
+		private final List<KeyDirectory> keyDirectories = new ArrayList<>();
+
+		private final boolean fair;
+
+		private volatile Iterator<KeyDirectory> iterator;
+
+		private volatile KeyDirectory current;
+
+		private volatile boolean initialized;
+
+		public StandardRotationPolicy(DelegatingSessionFactory<?> factory, List<KeyDirectory> keyDirectories,
+				boolean fair) {
+
+			Assert.notNull(factory, "factory cannot be null");
+			Assert.notNull(keyDirectories, "keyDirectories cannot be null");
+			Assert.isTrue(keyDirectories.size() > 0, "At least one KeyDirectory is required");
+			this.factory = factory;
+			this.keyDirectories.addAll(keyDirectories);
+			this.fair = fair;
+			this.iterator = this.keyDirectories.iterator();
+		}
+
+		protected Iterator<KeyDirectory> getIterator() {
+			return this.iterator;
+		}
+
+		protected void setIterator(Iterator<KeyDirectory> iterator) {
+			this.iterator = iterator;
+		}
+
+		protected boolean isInitialized() {
+			return this.initialized;
+		}
+
+		protected void setInitialized(boolean initialized) {
+			this.initialized = initialized;
+		}
+
+		protected DelegatingSessionFactory<?> getFactory() {
+			return this.factory;
+		}
+
+		protected List<KeyDirectory> getKeyDirectories() {
+			return this.keyDirectories;
+		}
+
+		protected boolean isFair() {
+			return this.fair;
+		}
+
+		@Override
+		public void beforeReceive(MessageSource<?> source) {
+			if (this.fair || !this.initialized) {
+				configureSource(source);
+				this.initialized = true;
+			}
+			if (this.logger.isTraceEnabled()) {
+				this.logger.trace("Next poll is for " + this.current);
+			}
+			this.factory.setThreadKey(this.current.getKey());
+		}
+
+		@Override
+		public void afterReceive(boolean messageReceived, MessageSource<?> source) {
+			if (this.logger.isTraceEnabled()) {
+				this.logger.trace("Poll produced "
+						+ (messageReceived ? "a" : "no")
+						+ " message");
+			}
+			this.factory.clearThreadKey();
+			if (!this.fair && !messageReceived) {
+				configureSource(source);
+			}
+		}
+
+		protected void configureSource(MessageSource<?> source) {
+			Assert.isTrue(source instanceof AbstractInboundFileSynchronizingMessageSource
+					|| source instanceof AbstractRemoteFileStreamingMessageSource,
+				"source must be an AbstractInboundFileSynchronizingMessageSource or a "
+				+ "AbstractRemoteFileStreamingMessageSource");
+			if (!this.iterator.hasNext()) {
+				this.iterator = this.keyDirectories.iterator();
+			}
+			this.current = this.iterator.next();
+			if (source instanceof AbstractRemoteFileStreamingMessageSource) {
+				((AbstractRemoteFileStreamingMessageSource<?>) source).setRemoteDirectory(this.current.getDirectory());
+			}
+			else {
+				((AbstractInboundFileSynchronizingMessageSource<?>) source).getSynchronizer()
+					.setRemoteDirectory(this.current.getDirectory());
+			}
+		}
+
+	}
+
+	/**
+	 * A {@link DelegatingSessionFactory} key/directory pair.
+	 */
+	public static class KeyDirectory {
+
+		private final String key;
+
+		private final String directory;
+
+		public KeyDirectory(String key, String directory) {
+			Assert.notNull(key, "key cannot be null");
+			Assert.notNull(directory, "directory cannot be null");
+			this.key = key;
+			this.directory = directory;
+		}
+
+		public String getKey() {
+			return this.key;
+		}
+
+		public String getDirectory() {
+			return this.directory;
+		}
+
+		@Override
+		public String toString() {
+			return "KeyDirectory [key=" + this.key + ", directory=" + this.directory + "]";
+		}
+
+	}
+
+}

spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/SessionFactoryMapBuilder.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2018 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.file.remote.session;
+
+import org.springframework.integration.support.MapBuilder;
+
+/**
+ * A {@link MapBuilder} to producer a map that maps objects to {@link SessionFactory}s.
+ *
+ * @author Gary Russell
+ * @since 5.0.7
+ *
+ */
+public class SessionFactoryMapBuilder<T> extends MapBuilder<SessionFactoryMapBuilder<T>, Object, SessionFactory<T>> {
+
+}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2017 the original author or authors.
+ * Copyright 2002-2018 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.
@@ -75,38 +75,43 @@ public abstract class AbstractInboundFileSynchronizer<F>
 
 	private final RemoteFileTemplate<F> remoteFileTemplate;
 
-	private volatile EvaluationContext evaluationContext;
+	private EvaluationContext evaluationContext;
 
-	private volatile String remoteFileSeparator = "/";
+	private String remoteFileSeparator = "/";
 
 	/**
 	 * Extension used when downloading files. We change it right after we know it's downloaded.
 	 */
-	private volatile String temporaryFileSuffix = ".writing";
+	private String temporaryFileSuffix = ".writing";
 
-	private volatile Expression localFilenameGeneratorExpression;
+	private Expression localFilenameGeneratorExpression;
 
 	/**
 	 * the path on the remote mount as a String.
 	 */
 	private volatile Expression remoteDirectoryExpression;
 
+	/**
+	 * The current evaluation of the expression.
+	 */
+	private volatile String evaluatedRemoteDirectory;
+
 	/**
 	 * An {@link FileListFilter} that runs against the <em>remote</em> file system view.
 	 */
-	private volatile FileListFilter<F> filter;
+	private FileListFilter<F> filter;
 
 	/**
 	 * Should we <em>delete</em> the remote <b>source</b> files
 	 * after copying to the local directory? By default this is false.
 	 */
-	private volatile boolean deleteRemoteFiles;
+	private boolean deleteRemoteFiles;
 
 	/**
 	 * Should we <em>transfer</em> the remote file <b>timestamp</b>
 	 * to the local file? By default this is false.
 	 */
-	private volatile boolean preserveTimestamp;
+	private boolean preserveTimestamp;
 
 	private BeanFactory beanFactory;
 
@@ -164,6 +169,7 @@ public abstract class AbstractInboundFileSynchronizer<F>
 	 */
 	public void setRemoteDirectory(String remoteDirectory) {
 		this.remoteDirectoryExpression = new LiteralExpression(remoteDirectory);
+		evaluateRemoteDirectory();
 	}
 
 	/**
@@ -182,13 +188,14 @@ public abstract class AbstractInboundFileSynchronizer<F>
 	 * @see #setRemoteDirectoryExpression(Expression)
 	 */
 	public void setRemoteDirectoryExpressionString(String remoteDirectoryExpression) {
-		setRemoteDirectoryExpression(EXPRESSION_PARSER.parseExpression(remoteDirectoryExpression));
+		doSetRemoteDirectoryExpression(EXPRESSION_PARSER.parseExpression(remoteDirectoryExpression));
 	}
 
 
 	protected final void doSetRemoteDirectoryExpression(Expression remoteDirectoryExpression) {
 		Assert.notNull(remoteDirectoryExpression, "'remoteDirectoryExpression' must not be null");
 		this.remoteDirectoryExpression = remoteDirectoryExpression;
+		evaluateRemoteDirectory();
 	}
 
 	/**
@@ -231,9 +238,11 @@ public abstract class AbstractInboundFileSynchronizer<F>
 		if (this.evaluationContext == null) {
 			this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.beanFactory);
 		}
+		evaluateRemoteDirectory();
 		doInit();
 	}
 
+
 	/**
 	 * Subclasses can override to perform initialization - called from
 	 * {@link InitializingBean#afterPropertiesSet()}.
@@ -269,10 +278,12 @@ public abstract class AbstractInboundFileSynchronizer<F>
 			}
 			return;
 		}
-		final String remoteDirectory = this.remoteDirectoryExpression.getValue(this.evaluationContext, String.class);
+		if (this.logger.isTraceEnabled()) {
+			this.logger.trace("Synchronizing " + this.evaluatedRemoteDirectory + " to " + localDirectory);
+		}
 		try {
 			int transferred = this.remoteFileTemplate.execute(session -> {
-				F[] files = session.list(remoteDirectory);
+				F[] files = session.list(this.evaluatedRemoteDirectory);
 				if (!ObjectUtils.isEmpty(files)) {
 					List<F> filteredFiles = filterFiles(files);
 					if (maxFetchSize >= 0 && filteredFiles.size() > maxFetchSize) {
@@ -288,10 +299,9 @@ public abstract class AbstractInboundFileSynchronizer<F>
 
 					for (F file : filteredFiles) {
 						try {
-							if (file != null) {
-								if (!copyFileToLocalDirectory(remoteDirectory, file, localDirectory, session)) {
-									copied--;
-								}
+							if (file != null && !copyFileToLocalDirectory(this.evaluatedRemoteDirectory, file,
+									localDirectory, session)) {
+								copied--;
 							}
 						}
 						catch (RuntimeException e1) {
@@ -328,8 +338,8 @@ public abstract class AbstractInboundFileSynchronizer<F>
 
 	protected boolean copyFileToLocalDirectory(String remoteDirectoryPath, F remoteFile, File localDirectory,
 			Session<F> session) throws IOException {
-		String remoteFileName = this.getFilename(remoteFile);
-		String localFileName = this.generateLocalFileName(remoteFileName);
+		String remoteFileName = getFilename(remoteFile);
+		String localFileName = generateLocalFileName(remoteFileName);
 		String remoteFilePath = remoteDirectoryPath != null
 				? (remoteDirectoryPath + this.remoteFileSeparator + remoteFileName)
 				: remoteFileName;
@@ -438,11 +448,20 @@ public abstract class AbstractInboundFileSynchronizer<F>
 
 	private String generateLocalFileName(String remoteFileName) {
 		if (this.localFilenameGeneratorExpression != null) {
-			return this.localFilenameGeneratorExpression.getValue(this.evaluationContext, remoteFileName, String.class);
+			return this.localFilenameGeneratorExpression.getValue(this.evaluationContext, remoteFileName,
+					String.class);
 		}
 		return remoteFileName;
 	}
 
+	protected void evaluateRemoteDirectory() {
+		if (this.evaluationContext != null) {
+			this.evaluatedRemoteDirectory = this.remoteDirectoryExpression.getValue(this.evaluationContext,
+					String.class);
+			this.evaluationContext.setVariable("remoteDirectory", this.evaluatedRemoteDirectory);
+		}
+	}
+
 	protected abstract boolean isFile(F file);
 
 	protected abstract String getFilename(F file);

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

@@ -164,6 +164,15 @@ public abstract class AbstractInboundFileSynchronizingMessageSource<F>
 		this.scannerExplicitlySet = true;
 	}
 
+	/**
+	 * Return the underlying synchronizer.
+	 * @return the synchronizer.
+	 * @since 5.0.7
+	 */
+	public AbstractInboundFileSynchronizer<F> getSynchronizer() {
+		return this.synchronizer;
+	}
+
 	@Override
 	public void afterPropertiesSet() throws Exception {
 		super.afterPropertiesSet();