diff --git a/applications/source/ftp-source/README.adoc b/applications/source/ftp-source/README.adoc index c37a4660..e98b80ad 100644 --- a/applications/source/ftp-source/README.adoc +++ b/applications/source/ftp-source/README.adoc @@ -92,7 +92,7 @@ $$ftp.supplier.tmp-file-suffix$$:: $$The suffix to use while the transfer is in == Examples ``` -java -jar ftp_source.jar --ftp.remote-dir=foo --file.mode=lines --trigger.fixed-delay=60 --ftp.factory.host=ftpserver \ +java -jar ftp_source.jar --ftp.supplier.remote-dir=foo --file.consumer.mode=lines --ftp.factory.host=ftpserver \ --ftp.factory.username=user --ftp.factory.password=pw --ftp.local-dir=/foo ``` //end::ref-doc[] diff --git a/applications/source/pom.xml b/applications/source/pom.xml index fc1c28f1..980a5caf 100644 --- a/applications/source/pom.xml +++ b/applications/source/pom.xml @@ -22,10 +22,11 @@ mongodb-source tcp-source rabbit-source - websocket-source s3-source + sftp-source twitter-stream-source twitter-search-source twitter-message-source + websocket-source diff --git a/applications/source/sftp-source/README.adoc b/applications/source/sftp-source/README.adoc new file mode 100644 index 00000000..cdf87bae --- /dev/null +++ b/applications/source/sftp-source/README.adoc @@ -0,0 +1,117 @@ +//tag::ref-doc[] += SFTP Source +This source application supports transfer of files using the SFTP protocol. +Files are transferred from the `remote` directory to the `local` directory where the app is deployed. +Messages emitted by the source are provided as a byte array by default. However, this can be +customized using the `--mode` option: + +- *ref* Provides a `java.io.File` reference +- *lines* Will split files line-by-line and emit a new message for each line +- *contents* The default. Provides the contents of a file as a byte array + +When using `--mode=lines`, you can also provide the additional option `--withMarkers=true`. +If set to `true`, the underlying `FileSplitter` will emit additional _start-of-file_ and _end-of-file_ marker messages before and after the actual data. +The payload of these 2 additional marker messages is of type `FileSplitter.FileMarker`. The option `withMarkers` defaults to `false` if not explicitly set. + +See link:../../../functions/supplier/sftp-supplier/README.adoc[`sftp-supplier`] for advanced configuration options. + +== Input + +N/A (Fetches files from an SFTP server). + +== Output + +=== mode = contents + +==== Headers: + +* `Content-Type: application/octet-stream` +* `file_name: ` +* `file_remoteFileInfo ` +* `file_remoteHostPort: ` +* `file_remoteDirectory: ` +* `file_remoteFile: ` +* `sftp_selectedServer: ` (if multi-source) + +==== Payload: + +A `byte[]` filled with the file contents. + +=== mode = lines + +==== Headers: + +* `Content-Type: text/plain` +* `file_name: ` +* `correlationId: ` (same for each line) +* `sequenceNumber: ` +* `sequenceSize: 0` (number of lines is not know until the file is read) +* `file_marker : ` (if with-markers is enabled) + +==== Payload: + +A `String` for each line. + +The first line is optionally preceded by a message with a `START` marker payload. +The last line is optionally followed by a message with an `END` marker payload. + +Marker presence and format are determined by the `with-markers` and `markers-json` properties. + +=== mode = ref + +==== Headers: + +* `file_remoteHostPort: ` +* `file_remoteDirectory: ` +* `file_remoteFile: ` +* `file_originalFile: ` +* `file_name ` +* `file_relativePath` +* `file_remoteFile: ` +* `sftp_selectedServer: ` (if multi-source) + +==== Payload: + +A `java.io.File` object. + +== Options + +The **$$ftp$$** $$source$$ has the following options: + +//tag::configuration-properties[] +$$file.consumer.markers-json$$:: $$When 'fileMarkers == true', specify if they should be produced as FileSplitter.FileMarker objects or JSON.$$ *($$Boolean$$, default: `$$true$$`)* +$$file.consumer.mode$$:: $$The FileReadingMode to use for file reading sources. Values are 'ref' - The File object, 'lines' - a message per line, or 'contents' - the contents as bytes.$$ *($$FileReadingMode$$, default: `$$$$`, possible values: `ref`,`lines`,`contents`)* +$$file.consumer.with-markers$$:: $$Set to true to emit start of file/end of file marker messages before/after the data. Only valid with FileReadingMode 'lines'.$$ *($$Boolean$$, default: `$$$$`)* +$$sftp.supplier.auto-create-local-dir$$:: $$Set to true to create the local directory if it does not exist.$$ *($$Boolean$$, default: `$$true$$`)* +$$sftp.supplier.delay-when-empty$$:: $$Duration of delay when no new files are detected.$$ *($$Duration$$, default: `$$1s$$`)* +$$sftp.supplier.delete-remote-files$$:: $$Set to true to delete remote files after successful transfer.$$ *($$Boolean$$, default: `$$false$$`)* +$$sftp.supplier.directories$$:: $$A list of factory "name.directory" pairs.$$ *($$String[]$$, default: `$$$$`)* +$$sftp.supplier.factories$$:: $$A map of factory names to factories.$$ *($$Map$$, default: `$$$$`)* +$$sftp.supplier.factory.allow-unknown-keys$$:: $$True to allow an unknown or changed key.$$ *($$Boolean$$, default: `$$false$$`)* +$$sftp.supplier.factory.host$$:: $$The host name of the server.$$ *($$String$$, default: `$$localhost$$`)* +$$sftp.supplier.factory.known-hosts-expression$$:: $$A SpEL expression resolving to the location of the known hosts file.$$ *($$Expression$$, default: `$$$$`)* +$$sftp.supplier.factory.pass-phrase$$:: $$Passphrase for user's private key.$$ *($$String$$, default: `$$$$`)* +$$sftp.supplier.factory.password$$:: $$The password to use to connect to the server.$$ *($$String$$, default: `$$$$`)* +$$sftp.supplier.factory.port$$:: $$The port of the server.$$ *($$Integer$$, default: `$$22$$`)* +$$sftp.supplier.factory.private-key$$:: $$Resource location of user's private key.$$ *($$Resource$$, default: `$$$$`)* +$$sftp.supplier.factory.username$$:: $$The username to use to connect to the server.$$ *($$String$$, default: `$$$$`)* +$$sftp.supplier.fair$$:: $$True for fair rotation of multiple servers/directories. This is false by default so if a source has more than one entry, these will be received before the other sources are visited.$$ *($$Boolean$$, default: `$$false$$`)* +$$sftp.supplier.filename-pattern$$:: $$A filter pattern to match the names of files to transfer.$$ *($$String$$, default: `$$$$`)* +$$sftp.supplier.filename-regex$$:: $$A filter regex pattern to match the names of files to transfer.$$ *($$Pattern$$, default: `$$$$`)* +$$sftp.supplier.list-only$$:: $$Set to true to return file metadata without the entire payload.$$ *($$Boolean$$, default: `$$false$$`)* +$$sftp.supplier.local-dir$$:: $$The local directory to use for file transfers.$$ *($$File$$, default: `$$$$`)* +$$sftp.supplier.max-fetch$$:: $$The maximum number of remote files to fetch per poll; default unlimited. Does not apply when listing files or building task launch requests.$$ *($$Integer$$, default: `$$$$`)* +$$sftp.supplier.preserve-timestamp$$:: $$Set to true to preserve the original timestamp.$$ *($$Boolean$$, default: `$$true$$`)* +$$sftp.supplier.remote-dir$$:: $$The remote FTP directory.$$ *($$String$$, default: `$$/$$`)* +$$sftp.supplier.remote-file-separator$$:: $$The remote file separator.$$ *($$String$$, default: `$$/$$`)* +$$sftp.supplier.stream$$:: $$Set to true to stream the file rather than copy to a local directory.$$ *($$Boolean$$, default: `$$false$$`)* +$$sftp.supplier.tmp-file-suffix$$:: $$The suffix to use while the transfer is in progress.$$ *($$String$$, default: `$$.tmp$$`)* +//end::configuration-properties[] + +== Examples + +``` +java -jar sftp_source.jar --sftp.supplier.remote-dir=foo --file.mode=lines --sftp.supplier.factory.host=sftpserver \ + --sftp.supplier.factory.username=user --ftp.supplier.factory.password=pw --sftp.supplier.local-dir=/foo +``` +//end::ref-doc[] diff --git a/applications/source/sftp-source/pom.xml b/applications/source/sftp-source/pom.xml new file mode 100644 index 00000000..c868b36e --- /dev/null +++ b/applications/source/sftp-source/pom.xml @@ -0,0 +1,95 @@ + + + 4.0.0 + sftp-source + 3.0.0-SNAPSHOT + sftp-source + sftp source apps + jar + + + org.springframework.cloud.stream.app + stream-applications-core + 3.0.0-SNAPSHOT + + + + + + org.springframework.boot + spring-boot-starter-test + test + + + org.springframework.cloud.fn + sftp-supplier + ${java-functions.version} + + + org.springframework.cloud.fn + function-test-support + ${java-functions.version} + test + + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${project.version} + test + + + + + + + org.springframework.cloud + spring-cloud-app-starter-doc-maven-plugin + + + org.springframework.cloud.stream.app.plugin + spring-cloud-stream-app-maven-plugin + + + sftp + source + ${project.version} + org.springframework.cloud.fn.supplier.sftp.SftpSupplierConfiguration.class + + + + org.springframework.cloud.fn + sftp-supplier + ${java-functions.version} + + + org.springframework.cloud.stream.app + stream-applications-composite-function-support + ${project.version} + + + + + + + + + + + true + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + + + false + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + + + diff --git a/applications/source/sftp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/source/sftp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..2517cd6b --- /dev/null +++ b/applications/source/sftp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1,3 @@ +configuration-properties.classes=org.springframework.cloud.fn.supplier.sftp.SftpSupplierProperties, \ + org.springframework.cloud.fn.supplier.sftp.SftpSupplierProperties$Factory,\ + org.springframework.cloud.fn.common.file.FileConsumerProperties diff --git a/applications/source/sftp-source/src/test/java/org/springframework/cloud/stream/app/source/sftp/SftpSourceTests.java b/applications/source/sftp-source/src/test/java/org/springframework/cloud/stream/app/source/sftp/SftpSourceTests.java new file mode 100644 index 00000000..9c17a8ca --- /dev/null +++ b/applications/source/sftp-source/src/test/java/org/springframework/cloud/stream/app/source/sftp/SftpSourceTests.java @@ -0,0 +1,103 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.cloud.stream.app.source.sftp; + +import java.io.File; +import java.nio.file.Paths; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.junit.jupiter.api.Test; + +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.cloud.fn.supplier.sftp.SftpSupplierConfiguration; +import org.springframework.cloud.fn.supplier.sftp.SftpSupplierProperties; +import org.springframework.cloud.fn.test.support.sftp.SftpTestSupport; +import org.springframework.cloud.stream.binder.test.OutputDestination; +import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; +import org.springframework.context.annotation.Import; +import org.springframework.messaging.Message; + +import static org.assertj.core.api.Assertions.assertThat; + +public class SftpSourceTests extends SftpTestSupport { + @Test + void simple() { + TestChannelBinderConfiguration.applicationContextRunner(TestApp.class) + .withPropertyValues("sftp.supplier.factory.username = foo", + "sftp.supplier.factory.password = foo", + "file.consumer.mode = ref", + "sftp.supplier.factory.cacheSessions = true", + "sftp.supplier.factory.port =${sftp.factory.port}", + "sftp.supplier.factory.allowUnknownKeys=true", + "sftp.supplier.remoteDir=sftpSource", + "spring.cloud.stream.function.definition=sftpSupplier") + .run(context -> { + OutputDestination output = context.getBean(OutputDestination.class); + SftpSupplierProperties config = context.getBean(SftpSupplierProperties.class); + Message message = output.receive(10000); + assertThat(new File(new String(message.getPayload()).replaceAll("\"", ""))).isEqualTo( + new File(config.getLocalDir(), "sftpSource1.txt")); + message = output.receive(10000); + assertThat(new File(new String(message.getPayload()).replaceAll("\"", ""))).isEqualTo( + new File(config.getLocalDir(), "sftpSource2.txt")); + }); + } + + @Test + void taskLaunchRequest() { + TestChannelBinderConfiguration.applicationContextRunner(TestApp.class) + .withPropertyValues("sftp.supplier.factory.username = foo", + "sftp.supplier.factory.password = foo", + "file.consumer.mode = ref", + "sftp.supplier.factory.cacheSessions = true", + "sftp.supplier.factory.port =${sftp.factory.port}", + "sftp.supplier.factory.allowUnknownKeys=true", + "sftp.supplier.remoteDir=sftpSource", + "sftp.supplier.localDir=" + this.targetLocalDirectory.toString(), + "--task.launch.request.arg-expressions=fileName=payload", + "--task.launch.request.task-name=myTask", + "spring.cloud.stream.function.definition=sftpSupplier|taskLaunchRequestFunction") + .run(context -> { + OutputDestination output = context.getBean(OutputDestination.class); + SftpSupplierProperties config = context.getBean(SftpSupplierProperties.class); + ObjectMapper objectMapper = context.getBean(ObjectMapper.class); + Message message = output.receive(10000); + Map taskLaunchRequest = objectMapper.readValue(message.getPayload(), HashMap.class); + assertThat(taskLaunchRequest.get("name")).isEqualTo("myTask"); + assertThat((List) taskLaunchRequest.get("args")) + .contains("fileName=" + Paths + .get(config.getLocalDir().toString(), "sftpSource1.txt") + .toString()); + message = output.receive(10000); + taskLaunchRequest = objectMapper.readValue(message.getPayload(), HashMap.class); + assertThat(taskLaunchRequest.get("name")).isEqualTo("myTask"); + assertThat((List) taskLaunchRequest.get("args")) + .containsExactly("fileName=" + Paths + .get(config.getLocalDir().toString(), "sftpSource2.txt") + .toString()); + }); + } + + @SpringBootApplication + @Import(SftpSupplierConfiguration.class) + public static class TestApp { + + } +} diff --git a/functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/remote/RemoteFileDeletingAdvice.java b/functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/remote/RemoteFileDeletingAdvice.java new file mode 100644 index 00000000..ec25de02 --- /dev/null +++ b/functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/remote/RemoteFileDeletingAdvice.java @@ -0,0 +1,55 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.cloud.fn.common.file.remote; + +import org.springframework.integration.aop.MessageSourceMutator; +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.file.remote.RemoteFileTemplate; +import org.springframework.messaging.Message; + +/** + * A {@link MessageSourceMutator} that deletes a remote file on success. + * + * @author David Turanski + * + */ +public class RemoteFileDeletingAdvice implements MessageSourceMutator { + + private final RemoteFileTemplate template; + + private final String remoteFileSeparator; + + /** + * Construct an instance with the provided template and separator. + * @param template the template. + * @param remoteFileSeparator the separator. + */ + public RemoteFileDeletingAdvice(RemoteFileTemplate template, + String remoteFileSeparator) { + this.template = template; + this.remoteFileSeparator = remoteFileSeparator; + } + + @Override + public Message afterReceive(Message result, MessageSource source) { + String remoteDir = (String) result.getHeaders().get(FileHeaders.REMOTE_DIRECTORY); + String remoteFile = (String) result.getHeaders().get(FileHeaders.REMOTE_FILE); + this.template.remove(remoteDir + this.remoteFileSeparator + remoteFile); + return result; + } +} diff --git a/functions/common/ftp-common/pom.xml b/functions/common/ftp-common/pom.xml index 9bd99f3f..3ae5b914 100644 --- a/functions/common/ftp-common/pom.xml +++ b/functions/common/ftp-common/pom.xml @@ -4,7 +4,7 @@ ftp-common 1.0.0-SNAPSHOT ftp-common - file consumer + ftp common org.springframework.cloud.fn diff --git a/functions/common/function-test-support/pom.xml b/functions/common/function-test-support/pom.xml index d259b9a2..7c3e7db2 100644 --- a/functions/common/function-test-support/pom.xml +++ b/functions/common/function-test-support/pom.xml @@ -18,6 +18,7 @@ 1.14.2 1.1.1 1.2.7.RELEASE + 4.0.3 @@ -61,6 +62,11 @@ org.springframework spring-websocket + + org.awaitility + awaitility + ${awaitility.version} + diff --git a/functions/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/sftp/SftpTestSupport.java b/functions/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/sftp/SftpTestSupport.java index ad82a7f7..9e83d95d 100644 --- a/functions/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/sftp/SftpTestSupport.java +++ b/functions/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/sftp/SftpTestSupport.java @@ -60,13 +60,13 @@ public class SftpTestSupport extends RemoteFileTestSupport { server = SshServer.setUpDefaultServer(); server.setPasswordAuthenticator((username, password, session) -> StringUtils.hasText(password) && !"badPassword".equals(password)); // fail if pub key validation failed - server.setPublickeyAuthenticator((username, key, session) -> key.equals(decodePublicKey("id_rsa_pp.pub.rename2"))); + server.setPublickeyAuthenticator((username, key, session) -> key.equals(decodePublicKey("id_rsa_pp.pub"))); server.setPort(0); server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider(new File("hostkey.ser"))); server.setSubsystemFactories(Collections.singletonList(new SftpSubsystemFactory())); server.setFileSystemFactory(new VirtualFileSystemFactory(remoteTemporaryFolder)); server.start(); - System.setProperty("sftp.consumer.factory.port", String.valueOf(server.getPort())); + System.setProperty("sftp.factory.port", String.valueOf(server.getPort())); System.setProperty("sftp.consumer.localDir", localTemporaryFolder + File.separator + "localTarget"); } @@ -78,7 +78,7 @@ public class SftpTestSupport extends RemoteFileTestSupport { if (hostkey.exists()) { hostkey.delete(); } - System.clearProperty("sftp.consumer.factory.port"); + System.clearProperty("sftp.factory.port"); System.clearProperty("sftp.consumer.localDir"); } diff --git a/functions/common/function-test-support/src/main/resources/id_rsa_pp b/functions/common/function-test-support/src/main/resources/id_rsa_pp new file mode 100644 index 00000000..7c7cba16 --- /dev/null +++ b/functions/common/function-test-support/src/main/resources/id_rsa_pp @@ -0,0 +1,30 @@ +-----BEGIN RSA PRIVATE KEY----- +Proc-Type: 4,ENCRYPTED +DEK-Info: AES-128-CBC,26D1755B05980BA01B1E8D3B65EF98D3 + +J5fMQDf0HrwcnfYujq+q05GEOSEMVWMU0vr0hBtz2WvUeaFBBVAvUWbJo7PHTdDV +vEdSv+k8FazOkIZpeOW26pKNaLSuFfN+lgtAW4p4yqGQhbL8byh+Ka5uGaPH1xQj +o4exMFqbwFaHG10LJoPp5NnK+T64w8McJEihBPxv/qwtsz0YhhDVl/1eSwKsGa5y +Usfdv0QbjNDtpV7+sy7OunpaaKjb8iQ/PbFDsX0TSiy6jJflPwCYVUoh3jCEJze3 +OUKwoQu7AiHmUrsnLtDuL49Q5hV+f9+IPJlzSqU5Fu8PlfCowH+e8WWLVQkh7Hht +iwOQs5UIWH+Nzbguu3Gbph5lqMtgkQwK6/PSFLQXuJmf9l6eo7E+Wh5mFWwoIENT +mKUlgVU0ymajPHcA7uOVXl5XX+Mt0DiMdGiB4N6RJ1OjWgqlRSJbVmGPrnZWmjdW +VltNux6JrO1bOgwtApCouDZhnP7/JDhM2PCa/F2+XCc6s8Hnmmt+gBw6IM4fb+YH +2gf1jx5Onp1yuBGe6tvbrfPAdXYU1R3kd2+Yf5mdsu2+xiC4wBKSfjleoHN4xJ/b +wopjQvGV5dj/VjdUt99lMHGWPr+p4NmdEBoriCgjYuRjRDvCiTV2qKZe2MU+0CVE +/jCA8iMRQnx3LMjnmJXP/96j2fyXHMowClTF43Cvdc5jh622WITOOFAIq9RfIXdS +47G6FAN54V+Qt0pXEgIOmvG+B2C24A041fo3jUPZxFRSYYuv9vG6QJby3kCsaqqt +ngYs2JSKx9CfOfGfyPbNt2/CO/bBsXgYzLR7REx5My1Mp9YsDmeIgbcd2V5hw/Me +rlSXrG4Eqs9gRDrBUvsydUOJFC1PlnIXH3VSBc6X9o4n2H6XECOsJQSeXLCaeMav +deKb0r1HvdbAYrdqw6mRM8Ok3fpSoD3mUsZQ8fp3luO3tHL1lddxHb7EsKzt/ubh +B9lEzDTcILlINlCl41X0OZKr/c+Ec8EdaSvITYJj0fvaZmDF7Wcs/dDWNft2XUaD +VcptOKbQVE1ufbrc2s1BdKOriAC6dSVKVDrAUQD/MlhT3p/YwbjSY2MqwRKztWRv +sA1Kqjg1IdUQzQizRKuHa322qnduLjHy0rb0ElrMpFe3B+OcIPs1E7Gvo4BVQ5jU +5GqHm83iaFIQaXmEsrVtCOBygVf00+WNRV3WFTOP9UEWFgtxsaHAU8UZhsDbKcZ5 +l/w6kQdNElQuJA+1n1OxRZ5wIpfrRIMaxBQg2plUTVb9Tgz1qTZxHIEYPAAePfvX +tDn9SBiktj/8dhy1T0ko89yXCekpGxkU9rbmAd4Lrp6Wbc6+bKt4KzAqw319qq9U +Pslq7EKNM0Zq5pfwn1MRjzvmyHxz2sYHeij0CmuZY7xuOa4NzXt1vr6nSlIpX77W +ng/Rnd7qpyG2IWYihi6ztFHyj9h3FEcBHMk4JINLcdYOxXICz3KsRMfntrwQ4E6O +NFJ3fPpVNkk4GZcxy6idNkUBz1s8ixVWC8yi36byxE+TTRtcvqXQnJVgs63vMUlC +HVwaGau4YeUp4Nj2ZO44Srd/kQRy8yuCOPDdlJEiO6eDD4+XKedJQg1LuGeoXMVn +-----END RSA PRIVATE KEY----- diff --git a/functions/common/function-test-support/src/main/resources/id_rsa_pp.pub b/functions/common/function-test-support/src/main/resources/id_rsa_pp.pub new file mode 100644 index 00000000..50cf189a --- /dev/null +++ b/functions/common/function-test-support/src/main/resources/id_rsa_pp.pub @@ -0,0 +1 @@ +AAAAB3NzaC1yc2EAAAADAQABAAABAQC6MIzgyVi8G1+HRhFHPWRH+3w/8/uxtiuIfb4puVPjHI53Lvf5odzfhv0T6Z2/jSXmI3I6dpjbsgiptdCTX4kqUFLXxkuJR4LHatNtgO1w32aVIdAvfj7KtrL3SmP2XWqQGVcUWHEn2H1RHFHKdC6ArYFb1X8p5N/BHSQjuttaeVi9FsDxvC5euIbtDEEJmmvjjfWlI1m/6qCqMYxDWA9i9APU/rB0QwFNUQ6HuZ2QzEaU/hQMGmqgW5o1I/W8JR0bqis8wZQDLv1fwCkXpWG5BAuiJH+FJMxRAkfEMBpVwO7Sl0ufePVuSM2BMAAe+4a75sVp8ahbOId6y0GUTeJl diff --git a/functions/pom.xml b/functions/pom.xml index 5b6156f5..2505bfa8 100644 --- a/functions/pom.xml +++ b/functions/pom.xml @@ -91,6 +91,7 @@ supplier/jms-supplier supplier/mongodb-supplier supplier/mqtt-supplier + supplier/sftp-supplier supplier/tcp-supplier supplier/time-supplier supplier/rabbit-supplier diff --git a/functions/supplier/sftp-supplier/README.adoc b/functions/supplier/sftp-supplier/README.adoc new file mode 100644 index 00000000..59db154b --- /dev/null +++ b/functions/supplier/sftp-supplier/README.adoc @@ -0,0 +1,86 @@ +# SFTP Supplier + +This module provides a SFTP supplier that can be reused and composed in other applications. +The `Supplier` uses various `Sftp` inbound adapters from Spring Integration to support a range of modes to consume data from an SFTP server. +These include: + +* Synchronize remote files to a local directory and supplying the File contents, the File reference, or a message per line. +* Stream remote file contents as a byte array directly without copying to a local directory. +* List the remote file names only. + +Messages emitted by the supplier are provided as a byte array by default. However, this can be customized using the `--mode` option: + +- *ref* Provides a `java.io.File` reference +- *lines* Will split files line-by-line and emit a new message for each line +- *contents* The default. Provides the contents of a file as a byte array + + +NOTE: When using `--mode=lines`, you can provide an additional option `--withMarkers=true`. +If set to `true`, the underlying `FileSplitter` will emit additional _start-of-file_ and _end-of-file_ marker messages before and after the actual data. +The payload of these 2 additional marker messages is of type `FileSplitter.FileMarker`. +The option `withMarkers` defaults to `false` if not explicitly set. + +When configuring the `sftp.factory.known-hosts-expression` option, the root object of the evaluation is the application context, an example might be `sftp.factory.known-hosts-expression = @systemProperties['user.home'] + '/.ssh/known_hosts'`. + +## Idempotency + +By default the supplier uses a https://docs.spring.io/spring-integration/api/org/springframework/integration/metadata/SimpleMetadataStore.html[SimpleMetadataStore], storing the last modified time to track files that have already been processed in memory. +If an application using this supplier is restarted, any existing files will be reprocessed. You can inject on of the persistent https://docs.spring.io/spring-integration/reference/html/meta-data-store.html[MetadataStore implementations] provided by Spring Integration, or your own of course, to maintain this state permanently. +See also link:../../common/metadata-store-common/README.adoc[`MetadataStore`] options for possible shared persistent store configuration for the `SftpPersistentAcceptOnceFileListFilter` used in the SFTP Source. + + +## Multiple SFTP Servers +This source supports consuming from multiple SFTP servers. +This requires configuring an SFTP Session Factory for each server. +The labels `one` and `two` shown below can be replaced by any names you want. +The following configuration will rotate between two SFTP servers (this can also be used for multiple directories on the same server), consuming files in a round-robin fashion: + +``` +sftp.supplier.factories.one.host=host1 +sftp.supplier.factories.one.port=1234, +sftp.supplier.factories.one.username = user1, +sftp.supplier.factories.one.password = pass1, +... +sftp.supplier.factories.two.host=host2, +sftp.supplier.factories.two.port=2345, +sftp.supplier.factories.two.username = user2, +sftp.supplier.factories.two.password = pass2, +sftp.supplier.directories=one.sftpSource,two.sftpSecondSource, +sftp.supplier.max-fetch=1, +sftp.supplier.fair=true +``` +--- + + + +`SFtpSupplier` is implemented as a `java.util.function.Supplier`. +This supplier gives you a reactive stream of objects from the provided directory(ies) as the supplier has a signature of `Supplier>>`. +Users have to subscribe to this `Flux` and receive the data. + +## Beans for injection + +You can import the `SftpSupplierConfiguration` in the application and then inject the following bean. + +`sftpSupplier` + +You need to inject this as `Supplier>>`. + +You can use `sftpSupplier` as a qualifier when injecting. + +Once injected, you can use the `get` method of the `Supplier` to invoke it and then subscribe to the returned `Flux`. + +## Configuration Options + +All configuration properties are prefixed with `ftp.supplier`. +There are also properties that need to be used with the prefix `file.consumer`. + +For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierProperties.java[SftpSupplierProperties]. +Also see link:src/main/java/org/springframework/cloud/fn/supplier/file/FileConsumerProperties.java[FileConsumerProperties]. + +## Examples + +See this link:src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java[test suite] for the various ways, this supplier is used. + +## Other usage + +See this link:../../../applications/source/sftp-source/README.adoc[README] where this supplier is used to create a Spring Cloud Stream application that provides an Sftp Source. \ No newline at end of file diff --git a/functions/supplier/sftp-supplier/pom.xml b/functions/supplier/sftp-supplier/pom.xml new file mode 100644 index 00000000..16a2ebef --- /dev/null +++ b/functions/supplier/sftp-supplier/pom.xml @@ -0,0 +1,76 @@ + + + 4.0.0 + sftp-supplier + 1.0.0-SNAPSHOT + sftp-supplier + sftp supplier + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.integration + spring-integration-sftp + + + org.springframework.boot + spring-boot-starter-integration + + + org.springframework.cloud.fn + file-common + ${project.version} + + + org.springframework.cloud.fn + ftp-common + ${project.version} + + + org.springframework.cloud.fn + metadata-store-common + ${project.version} + + + org.springframework.boot + spring-boot-starter-json + true + + + org.springframework.boot + spring-boot-starter-validation + + + org.springframework.boot + spring-boot-starter-logging + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.springframework.boot + spring-boot-starter-test + test + + + io.projectreactor + reactor-test + test + + + org.springframework.cloud.fn + function-test-support + ${project.version} + test + + + + diff --git a/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java b/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java new file mode 100644 index 00000000..b2908617 --- /dev/null +++ b/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java @@ -0,0 +1,364 @@ +/* + * Copyright 2018-2020 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 + * + * https://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.cloud.fn.supplier.sftp; + +import java.io.IOException; +import java.util.Arrays; +import java.util.List; +import java.util.function.Supplier; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +import org.reactivestreams.Publisher; + +import org.springframework.aop.framework.ProxyFactoryBean; +import org.springframework.aop.support.NameMatchMethodPointcutAdvisor; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.file.FileConsumerProperties; +import org.springframework.cloud.fn.common.file.FileUtils; +import org.springframework.cloud.fn.common.file.remote.RemoteFileDeletingAdvice; +import org.springframework.context.Lifecycle; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.context.annotation.Primary; +import org.springframework.integration.aop.ReceiveMessageAdvice; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.endpoint.MessageProducerSupport; +import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.file.filters.ChainFileListFilter; +import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.handler.MessageProcessor; +import org.springframework.integration.metadata.ConcurrentMetadataStore; +import org.springframework.integration.sftp.dsl.Sftp; +import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter; +import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter; +import org.springframework.integration.sftp.session.SftpRemoteFileTemplate; +import org.springframework.integration.util.IntegrationReactiveUtils; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.MessagingException; +import org.springframework.messaging.PollableChannel; +import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.Assert; +import org.springframework.util.CollectionUtils; +import org.springframework.util.MimeTypeUtils; +import org.springframework.util.StringUtils; + +import com.jcraft.jsch.ChannelSftp.LsEntry; +import reactor.core.publisher.Flux; + +/** + * @author Gary Russell + * @author Artem Bilan + * @author Chris Schaefer + * @author Christian Tzolov + * @author David Turanski + */ + +@Configuration +@EnableConfigurationProperties({SftpSupplierProperties.class, FileConsumerProperties.class}) +@Import({SftpSupplierFactoryConfiguration.class}) +@SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection") +public class SftpSupplierConfiguration { + + private static final String METADATA_STORE_PREFIX = "sftpSource/"; + + @Bean + public Supplier>> sftpSupplier(MessageSource sftpMessageSource, + @Nullable Publisher> sftpReadingFlow, + SftpSupplierProperties sftpSupplierProperties) { + + Flux> flux = sftpReadingFlow == null + ? sftpMessageFlux(sftpMessageSource, sftpSupplierProperties) + : Flux.from(sftpReadingFlow); + + return () -> flux.doOnSubscribe(s -> { + if (sftpMessageSource instanceof Lifecycle) { + ((Lifecycle) sftpMessageSource).start(); + } + }); + } + + @Bean + @Primary + public MessageSource sftpMessageSource( + MessageSource messageSource, + BeanFactory beanFactory, + @Nullable List receiveMessageAdvice) { + + if (CollectionUtils.isEmpty(receiveMessageAdvice)) { + return messageSource; + } + + ProxyFactoryBean proxyFactoryBean = new ProxyFactoryBean(); + proxyFactoryBean.setTarget(messageSource); + proxyFactoryBean.setBeanFactory(beanFactory); + receiveMessageAdvice.stream().map(advice -> { + NameMatchMethodPointcutAdvisor advisor = new NameMatchMethodPointcutAdvisor(advice); + advisor.addMethodName("receive"); + return advisor; + }).forEach(proxyFactoryBean::addAdvisor); + + return (MessageSource) proxyFactoryBean.getObject(); + } + + /* + * Configure the standard filters for SFTP inbound adapters. + */ + @Bean + public ChainFileListFilter chainFilter(SftpSupplierProperties sftpSupplierProperties, + ConcurrentMetadataStore metadataStore) { + ChainFileListFilter chainFilter = new ChainFileListFilter<>(); + + if (StringUtils.hasText(sftpSupplierProperties.getFilenamePattern())) { + chainFilter + .addFilter(new SftpRegexPatternFileListFilter(sftpSupplierProperties.getFilenamePattern())); + } + else if (sftpSupplierProperties.getFilenameRegex() != null) { + chainFilter + .addFilter(new SftpRegexPatternFileListFilter(sftpSupplierProperties.getFilenameRegex())); + } + // TODO: Temporary work-around for + // https://github.com/spring-projects/spring-integration/issues/3315. + chainFilter.addFilter(Arrays::asList); + chainFilter.addFilter(new SftpPersistentAcceptOnceFileListFilter(metadataStore, METADATA_STORE_PREFIX)); + return chainFilter; + } + + /* + * Create a Flux from a MessageSource that will be used by the supplier. + */ + private Flux> sftpMessageFlux(MessageSource sftpMessageSource, + SftpSupplierProperties sftpSupplierProperties) { + + return IntegrationReactiveUtils.messageSourceToFlux(sftpMessageSource) + .subscriberContext(context -> context.put(IntegrationReactiveUtils.DELAY_WHEN_EMPTY_KEY, + sftpSupplierProperties.getDelayWhenEmpty())); + + } + + private static String remoteDirectory(SftpSupplierProperties sftpSupplierProperties) { + return sftpSupplierProperties.isMultiSource() + ? SftpSupplierProperties.keyDirectories(sftpSupplierProperties).get(0).getDirectory() + : sftpSupplierProperties.getRemoteDir(); + } + + @Configuration + @ConditionalOnProperty(prefix = "sftp.supplier", name = "stream") + static class StreamingConfiguration { + + @Bean + public SftpRemoteFileTemplate sftpTemplate(SftpSupplierFactoryConfiguration.DelegatingFactoryWrapper wrapper) { + return new SftpRemoteFileTemplate(wrapper.getFactory()); + } + + /** + * Streaming {@link MessageSource} that provides an InputStream for each remote file. It + * does not synchronize files to a local directory. + * @return a {@link MessageSource}. + */ + @Bean + public MessageSource targetMessageSource(SftpRemoteFileTemplate sftpTemplate, + SftpSupplierProperties sftpSupplierProperties, + FileListFilter fileListFilter) { + + return Sftp.inboundStreamingAdapter(sftpTemplate) + .remoteDirectory(remoteDirectory(sftpSupplierProperties)) + .remoteFileSeparator(sftpSupplierProperties.getRemoteFileSeparator()) + .filter(fileListFilter) + .maxFetchSize(sftpSupplierProperties.getMaxFetch()).get(); + } + + @Bean + public Publisher> sftpReadingFlow( + MessageSource sftpMessageSource, + SftpSupplierProperties sftpSupplierProperties, + FileConsumerProperties fileConsumerProperties) { + + return FileUtils.enhanceStreamFlowForReadingMode(IntegrationFlows + .from(IntegrationReactiveUtils.messageSourceToFlux(sftpMessageSource) + .subscriberContext( + context -> (context.put(IntegrationReactiveUtils.DELAY_WHEN_EMPTY_KEY, + sftpSupplierProperties.getDelayWhenEmpty())))), + fileConsumerProperties) + .toReactivePublisher(); + } + + @Bean + @ConditionalOnProperty(prefix = "sftp.supplier", value = "delete-remote-files") + public RemoteFileDeletingAdvice remoteFileDeletingAdvice(SftpRemoteFileTemplate sftpTemplate, + SftpSupplierProperties sftpSupplierProperties) { + return new RemoteFileDeletingAdvice(sftpTemplate, sftpSupplierProperties.getRemoteFileSeparator()); + } + + } + + @Configuration + @ConditionalOnExpression("environment['sftp.supplier.stream']!='true'") + static class NonStreamingConfiguration { + + /** + * Enrich the flow to provide some standard headers, depending on + * {@link FileConsumerProperties}, when consuming file contents. + * @param sftpMessageSource the {@link MessageSource}. + * @param fileConsumerProperties the {@code FileConsumerProperties}. + * @return a {@code Publisher}. + */ + @Bean + @ConditionalOnExpression("environment['file.consumer.mode']!='ref' && environment['sftp.supplier.list-only']!='true'") + public Publisher> sftpReadingFlow( + MessageSource sftpMessageSource, + FileConsumerProperties fileConsumerProperties) { + + return FileUtils.enhanceFlowForReadingMode(IntegrationFlows + .from(IntegrationReactiveUtils.messageSourceToFlux(sftpMessageSource)), + fileConsumerProperties) + .toReactivePublisher(); + } + + /** + * A {@link MessageSource} that synchronizes files to a local directory. + * @return the {code MessageSource}. + */ + @ConditionalOnExpression("environment['sftp.supplier.list-only'] != 'true'") + @Bean + public MessageSource targetMessageSource(SftpSupplierProperties sftpSupplierProperties, + SftpSupplierFactoryConfiguration.DelegatingFactoryWrapper delegatingFactoryWrapper, + FileListFilter fileListFilter) { + + return Sftp + .inboundAdapter(delegatingFactoryWrapper.getFactory()) + .preserveTimestamp(sftpSupplierProperties.isPreserveTimestamp()) + .autoCreateLocalDirectory(sftpSupplierProperties.isAutoCreateLocalDir()) + .deleteRemoteFiles(sftpSupplierProperties.isDeleteRemoteFiles()) + .localDirectory(sftpSupplierProperties.getLocalDir()) + .remoteDirectory(remoteDirectory(sftpSupplierProperties)) + .remoteFileSeparator(sftpSupplierProperties.getRemoteFileSeparator()) + .temporaryFileSuffix(sftpSupplierProperties.getTmpFileSuffix()) + .metadataStorePrefix(METADATA_STORE_PREFIX) + .maxFetchSize(sftpSupplierProperties.getMaxFetch()) + .filter(fileListFilter).get(); + } + + } + + /* + * List only configuration + */ + @Configuration + @ConditionalOnProperty(prefix = "sftp.supplier", name = "list-only") + static class ListingOnlyConfiguration { + + @Bean + PollableChannel listingChannel() { + return new QueueChannel(); + } + + @Bean + @SuppressWarnings("unchecked") + public MessageSource targetMessageSource(PollableChannel listingChannel, + SftpListingMessageProducer sftpListingMessageProducer) { + return () -> { + sftpListingMessageProducer.listNames(); + return (Message) listingChannel.receive(); + }; + + } + + @Bean + public SftpListingMessageProducer sftpListingMessageProducer(SftpSupplierProperties sftpSupplierProperties, + SftpSupplierFactoryConfiguration.DelegatingFactoryWrapper delegatingFactoryWrapper) { + + return new SftpListingMessageProducer(delegatingFactoryWrapper.getFactory(), + remoteDirectory(sftpSupplierProperties), + sftpSupplierProperties.getRemoteFileSeparator()); + } + + @Bean + public IntegrationFlow listingFlow(MessageProducerSupport messageProducerSupport, + MessageChannel listingChannel, MessageProcessor metadataWriter) { + + return IntegrationFlows.from(messageProducerSupport) + .split() + .transform(metadataWriter) + .channel(listingChannel) + .get(); + } + + @Bean + public MessageProcessor metadataWriter(ConcurrentMetadataStore metadataStore) { + return message -> { + MessageHeaders messageHeaders = message.getHeaders(); + Assert.notNull(messageHeaders, "Cannot transform message with null headers"); + Assert.isTrue(messageHeaders.containsKey(FileHeaders.REMOTE_DIRECTORY), + "Remote directory header not found"); + Assert.hasText((String) message.getPayload(), "Payload must not be empty."); + + metadataStore.putIfAbsent( + message.getHeaders().get(FileHeaders.REMOTE_DIRECTORY).toString() + message.getPayload(), + String.valueOf(message.getHeaders().getTimestamp())); + return message; + }; + } + + static class SftpListingMessageProducer extends MessageProducerSupport { + + private final String remoteDirectory; + + private final SessionFactory sessionFactory; + + private final String remoteFileSeparator; + + SftpListingMessageProducer(SessionFactory sessionFactory, String remoteDirectory, + String remoteFileSeparator) { + + this.sessionFactory = sessionFactory; + this.remoteDirectory = remoteDirectory; + this.remoteFileSeparator = remoteFileSeparator; + } + + public void listNames() { + String[] names = {}; + try { + names = Stream.of(this.sessionFactory.getSession().listNames(this.remoteDirectory)) + .map(name -> String.join(this.remoteFileSeparator, this.remoteDirectory, name)) + .collect(Collectors.toList()).toArray(names); + } + catch (IOException e) { + throw new MessagingException(e.getMessage(), e); + } + sendMessage(MessageBuilder.withPayload(names) + .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN) + .setHeader(FileHeaders.REMOTE_DIRECTORY, this.remoteDirectory + this.remoteFileSeparator) + .build()); + } + + } + + } + +} diff --git a/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierFactoryConfiguration.java b/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierFactoryConfiguration.java new file mode 100644 index 00000000..194a0932 --- /dev/null +++ b/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierFactoryConfiguration.java @@ -0,0 +1,127 @@ +/* + * Copyright 2018-2020 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 + * + * https://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.cloud.fn.supplier.sftp; + +import java.util.HashMap; +import java.util.Map; + +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.DisposableBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.context.annotation.Bean; +import org.springframework.core.io.FileSystemResource; +import org.springframework.integration.context.IntegrationContextUtils; +import org.springframework.integration.file.remote.aop.StandardRotationPolicy; +import org.springframework.integration.file.remote.session.DelegatingSessionFactory; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.sftp.session.DefaultSftpSessionFactory; +import org.springframework.lang.Nullable; + +import com.jcraft.jsch.ChannelSftp.LsEntry; + +/** + * Session factory configuration. + * + * @author Gary Russell + * @author Artem Bilan + * @author David Turanski + * + */ +public class SftpSupplierFactoryConfiguration { + + @Bean + @ConditionalOnMissingBean + public SessionFactory sftpSessionFactory(SftpSupplierProperties properties, BeanFactory beanFactory) { + return buildFactory(beanFactory, properties.getFactory()); + } + + @Bean + public DelegatingFactoryWrapper delegatingFactoryWrapper(SftpSupplierProperties properties, + SessionFactory defaultFactory, BeanFactory beanFactory) { + return new DelegatingFactoryWrapper(properties, defaultFactory, beanFactory); + } + + @Bean + StandardRotationPolicy rotationPolicy(SftpSupplierProperties properties, DelegatingFactoryWrapper factory) { + + return properties.isMultiSource() + ? new StandardRotationPolicy(factory.getFactory(), + SftpSupplierProperties.keyDirectories(properties), properties.isFair()) + : null; + } + + @Bean + public SftpSupplierRotator rotatingAdvice(SftpSupplierProperties properties, + @Nullable StandardRotationPolicy rotationPolicy) { + return properties.isMultiSource() + ? new SftpSupplierRotator(properties, rotationPolicy) + : null; + } + + static SessionFactory buildFactory(BeanFactory beanFactory, SftpSupplierProperties.Factory factory) { + DefaultSftpSessionFactory sftpSessionFactory = new DefaultSftpSessionFactory(true); + sftpSessionFactory.setHost(factory.getHost()); + sftpSessionFactory.setPort(factory.getPort()); + sftpSessionFactory.setUser(factory.getUsername()); + sftpSessionFactory.setPassword(factory.getPassword()); + sftpSessionFactory.setPrivateKey(factory.getPrivateKey()); + sftpSessionFactory.setPrivateKeyPassphrase(factory.getPassPhrase()); + sftpSessionFactory.setAllowUnknownKeys(factory.isAllowUnknownKeys()); + if (factory.getKnownHostsExpression() != null) { + String path = factory.getKnownHostsExpression() + .getValue(IntegrationContextUtils.getEvaluationContext(beanFactory), String.class); + sftpSessionFactory.setKnownHostsResource(new FileSystemResource(path)); + } + + return sftpSessionFactory; + } + + public final static class DelegatingFactoryWrapper implements DisposableBean { + + private final DelegatingSessionFactory delegatingSessionFactory; + + private final Map> factories = new HashMap<>(); + + DelegatingFactoryWrapper(SftpSupplierProperties properties, SessionFactory defaultFactory, + BeanFactory beanFactory) { + properties.getFactories().forEach((key, factory) -> { + this.factories.put(key, SftpSupplierFactoryConfiguration.buildFactory(beanFactory, factory)); + }); + this.delegatingSessionFactory = new DelegatingSessionFactory<>(this.factories, defaultFactory); + } + + public DelegatingSessionFactory getFactory() { + return this.delegatingSessionFactory; + } + + @Override + public void destroy() { + this.factories.values().forEach(f -> { + if (f instanceof DisposableBean) { + try { + ((DisposableBean) f).destroy(); + } + catch (Exception e) { + // empty + } + } + }); + } + + } + +} diff --git a/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierProperties.java b/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierProperties.java new file mode 100644 index 00000000..93204116 --- /dev/null +++ b/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierProperties.java @@ -0,0 +1,406 @@ +/* + * Copyright 2018-2020 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 + * + * https://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.cloud.fn.supplier.sftp; + +import java.io.File; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.regex.Pattern; + +import javax.validation.constraints.AssertTrue; +import javax.validation.constraints.NotBlank; +import javax.validation.constraints.NotNull; + +import org.hibernate.validator.constraints.Range; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.core.io.Resource; +import org.springframework.expression.Expression; +import org.springframework.integration.file.remote.aop.RotationPolicy; +import org.springframework.util.Assert; +import org.springframework.validation.annotation.Validated; + +/** + * @author Gary Russell + * @author Artem Bilan + * @author Chris Schaefer + * @author David Turanski + */ +@ConfigurationProperties("sftp.supplier") +@Validated +public class SftpSupplierProperties { + + /** + * Session factory properties. + */ + private final Factory factory = new Factory(); + + /** + * The remote FTP directory. + */ + private String remoteDir = "/"; + + /** + * The suffix to use while the transfer is in progress. + */ + private String tmpFileSuffix = ".tmp"; + + /** + * The remote file separator. + */ + private String remoteFileSeparator = "/"; + + /** + * Set to true to delete remote files after successful transfer. + */ + private boolean deleteRemoteFiles = false; + + /** + * The local directory to use for file transfers. + */ + private File localDir = new File(System.getProperty("java.io.tmpdir"), "sftp-supplier"); + + /** + * Set to true to create the local directory if it does not exist. + */ + private boolean autoCreateLocalDir = true; + + /** + * A filter pattern to match the names of files to transfer. + */ + private String filenamePattern; + + /** + * A filter regex pattern to match the names of files to transfer. + */ + private Pattern filenameRegex; + + /** + * Set to true to preserve the original timestamp. + */ + private boolean preserveTimestamp = true; + + /** + * Set to true to stream the file rather than copy to a local directory. + */ + private boolean stream = false; + + /** + * Set to true to return file metadata without the entire payload. + */ + private boolean listOnly = false; + + /** + * Duration of delay when no new files are detected. + */ + private Duration delayWhenEmpty = Duration.ofSeconds(1); + + /** + * The maximum number of remote files to fetch per poll; default unlimited. Does not apply + * when listing files or building task launch requests. + */ + private int maxFetch = Integer.MIN_VALUE; + + /** + * True for fair rotation of multiple servers/directories. This is false by default so if + * a source has more than one entry, these will be received before the other sources are + * visited. + */ + private boolean fair; + + /** + * A map of factory names to factories. + */ + private Map factories = Collections.emptyMap(); + + /** + * A list of factory "name.directory" pairs. + */ + private String[] directories; + + @NotBlank + public String getRemoteDir() { + return remoteDir; + } + + public void setRemoteDir(String remoteDir) { + this.remoteDir = remoteDir; + } + + @NotBlank + public String getTmpFileSuffix() { + return tmpFileSuffix; + } + + public void setTmpFileSuffix(String tmpFileSuffix) { + this.tmpFileSuffix = tmpFileSuffix; + } + + @NotBlank + public String getRemoteFileSeparator() { + return remoteFileSeparator; + } + + public void setRemoteFileSeparator(String remoteFileSeparator) { + this.remoteFileSeparator = remoteFileSeparator; + } + + public boolean isAutoCreateLocalDir() { + return autoCreateLocalDir; + } + + public void setAutoCreateLocalDir(boolean autoCreateLocalDir) { + this.autoCreateLocalDir = autoCreateLocalDir; + } + + public boolean isDeleteRemoteFiles() { + return deleteRemoteFiles; + } + + public void setDeleteRemoteFiles(boolean deleteRemoteFiles) { + this.deleteRemoteFiles = deleteRemoteFiles; + } + + @NotNull + public File getLocalDir() { + return localDir; + } + + public final void setLocalDir(File localDir) { + this.localDir = localDir; + } + + public String getFilenamePattern() { + return filenamePattern; + } + + public void setFilenamePattern(String filenamePattern) { + this.filenamePattern = filenamePattern; + } + + public Pattern getFilenameRegex() { + return filenameRegex; + } + + public void setFilenameRegex(Pattern filenameRegex) { + this.filenameRegex = filenameRegex; + } + + public boolean isPreserveTimestamp() { + return preserveTimestamp; + } + + public void setPreserveTimestamp(boolean preserveTimestamp) { + this.preserveTimestamp = preserveTimestamp; + } + + @AssertTrue(message = "filenamePattern and filenameRegex are mutually exclusive") + public boolean isExclusivePatterns() { + return !(this.filenamePattern != null && this.filenameRegex != null); + } + + public boolean isListOnly() { + return listOnly; + } + + public void setListOnly(boolean listOnly) { + this.listOnly = listOnly; + } + + public boolean isMultiSource() { + return this.directories != null && this.directories.length > 0; + } + + public int getMaxFetch() { + return maxFetch; + } + + public void setMaxFetch(int maxFetch) { + this.maxFetch = maxFetch; + } + + public boolean isFair() { + return this.fair; + } + + public void setFair(boolean fair) { + this.fair = fair; + } + + public Map getFactories() { + return this.factories; + } + + public void setFactories(Map factories) { + this.factories = factories; + } + + public String[] getDirectories() { + return this.directories; + } + + public void setDirectories(String[] directories) { + this.directories = directories; + } + + public boolean isStream() { + return stream; + } + + public void setStream(boolean stream) { + this.stream = stream; + } + + public Factory getFactory() { + return factory; + } + + public Duration getDelayWhenEmpty() { + return delayWhenEmpty; + } + + public void setDelayWhenEmpty(Duration delayWhenEmpty) { + this.delayWhenEmpty = delayWhenEmpty; + } + + static List keyDirectories(SftpSupplierProperties properties) { + List keyDirs = new ArrayList<>(); + Assert.isTrue(properties.getDirectories().length > 0, "At least one key.directory required"); + for (String keyDir : properties.getDirectories()) { + String[] split = keyDir.split("\\."); + Assert.isTrue(split.length == 2, () -> "key/directory can only have one '.': " + keyDir); + keyDirs.add(new RotationPolicy.KeyDirectory(split[0], split[1])); + } + return keyDirs; + } + + public static class Factory { + + /** + * The host name of the server. + */ + private String host = "localhost"; + + /** + * The username to use to connect to the server. + */ + + private String username; + + /** + * The password to use to connect to the server. + */ + private String password; + + /** + * The port of the server. + */ + private int port = 22; + + /** + * Resource location of user's private key. + */ + private Resource privateKey; + + /** + * Passphrase for user's private key. + */ + private String passPhrase = ""; + + /** + * True to allow an unknown or changed key. + */ + private boolean allowUnknownKeys = false; + + /** + * A SpEL expression resolving to the location of the known hosts file. + */ + private Expression knownHostsExpression = null; + + @NotBlank + public String getHost() { + return this.host; + } + + public void setHost(String host) { + this.host = host; + } + + @NotBlank + public String getUsername() { + return this.username; + } + + public void setUsername(String username) { + this.username = username; + } + + public String getPassword() { + return this.password; + } + + public void setPassword(String password) { + this.password = password; + } + + @Range(min = 0, max = 65535) + public int getPort() { + return this.port; + } + + public void setPort(int port) { + this.port = port; + } + + public Resource getPrivateKey() { + return this.privateKey; + } + + public void setPrivateKey(Resource privateKey) { + this.privateKey = privateKey; + } + + public String getPassPhrase() { + return this.passPhrase; + } + + public void setPassPhrase(String passPhrase) { + this.passPhrase = passPhrase; + } + + public boolean isAllowUnknownKeys() { + return this.allowUnknownKeys; + } + + public void setAllowUnknownKeys(boolean allowUnknownKeys) { + this.allowUnknownKeys = allowUnknownKeys; + } + + public Expression getKnownHostsExpression() { + return this.knownHostsExpression; + } + + public void setKnownHostsExpression(Expression knownHosts) { + this.knownHostsExpression = knownHosts; + } + + } + +} diff --git a/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierRotator.java b/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierRotator.java new file mode 100644 index 00000000..4f3e827a --- /dev/null +++ b/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierRotator.java @@ -0,0 +1,63 @@ +/* + * Copyright 2018-2020 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 + * + * https://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.cloud.fn.supplier.sftp; + +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.file.remote.aop.RotatingServerAdvice; +import org.springframework.integration.file.remote.aop.StandardRotationPolicy; +import org.springframework.messaging.Message; +import org.springframework.messaging.support.MessageBuilder; + +/** + * An {@link RotatingServerAdvice} for listing files on multiple directories/servers. + * + * @author Gary Russell + * @author David Turanski + * @since 2.0 + */ +public class SftpSupplierRotator extends RotatingServerAdvice { + + private static String SFTP_SELECTED_SERVER_PROPERTY_KEY = "sftp_selectedServer"; + + private final SftpSupplierProperties properties; + + private final StandardRotationPolicy rotationPolicy; + + public SftpSupplierRotator(SftpSupplierProperties properties, StandardRotationPolicy rotationPolicy) { + super(rotationPolicy); + this.properties = properties; + this.rotationPolicy = rotationPolicy; + } + + public String getCurrentKey() { + return this.rotationPolicy.getCurrent().getKey().toString(); + } + + public String getCurrentDirectory() { + return this.rotationPolicy.getCurrent().getDirectory(); + } + + @Override + public Message afterReceive(Message result, MessageSource source) { + if (result != null) { + result = MessageBuilder.fromMessage(result) + .setHeader(SFTP_SELECTED_SERVER_PROPERTY_KEY, this.getCurrentKey()).build(); + } + this.rotationPolicy.afterReceive(result != null, source); + return result; + } +} diff --git a/functions/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java b/functions/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java new file mode 100644 index 00000000..c20f0b9f --- /dev/null +++ b/functions/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java @@ -0,0 +1,314 @@ +/* + * Copyright 2020-2020 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 + * + * https://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.cloud.fn.supplier.sftp; + +import java.io.File; +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.time.Duration; +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.cloud.fn.test.support.sftp.SftpTestSupport; +import org.springframework.http.MediaType; +import org.springframework.integration.file.splitter.FileSplitter; +import org.springframework.integration.json.JsonPathUtils; +import org.springframework.integration.metadata.MetadataStore; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.fail; +import static org.awaitility.Awaitility.await; + +public class SftpSupplierApplicationTests extends SftpTestSupport { + + ApplicationContextRunner defaultApplicationContextRunner; + + @BeforeEach + void setUpDefaultProperties() { + defaultApplicationContextRunner = new ApplicationContextRunner() + .withUserConfiguration(TestApp.class) + .withPropertyValues( + "sftp.supplier.factory.host=localhost", + "sftp.supplier.factory.port=${sftp.factory.port}", + "sftp.supplier.factory.username=user", + "sftp.supplier.factory.password=pass", + "sftp.supplier.factory.cache-sessions=true", + "sftp.supplier.factory.allowUnknownKeys=true", + "sftp.supplier.localDir=" + this.targetLocalDirectory.getAbsolutePath(), + "sftp.supplier.remoteDir=sftpSource"); + } + + @Test + void supplierForListOnly() { + defaultApplicationContextRunner + .withPropertyValues("sftp.supplier.listOnly=true") + .run(context -> { + Supplier>> sftpSupplier = context.getBean("sftpSupplier", + Supplier.class); + SftpSupplierProperties properties = context.getBean(SftpSupplierProperties.class); + HashSet fileNames = new HashSet<>(); + fileNames.add(String.join(properties.getRemoteFileSeparator(), properties.getRemoteDir(), + "sftpSource1.txt")); + fileNames.add(String.join(properties.getRemoteFileSeparator(), properties.getRemoteDir(), + "sftpSource2.txt")); + final AtomicReference> expectedFileNames = new AtomicReference<>(fileNames); + StepVerifier.create(sftpSupplier.get()) + .assertNext(message -> { + assertThat(expectedFileNames.get()).contains(message.getPayload()); + expectedFileNames.get().remove(message.getPayload()); + assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE)) + .isEqualTo(MediaType.TEXT_PLAIN); + }) + .assertNext(message -> { + assertThat(expectedFileNames.get()).contains(message.getPayload()); + assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE)) + .isEqualTo(MediaType.TEXT_PLAIN); + }) + .thenCancel() + .verify(Duration.ofSeconds(10)); + + }); + } + + @Test + void supplierForFileRef() { + defaultApplicationContextRunner + .withPropertyValues( + "sftp.supplier.localDir=" + getTargetLocalDirectory().getAbsolutePath(), + "file.consumer.mode=ref") + .run(context -> { + Supplier>> sftpSupplier = context.getBean("sftpSupplier", + Supplier.class); + SftpSupplierProperties properties = context.getBean(SftpSupplierProperties.class); + MetadataStore metadataStore = context.getBean(MetadataStore.class); + HashSet fileNames = new HashSet<>(); + fileNames.add(properties.getLocalDir() + File.separator + "sftpSource1.txt"); + fileNames.add(properties.getLocalDir() + File.separator + "sftpSource2.txt"); + final AtomicReference> expectedFileNames = new AtomicReference<>(fileNames); + StepVerifier.create(sftpSupplier.get()) + .assertNext(message -> { + File file = (File) message.getPayload(); + assertThat(expectedFileNames.get()).contains(file.getAbsolutePath()); + expectedFileNames.get().remove(file.getAbsolutePath()); + }) + .expectNextMatches( + message -> expectedFileNames.get().contains(message.getPayload().getAbsolutePath())) + .thenCancel() + .verify(Duration.ofSeconds(10)); + + assertThat(metadataStore.get("sftpSource/sftpSource1.txt")).isNotNull(); + assertThat(metadataStore.get("sftpSource/sftpSource2.txt")).isNotNull(); + assertThat(Files.exists(Paths.get(getTargetLocalDirectory().getAbsolutePath(), "sftpSource1.txt"))) + .isTrue(); + assertThat(Files.exists(Paths.get(getTargetLocalDirectory().getAbsolutePath(), "sftpSource2.txt"))) + .isTrue(); + }); + } + + @Test + void deleteRemoteFiles() { + defaultApplicationContextRunner + .withPropertyValues( + "sftp.supplier.stream=true", + "sftp.supplier.delete-remote-files=true") + .run(context -> { + Supplier>> sftpSupplier = context.getBean("sftpSupplier", + Supplier.class); + StepVerifier.create(sftpSupplier.get()) + .expectNextMatches(message -> message.getPayload().length > 0) + .expectNextMatches(message -> message.getPayload().length > 0) + .thenCancel() + .verify(Duration.ofSeconds(10)); + await().atMost(Duration.ofSeconds(10)) + .until(() -> getSourceRemoteDirectory().list().length == 0); + }); + + } + + @Test + public void streamSourceFilesInLineMode() { + defaultApplicationContextRunner + .withPropertyValues( + "sftp.supplier.stream=true", + "sftp.supplier.factory.private-key = classpath:id_rsa_pp", + "sftp.supplier.factory.passphrase = secret", + "sftp.supplier.factory.password = badPassword", // ensure public key was used + "file.consumer.mode=lines", + "file.consumer.with-markers=true", + "file.consumer.markers-json=true") + .run(context -> { + Supplier>> sftpSupplier = context.getBean("sftpSupplier", + Supplier.class); + StepVerifier.create(sftpSupplier.get()) + .assertNext(message -> { + final Object evaluate; + try { + evaluate = JsonPathUtils.evaluate(message.getPayload(), "$.mark"); + assertThat(evaluate).isEqualTo(FileSplitter.FileMarker.Mark.START.name()); + } + catch (IOException e) { + fail(e.getMessage()); + } + }) + .expectNextMatches(message -> message.getPayload().startsWith("source")) + .assertNext(message -> { + final Object evaluate; + try { + evaluate = JsonPathUtils.evaluate(message.getPayload(), "$.mark"); + assertThat(evaluate).isEqualTo(FileSplitter.FileMarker.Mark.END.name()); + } + catch (IOException e) { + fail(e.getMessage()); + } + }) + .thenCancel() + .verify(Duration.ofSeconds(10)); + }); + } + + @Test + void supplierWithMultiSourceAndStreamContentsSource3ComesSecond() throws Exception { + Path newSource = createNewRemoteSource( + Paths.get(remoteTemporaryFolder.toString(), "sftpSecondSource", "doesNotMatter.txt"), + "source3"); + new ApplicationContextRunner() + .withUserConfiguration(TestApp.class) + .withPropertyValues( + "sftp.supplier.stream=true", + "sftp.supplier.factories.one.host=localhost", + "sftp.supplier.factories.one.port=${sftp.factory.port}", + "sftp.supplier.factories.one.username=user", + "sftp.supplier.factories.one.password=pass", + "sftp.supplier.factories.one.cache-sessions=true", + "sftp.supplier.factories.one.allowUnknownKeys=true", + "sftp.supplier.factories.two.host=localhost", + "sftp.supplier.factories.two.port=${sftp.factory.port}", + "sftp.supplier.factories.two.username = user", + "sftp.supplier.factories.two.password = pass", + "sftp.supplier.factories.two.cache-sessions = true", + "sftp.supplier.factories.two.allowUnknownKeys = true", + "sftp.supplier.directories=one.sftpSource,two.sftpSecondSource", + "sftp.supplier.max-fetch=1", + "sftp.supplier.fair=true") + .run(context -> { + + Supplier>> sftpSupplier = context.getBean("sftpSupplier", Supplier.class); + HashSet contents = new HashSet<>(); + contents.add("source1"); + contents.add("source2"); + final AtomicReference> expectedContentsOfAllFiles = new AtomicReference<>(contents); + + StepVerifier.create(sftpSupplier.get()) + .assertNext(message -> { + String payload = new String(message.getPayload()); + assertThat(expectedContentsOfAllFiles.get()).contains(payload); + expectedContentsOfAllFiles.get().remove(payload); + }) + .expectNextMatches(message -> new String(message.getPayload()).equals("source3")) + .expectNextMatches(message -> expectedContentsOfAllFiles.get() + .contains(new String(message.getPayload()))) + .thenCancel() + .verify(Duration.ofSeconds(10)); + }); + + deleteNewSource(newSource); + } + + @Test + void supplierMultiSourceRefTestsFor200Alex() throws Exception { + Path newSource = createNewRemoteSource( + Paths.get(remoteTemporaryFolder.toString(), "sftpSecondSource", "sftpSource3.txt"), + "doesNotMatter"); + new ApplicationContextRunner() + .withUserConfiguration(TestApp.class) + .withPropertyValues( + "file.consumer.mode = ref", + "sftp.supplier.localDir=" + this.targetLocalDirectory.getAbsolutePath(), + "sftp.supplier.factories.one.host=localhost", + "sftp.supplier.factories.one.port=${sftp.factory.port}", + "sftp.supplier.factories.one.username = user", + "sftp.supplier.factories.one.password = pass", + "sftp.supplier.factories.one.cache-sessions = true", + "sftp.supplier.factories.one.allowUnknownKeys = true", + "sftp.supplier.factories.two.host=localhost", + "sftp.supplier.factories.two.port=${sftp.factory.port}", + "sftp.supplier.factories.two.username = user", + "sftp.supplier.factories.two.password = pass", + "sftp.supplier.factories.two.cache-sessions = true", + "sftp.supplier.factories.two.allowUnknownKeys = true", + "sftp.supplier.factories.empty.host=localhost", + "sftp.supplier.factories.empty.port=${sftp.factory.port}", + "sftp.supplier.factories.empty.username=user", + "sftp.supplier.factories.empty.password=pass", + "sftp.supplier.factories.empty.allowUnknownKeys = true", + "sftp.supplier.directories=one.sftpSource,two.sftpSecondSource,empty.sftpSource", + "sftp.supplier.max-fetch=1", + "sftp.supplier.fair=true") + .run(context -> { + Supplier>> sftpSupplier = context.getBean("sftpSupplier", Supplier.class); + SftpSupplierProperties properties = context.getBean(SftpSupplierProperties.class); + String localDir = properties.getLocalDir().getPath(); + HashSet firstSourceFiles = new HashSet<>(); + firstSourceFiles.add(Paths.get(localDir, "sftpSource1.txt").toString()); + firstSourceFiles.add(Paths.get(localDir, "sftpSource2.txt").toString()); + final AtomicReference> expectedFirstSourcePaths = new AtomicReference<>( + firstSourceFiles); + + StepVerifier.create(sftpSupplier.get()) + .assertNext(message -> { + assertThat(expectedFirstSourcePaths.get()).contains(message.getPayload().getPath()); + expectedFirstSourcePaths.get().remove(message.getPayload().getPath()); + }) + .expectNextMatches(message -> message.getPayload().getPath().equals( + Paths.get(localDir, "sftpSource3.txt").toString())) + .expectNextMatches(message -> expectedFirstSourcePaths.get() + .contains(message.getPayload().getPath())) + .thenCancel() + .verify(Duration.ofSeconds(10)); + }); + deleteNewSource(newSource); + } + + private Path createNewRemoteSource(Path remotePath, String contents) throws Exception { + Files.createDirectory(remotePath.getParent()); + Files.write(Files.createFile(remotePath), contents.getBytes()); + return remotePath; + } + + private void deleteNewSource(Path newSource) throws IOException { + Files.delete(newSource); + Files.delete(newSource.getParent()); + } + + @SpringBootApplication + static class TestApp { + + } +} diff --git a/functions/supplier/sftp-supplier/src/test/resources/logback.xml b/functions/supplier/sftp-supplier/src/test/resources/logback.xml new file mode 100644 index 00000000..c85b56f1 --- /dev/null +++ b/functions/supplier/sftp-supplier/src/test/resources/logback.xml @@ -0,0 +1,16 @@ + + + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n + + + + + + + + + +