INT-4547: (S)FTP RFOG MPUT with collection payload

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

* PR Comments; use `MutableMessage` internally.

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

@@ -293,7 +293,7 @@ public class RemoteFileTemplate<F> implements RemoteFileOperations<F>, Initializ
 				"Cannot append when using a temporary file name");
 		Assert.isTrue(!FileExistsMode.REPLACE_IF_MODIFIED.equals(mode),
 				"FilExistsMode.REPLACE_IF_MODIFIED can only be used for local files");
-		final StreamHolder inputStreamHolder = this.payloadToInputStream(message);
+		final StreamHolder inputStreamHolder = payloadToInputStream(message);
 		if (inputStreamHolder != null) {
 			try {
 				return this.execute(session -> {

spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java

@@ -30,6 +30,7 @@ import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.Expression;
@@ -806,16 +807,24 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
 
 	private Object doMput(Message<?> requestMessage) {
 		File file = null;
-		if (requestMessage.getPayload() instanceof File) {
-			file = (File) requestMessage.getPayload();
+		Object payload = requestMessage.getPayload();
+		if (payload instanceof File) {
+			file = (File) payload;
 		}
-		else if (requestMessage.getPayload() instanceof String) {
-			file = new File((String) requestMessage.getPayload());
+		else if (payload instanceof String) {
+			file = new File((String) payload);
 		}
-		else {
-			throw new IllegalArgumentException("Only File or String payloads allowed for 'mput'");
+		else if (!(payload instanceof Collection)) {
+			throw new IllegalArgumentException(
+					"Only File or String payloads (or Collection of File/String) allowed for 'mput', received: "
+							+ payload.getClass());
 		}
-		if (!file.isDirectory()) {
+		if ((payload instanceof Collection)) {
+			return ((Collection<?>) payload).stream()
+					.map(p -> doMput(new MutableMessage<>(p, requestMessage.getHeaders())))
+					.collect(Collectors.toList());
+		}
+		else if (!file.isDirectory()) {
 			return doPut(requestMessage);
 		}
 		else {

spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java

@@ -967,6 +967,40 @@ public class RemoteFileOutboundGatewayTests {
 				equalTo("foo/baz.txt"), equalTo("foo/qux.txt"), equalTo("foo/" + dir1.getName() + "/" + file3.getName())));
 	}
 
+	@Test
+	public void testMputCollection() throws Exception {
+		@SuppressWarnings("unchecked")
+		SessionFactory<TestLsEntry> sessionFactory = mock(SessionFactory.class);
+		@SuppressWarnings("unchecked")
+		Session<TestLsEntry> session = mock(Session.class);
+		RemoteFileTemplate<TestLsEntry> template = new RemoteFileTemplate<TestLsEntry>(sessionFactory);
+		template.setRemoteDirectoryExpression(new LiteralExpression("foo/"));
+		template.setBeanFactory(mock(BeanFactory.class));
+		template.afterPropertiesSet();
+		TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway(template, "mput", "payload");
+		gw.afterPropertiesSet();
+		when(sessionFactory.getSession()).thenReturn(session);
+		final AtomicReference<String> written = new AtomicReference<String>();
+		doAnswer(invocation -> {
+			written.set(invocation.getArgument(1));
+			return null;
+		}).when(session).write(any(InputStream.class), anyString());
+		List<File> files = new ArrayList<>();
+		files.add(tempFolder.newFile("fiz.txt"));
+		files.add(tempFolder.newFile("buz.txt"));
+		Message<List<File>> requestMessage = MessageBuilder.withPayload(files)
+				.build();
+		@SuppressWarnings("unchecked")
+		List<String> out = (List<String>) gw.handleRequestMessage(requestMessage);
+		assertEquals(2, out.size());
+		assertThat(out.get(0),
+				not(equalTo(out.get(1))));
+		assertThat(out.get(0), equalTo("foo/fiz.txt"));
+		assertThat(out.get(1), equalTo("foo/buz.txt"));
+		assertThat(written.get(), equalTo("foo/buz.txt.writing"));
+		verify(session).rename("foo/buz.txt.writing", "foo/buz.txt");
+	}
+
 	abstract static class TestSession implements Session<TestLsEntry> {
 
 		private boolean open;

src/reference/asciidoc/ftp.adoc

@@ -1172,7 +1172,8 @@ The `mput` sends multiple files to the server and supports only one option:
 * `-R`: Recursive.
 Send all files (possibly filtered) in the directory and its subdirectories.
 
-The message payload must be a `java.io.File` that represents a local directory.
+The message payload must be a `java.io.File` (or `String`) that represents a local directory.
+Since version 5.1, a collection of `File` or `String` is also supported.
 
 This command supports the same attributes as the <<ftp-put-command,`put` command>>.
 In addition, files in the local directory can be filtered with one of `mput-pattern`, `mput-regex`, `mput-filter`, or `mput-filter-expression`.

src/reference/asciidoc/sftp.adoc

@@ -1136,7 +1136,8 @@ When configuring the adapter using java, you can use `setChmod(0600)`.
 
 * `-R`: Recursive -- send all files (possibly filtered) in the directory and subdirectories
 
-The message payload must be a `java.io.File` that represents a local directory.
+The message payload must be a `java.io.File` (or `String`) that represents a local directory.
+Since version 5.1, a collection of `File` or `String` is also supported.
 
 The same attributes as the <<sftp-put-command,`put` command>> are supported.
 In addition, you can filter files in the local directory with one of `mput-pattern`, `mput-regex`, `mput-filter`, or `mput-filter-expression`.

src/reference/asciidoc/whats-new.adoc

@@ -166,6 +166,9 @@ The `CachingSessionFactory` has a new property `testSession` which, when true, c
 
 See <<sftp-session-caching>> and <<ftp-session-caching>> for more information.
 
+The outbound gateway MPUT command now supports a message payload with a collection of files or strings.
+See <<sftp-outbound-gateway>> and <<ftp-outbound-gateway>> for more information.
+
 [[x51.-tcp]]
 === TCP Support