From a4f779d09ecdfc908e79bb168524c3916e0241a9 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 24 Oct 2018 15:33:29 -0400 Subject: [PATCH] INT-4547: (S)FTP RFOG MPUT with collection payload JIRA: https://jira.spring.io/browse/INT-4547 * PR Comments; use `MutableMessage` internally. --- .../file/remote/RemoteFileTemplate.java | 2 +- .../AbstractRemoteFileOutboundGateway.java | 23 +++++++++---- .../RemoteFileOutboundGatewayTests.java | 34 +++++++++++++++++++ src/reference/asciidoc/ftp.adoc | 3 +- src/reference/asciidoc/sftp.adoc | 3 +- src/reference/asciidoc/whats-new.adoc | 3 ++ 6 files changed, 58 insertions(+), 10 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java index 3267e5cb1c..2444d56c3f 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java @@ -293,7 +293,7 @@ public class RemoteFileTemplate implements RemoteFileOperations, 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 -> { diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java index d7733a5333..084a258b54 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java +++ b/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 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 { diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java index 003908cca4..eca585f068 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java +++ b/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 sessionFactory = mock(SessionFactory.class); + @SuppressWarnings("unchecked") + Session session = mock(Session.class); + RemoteFileTemplate template = new RemoteFileTemplate(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 written = new AtomicReference(); + doAnswer(invocation -> { + written.set(invocation.getArgument(1)); + return null; + }).when(session).write(any(InputStream.class), anyString()); + List files = new ArrayList<>(); + files.add(tempFolder.newFile("fiz.txt")); + files.add(tempFolder.newFile("buz.txt")); + Message> requestMessage = MessageBuilder.withPayload(files) + .build(); + @SuppressWarnings("unchecked") + List out = (List) 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 { private boolean open; diff --git a/src/reference/asciidoc/ftp.adoc b/src/reference/asciidoc/ftp.adoc index 2fd2c19f2c..741274855b 100644 --- a/src/reference/asciidoc/ftp.adoc +++ b/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 <>. In addition, files in the local directory can be filtered with one of `mput-pattern`, `mput-regex`, `mput-filter`, or `mput-filter-expression`. diff --git a/src/reference/asciidoc/sftp.adoc b/src/reference/asciidoc/sftp.adoc index 66b645ddfd..456787e1ff 100644 --- a/src/reference/asciidoc/sftp.adoc +++ b/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 <> 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`. diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 22f4a0d6cc..a4fee47c75 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -166,6 +166,9 @@ The `CachingSessionFactory` has a new property `testSession` which, when true, c See <> and <> for more information. +The outbound gateway MPUT command now supports a message payload with a collection of files or strings. +See <> and <> for more information. + [[x51.-tcp]] === TCP Support