diff --git a/applications/sink/ftp-sink/README.adoc b/applications/sink/ftp-sink/README.adoc index ce99d9be..bbe268f9 100644 --- a/applications/sink/ftp-sink/README.adoc +++ b/applications/sink/ftp-sink/README.adoc @@ -40,6 +40,12 @@ $$ftp.consumer.remote-file-separator$$:: $$The remote file separator.$$ *($$Stri $$ftp.consumer.temporary-remote-dir$$:: $$A temporary directory where the file will be written if '#isUseTemporaryFilename()' is true.$$ *($$String$$, default: `$$/$$`)* $$ftp.consumer.tmp-file-suffix$$:: $$The suffix to use while the transfer is in progress.$$ *($$String$$, default: `$$.tmp$$`)* $$ftp.consumer.use-temporary-filename$$:: $$Whether or not to write to a temporary file and rename.$$ *($$Boolean$$, default: `$$true$$`)* +$$ftp.factory.cache-sessions$$:: $$Cache sessions.$$ *($$Boolean$$, default: `$$$$`)* +$$ftp.factory.client-mode$$:: $$The client mode to use for the FTP session.$$ *($$ClientMode$$, default: `$$$$`, possible values: `ACTIVE`,`PASSIVE`)* +$$ftp.factory.host$$:: $$The host name of the server.$$ *($$String$$, default: `$$localhost$$`)* +$$ftp.factory.password$$:: $$The password to use to connect to the server.$$ *($$String$$, default: `$$$$`)* +$$ftp.factory.port$$:: $$The port of the server.$$ *($$Integer$$, default: `$$21$$`)* +$$ftp.factory.username$$:: $$The username to use to connect to the server.$$ *($$String$$, default: `$$$$`)* //end::configuration-properties[] //end::ref-doc[] diff --git a/applications/sink/ftp-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/sink/ftp-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties index 43893b0d..c7aa0759 100644 --- a/applications/sink/ftp-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties +++ b/applications/sink/ftp-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -1 +1,2 @@ -configuration-properties.classes=org.springframework.cloud.fn.consumer.ftp.FtpConsumerProperties +configuration-properties.classes=org.springframework.cloud.fn.consumer.ftp.FtpConsumerProperties, \ + org.springframework.cloud.fn.common.ftp.FtpSessionFactoryProperties diff --git a/applications/source/file-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/source/file-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties index 92db2c7a..acfefa6f 100644 --- a/applications/source/file-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties +++ b/applications/source/file-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -1,2 +1,2 @@ configuration-properties.classes=org.springframework.cloud.fn.supplier.file.FileSupplierProperties,\ - org.springframework.cloud.fn.supplier.file.FileConsumerProperties + org.springframework.cloud.fn.common.file.FileConsumerProperties diff --git a/applications/source/ftp-source/README.adoc b/applications/source/ftp-source/README.adoc new file mode 100644 index 00000000..c37a4660 --- /dev/null +++ b/applications/source/ftp-source/README.adoc @@ -0,0 +1,98 @@ +//tag::ref-doc[] += FTP Source +This source application supports transfer of files using the FTP 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 also https://github.com/spring-cloud/stream-applications/blob/master/functions/common/metadata-store-common/README.adoc[`MetaDataStore`] options for possible shared persistent store configuration for the `FtpPersistentAcceptOnceFileListFilter` used in the FTP Source. + +== Input + +N/A (Fetches files from an FTP server). + +== Output + +=== mode = contents + +==== Headers: + +* `Content-Type: application/octet-stream` +* `file_originalFile: ` +* `file_name: ` + +==== Payload: + +A `byte[]` filled with the file contents. + +=== mode = lines + +==== Headers: + +* `Content-Type: text/plain` +* `file_orginalFile: ` +* `file_name: ` +* `correlationId: ` (same for each line) +* `sequenceNumber: ` +* `sequenceSize: 0` (number of lines is not know until the file is read) + +==== 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: + +None. + +==== 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: `$$$$`)* +$$ftp.factory.cache-sessions$$:: $$Cache sessions.$$ *($$Boolean$$, default: `$$$$`)* +$$ftp.factory.client-mode$$:: $$The client mode to use for the FTP session.$$ *($$ClientMode$$, default: `$$$$`, possible values: `ACTIVE`,`PASSIVE`)* +$$ftp.factory.host$$:: $$The host name of the server.$$ *($$String$$, default: `$$localhost$$`)* +$$ftp.factory.password$$:: $$The password to use to connect to the server.$$ *($$String$$, default: `$$$$`)* +$$ftp.factory.port$$:: $$The port of the server.$$ *($$Integer$$, default: `$$21$$`)* +$$ftp.factory.username$$:: $$The username to use to connect to the server.$$ *($$String$$, default: `$$$$`)* +$$ftp.supplier.auto-create-local-dir$$:: $$Set to true to create the local directory if it does not exist.$$ *($$Boolean$$, default: `$$true$$`)* +$$ftp.supplier.delay-when-empty$$:: $$Duration of delay when no new files are detected.$$ *($$Duration$$, default: `$$1s$$`)* +$$ftp.supplier.delete-remote-files$$:: $$Set to true to delete remote files after successful transfer.$$ *($$Boolean$$, default: `$$false$$`)* +$$ftp.supplier.filename-pattern$$:: $$A filter pattern to match the names of files to transfer.$$ *($$String$$, default: `$$$$`)* +$$ftp.supplier.filename-regex$$:: $$A filter regex pattern to match the names of files to transfer.$$ *($$Pattern$$, default: `$$$$`)* +$$ftp.supplier.local-dir$$:: $$The local directory to use for file transfers.$$ *($$File$$, default: `$$$$`)* +$$ftp.supplier.preserve-timestamp$$:: $$Set to true to preserve the original timestamp.$$ *($$Boolean$$, default: `$$true$$`)* +$$ftp.supplier.remote-dir$$:: $$The remote FTP directory.$$ *($$String$$, default: `$$/$$`)* +$$ftp.supplier.remote-file-separator$$:: $$The remote file separator.$$ *($$String$$, default: `$$/$$`)* +$$ftp.supplier.tmp-file-suffix$$:: $$The suffix to use while the transfer is in progress.$$ *($$String$$, default: `$$.tmp$$`)* +//end::configuration-properties[] + +== Examples + +``` +java -jar ftp_source.jar --ftp.remote-dir=foo --file.mode=lines --trigger.fixed-delay=60 --ftp.factory.host=ftpserver \ + --ftp.factory.username=user --ftp.factory.password=pw --ftp.local-dir=/foo +``` +//end::ref-doc[] diff --git a/applications/source/ftp-source/pom.xml b/applications/source/ftp-source/pom.xml new file mode 100644 index 00000000..c5b97d81 --- /dev/null +++ b/applications/source/ftp-source/pom.xml @@ -0,0 +1,83 @@ + + + 4.0.0 + ftp-source + 3.0.0-SNAPSHOT + ftp-source + ftp 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 + ftp-supplier + ${java-functions.version} + + + org.springframework.cloud.fn + function-test-support + ${java-functions.version} + + + + + + + org.springframework.cloud + spring-cloud-app-starter-doc-maven-plugin + + + org.springframework.cloud.stream.app.plugin + spring-cloud-stream-app-maven-plugin + + + ftp + source + ${project.version} + org.springframework.cloud.fn.supplier.ftp.FtpSupplierConfiguration.class + + + + org.springframework.cloud.fn + ftp-supplier + ${java-functions.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/ftp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/source/ftp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties new file mode 100644 index 00000000..b4779474 --- /dev/null +++ b/applications/source/ftp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -0,0 +1,3 @@ +configuration-properties.classes=org.springframework.cloud.fn.supplier.ftp.FtpSupplierProperties, \ + org.springframework.cloud.fn.common.ftp.FtpSessionFactoryProperties, \ + org.springframework.cloud.fn.common.file.FileConsumerProperties diff --git a/applications/source/ftp-source/src/test/java/org/springframework/cloud/stream/app/source/ftp/FtpSourceTests.java b/applications/source/ftp-source/src/test/java/org/springframework/cloud/stream/app/source/ftp/FtpSourceTests.java new file mode 100644 index 00000000..f41d62e4 --- /dev/null +++ b/applications/source/ftp-source/src/test/java/org/springframework/cloud/stream/app/source/ftp/FtpSourceTests.java @@ -0,0 +1,79 @@ +/* + * Copyright 2015-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.ftp; + +import java.io.File; + +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.fn.supplier.ftp.FtpSupplierConfiguration; +import org.springframework.cloud.fn.supplier.ftp.FtpSupplierProperties; +import org.springframework.cloud.fn.test.support.ftp.FtpTestSupport; +import org.springframework.cloud.stream.binder.test.OutputDestination; +import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Import; +import org.springframework.context.annotation.Primary; +import org.springframework.messaging.Message; +import org.springframework.test.annotation.DirtiesContext; + +import static org.assertj.core.api.Assertions.assertThat; + +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, + properties = { + "ftp.factory.username = foo", + "ftp.factory.password = foo", + "file.consumer.mode = ref", + "ftp.factory.cacheSessions = true" + }) +@DirtiesContext +public class FtpSourceTests extends FtpTestSupport { + + @Autowired + private OutputDestination output; + + @Autowired + FtpSupplierProperties config; + + @Test + public void testFtpource() { + Message message = output.receive(10000); + assertThat(new File(new String(message.getPayload()).replaceAll("\"", ""))).isEqualTo( + new File(this.config.getLocalDir(), "ftpSource1.txt")); + message = output.receive(10000); + assertThat(new File(new String(message.getPayload()).replaceAll("\"", ""))).isEqualTo( + new File(this.config.getLocalDir(), "ftpSource2.txt")); + } + + @SpringBootApplication + @Import({TestChannelBinderConfiguration.class, FtpSupplierConfiguration.class}) + public static class SampleConfiguration { + + @Bean + @Primary + public FtpSupplierProperties ftpSupplierProperties() { + final FtpSupplierProperties ftpSupplierProperties = new FtpSupplierProperties(); + ftpSupplierProperties.setRemoteDir("ftpSource"); + ftpSupplierProperties.setFilenamePattern("*"); + return ftpSupplierProperties; + } + + } +} diff --git a/applications/source/pom.xml b/applications/source/pom.xml index 012addbb..b0008c54 100644 --- a/applications/source/pom.xml +++ b/applications/source/pom.xml @@ -12,6 +12,7 @@ geode-source file-source + ftp-source jdbc-source http-source mqtt-source diff --git a/applications/stream-applications-core/common/stream-applications-metadata-store-common/src/main/resources/META-INF/spring.factories b/applications/stream-applications-core/common/stream-applications-metadata-store-common/src/main/resources/META-INF/spring.factories deleted file mode 100644 index adb3df0d..00000000 --- a/applications/stream-applications-core/common/stream-applications-metadata-store-common/src/main/resources/META-INF/spring.factories +++ /dev/null @@ -1,2 +0,0 @@ -org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ - org.springframework.cloud.stream.app.metadata.MetadataStoreAutoConfiguration diff --git a/applications/stream-applications-core/pom.xml b/applications/stream-applications-core/pom.xml index c3cd2b9a..38db85d4 100644 --- a/applications/stream-applications-core/pom.xml +++ b/applications/stream-applications-core/pom.xml @@ -56,7 +56,6 @@ common/stream-applications-test-support common/stream-applications-postprocessor-common common/stream-applications-micrometer-common - common/stream-applications-metadata-store-common common/stream-applications-task-launch-request-common common/stream-applications-security-common diff --git a/functions/common/file-common/pom.xml b/functions/common/file-common/pom.xml new file mode 100644 index 00000000..1c8c382c --- /dev/null +++ b/functions/common/file-common/pom.xml @@ -0,0 +1,36 @@ + + + 4.0.0 + file-common + 1.0.0-SNAPSHOT + file-common + file common + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.boot + spring-boot-starter-validation + + + org.springframework.boot + spring-boot-configuration-processor + provided + + + org.springframework.boot + spring-boot-starter-integration + + + org.springframework.integration + spring-integration-file + + + + diff --git a/functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileConsumerProperties.java b/functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/FileConsumerProperties.java similarity index 92% rename from functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileConsumerProperties.java rename to functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/FileConsumerProperties.java index c878cb04..ed6969ef 100644 --- a/functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileConsumerProperties.java +++ b/functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/FileConsumerProperties.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.fn.supplier.file; +package org.springframework.cloud.fn.common.file; import javax.validation.constraints.AssertTrue; import javax.validation.constraints.NotNull; @@ -23,9 +23,8 @@ import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.validation.annotation.Validated; /** - * TODO: This will be used in other apps like (S)FTP and S3. Therefore, it might be moved to a common place. * - * @author David Turanski + * @author David Turanski * @author Artem Bilan */ @ConfigurationProperties("file.consumer") diff --git a/functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileReadingMode.java b/functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/FileReadingMode.java similarity index 94% rename from functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileReadingMode.java rename to functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/FileReadingMode.java index bf807da2..c6db428e 100644 --- a/functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileReadingMode.java +++ b/functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/FileReadingMode.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.fn.supplier.file; +package org.springframework.cloud.fn.common.file; /** * Defines the supported modes of reading and processing files. diff --git a/functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileUtils.java b/functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/FileUtils.java similarity index 98% rename from functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileUtils.java rename to functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/FileUtils.java index 163d8699..9a56618b 100644 --- a/functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileUtils.java +++ b/functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/FileUtils.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.fn.supplier.file; +package org.springframework.cloud.fn.common.file; import java.util.Collections; diff --git a/applications/stream-applications-core/common/stream-applications-metadata-store-common/README.adoc b/functions/common/metadata-store-common/README.adoc similarity index 100% rename from applications/stream-applications-core/common/stream-applications-metadata-store-common/README.adoc rename to functions/common/metadata-store-common/README.adoc diff --git a/applications/stream-applications-core/common/stream-applications-metadata-store-common/pom.xml b/functions/common/metadata-store-common/pom.xml similarity index 84% rename from applications/stream-applications-core/common/stream-applications-metadata-store-common/pom.xml rename to functions/common/metadata-store-common/pom.xml index 1b4e6c50..7bd5cd07 100644 --- a/applications/stream-applications-core/common/stream-applications-metadata-store-common/pom.xml +++ b/functions/common/metadata-store-common/pom.xml @@ -1,19 +1,18 @@ - + 4.0.0 + metadata-store-common + 1.0.0-SNAPSHOT + metadata-store-common + metadata-store common - stream-applications-core - org.springframework.cloud.stream.app - 3.0.0-SNAPSHOT - ../.. + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent - stream-applications-metadata-store-common - stream-applications-metadata-store-common - 1.11.439 2.0.0.RELEASE @@ -27,6 +26,16 @@ org.springframework.integration spring-integration-core + + org.springframework.boot + spring-boot-starter + + + + org.springframework.boot + spring-boot-starter-test + test + diff --git a/applications/stream-applications-core/common/stream-applications-metadata-store-common/src/main/java/org/springframework/cloud/stream/app/metadata/ClientCacheAutoConfiguration.java b/functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/ClientCacheAutoConfiguration.java similarity index 96% rename from applications/stream-applications-core/common/stream-applications-metadata-store-common/src/main/java/org/springframework/cloud/stream/app/metadata/ClientCacheAutoConfiguration.java rename to functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/ClientCacheAutoConfiguration.java index f3d7e510..22aed185 100644 --- a/applications/stream-applications-core/common/stream-applications-metadata-store-common/src/main/java/org/springframework/cloud/stream/app/metadata/ClientCacheAutoConfiguration.java +++ b/functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/ClientCacheAutoConfiguration.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.app.metadata; +package org.springframework.cloud.fn.common.metadata.store; import org.apache.geode.cache.GemFireCache; import org.apache.geode.cache.client.ClientCache; diff --git a/applications/stream-applications-core/common/stream-applications-metadata-store-common/src/main/java/org/springframework/cloud/stream/app/metadata/MetadataStoreAutoConfiguration.java b/functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java similarity index 99% rename from applications/stream-applications-core/common/stream-applications-metadata-store-common/src/main/java/org/springframework/cloud/stream/app/metadata/MetadataStoreAutoConfiguration.java rename to functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java index d6799aa5..e185d9e3 100644 --- a/applications/stream-applications-core/common/stream-applications-metadata-store-common/src/main/java/org/springframework/cloud/stream/app/metadata/MetadataStoreAutoConfiguration.java +++ b/functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.app.metadata; +package org.springframework.cloud.fn.common.metadata.store; import com.amazonaws.auth.AWSCredentialsProvider; import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync; diff --git a/applications/stream-applications-core/common/stream-applications-metadata-store-common/src/main/java/org/springframework/cloud/stream/app/metadata/MetadataStoreProperties.java b/functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java similarity index 98% rename from applications/stream-applications-core/common/stream-applications-metadata-store-common/src/main/java/org/springframework/cloud/stream/app/metadata/MetadataStoreProperties.java rename to functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java index a8cc3ce1..39b898b1 100644 --- a/applications/stream-applications-core/common/stream-applications-metadata-store-common/src/main/java/org/springframework/cloud/stream/app/metadata/MetadataStoreProperties.java +++ b/functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.app.metadata; +package org.springframework.cloud.fn.common.metadata.store; import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; diff --git a/functions/common/metadata-store-common/src/main/resources/META-INF/spring.factories b/functions/common/metadata-store-common/src/main/resources/META-INF/spring.factories new file mode 100644 index 00000000..f9789086 --- /dev/null +++ b/functions/common/metadata-store-common/src/main/resources/META-INF/spring.factories @@ -0,0 +1,2 @@ +org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreAutoConfiguration diff --git a/applications/stream-applications-core/common/stream-applications-metadata-store-common/src/test/java/org/springframework/cloud/stream/app/metadata/MetadataStoreAutoConfigurationTests.java b/functions/common/metadata-store-common/src/test/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfigurationTests.java similarity index 98% rename from applications/stream-applications-core/common/stream-applications-metadata-store-common/src/test/java/org/springframework/cloud/stream/app/metadata/MetadataStoreAutoConfigurationTests.java rename to functions/common/metadata-store-common/src/test/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfigurationTests.java index 81faa0e5..eea2f36b 100644 --- a/applications/stream-applications-core/common/stream-applications-metadata-store-common/src/test/java/org/springframework/cloud/stream/app/metadata/MetadataStoreAutoConfigurationTests.java +++ b/functions/common/metadata-store-common/src/test/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfigurationTests.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.app.metadata; +package org.springframework.cloud.fn.common.metadata.store; import java.beans.Introspector; import java.nio.charset.StandardCharsets; diff --git a/functions/pom.xml b/functions/pom.xml index f582e55d..a81044f6 100644 --- a/functions/pom.xml +++ b/functions/pom.xml @@ -42,9 +42,11 @@ common/config-common + common/file-common common/ftp-common common/function-test-support common/geode-common + common/metadata-store-common common/mqtt-common common/tcp-common @@ -71,6 +73,7 @@ function/splitter-function supplier/file-supplier + supplier/ftp-supplier supplier/geode-supplier supplier/http-supplier supplier/jdbc-supplier diff --git a/functions/supplier/file-supplier/pom.xml b/functions/supplier/file-supplier/pom.xml index 422cb67c..0bd10578 100644 --- a/functions/supplier/file-supplier/pom.xml +++ b/functions/supplier/file-supplier/pom.xml @@ -22,6 +22,11 @@ org.springframework.boot spring-boot-starter-integration + + org.springframework.cloud.fn + file-common + ${project.version} + org.springframework.boot spring-boot-starter-json diff --git a/functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java b/functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java index 3261c4fe..5761713f 100644 --- a/functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java +++ b/functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java @@ -27,6 +27,9 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.file.FileConsumerProperties; +import org.springframework.cloud.fn.common.file.FileReadingMode; +import org.springframework.cloud.fn.common.file.FileUtils; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Lazy; diff --git a/functions/supplier/ftp-supplier/README.adoc b/functions/supplier/ftp-supplier/README.adoc new file mode 100644 index 00000000..3cc9bb90 --- /dev/null +++ b/functions/supplier/ftp-supplier/README.adoc @@ -0,0 +1,35 @@ +# FTP Supplier + +This module provides a FTP supplier that can be reused and composed in other applications. +The `Supplier` uses the `FtpInboundChannelAdapter` from Spring Integration. +`FtpSupplier` is implemented as a `java.util.function.Supplier`. +This supplier gives you a reactive stream of files from the provided directory 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 `FtpSupplierConfiguration` in the application and then inject the following bean. + +`ftpSupplier` + +You need to inject this as `Supplier>>`. + +You can use `ftpSupplier` 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/ftp/FtpSupplierProperties.java[FtpSupplierProperties]. +See link:src/main/java/org/springframework/cloud/fn/supplier/file/FileConsumerProperties.java[this] also. + +## Tests + +See this link:src/test/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierTests.java[test suite] for the various ways, this supplier is used. + +## Other usage + +See this https://github.com/spring-cloud/stream-applications/blob/master/applications/source/ftp-source/README.adoc[README] where this supplier is used to create a Spring Cloud Stream application where it makes a File Source. \ No newline at end of file diff --git a/functions/supplier/ftp-supplier/pom.xml b/functions/supplier/ftp-supplier/pom.xml new file mode 100644 index 00000000..e509e702 --- /dev/null +++ b/functions/supplier/ftp-supplier/pom.xml @@ -0,0 +1,72 @@ + + + 4.0.0 + ftp-supplier + 1.0.0-SNAPSHOT + ftp-supplier + ftp supplier + + + org.springframework.cloud.fn + spring-functions-parent + 1.0.0-SNAPSHOT + ../../spring-functions-parent + + + + + org.springframework.integration + spring-integration-ftp + + + 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-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/ftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierConfiguration.java b/functions/supplier/ftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierConfiguration.java new file mode 100644 index 00000000..df504f47 --- /dev/null +++ b/functions/supplier/ftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierConfiguration.java @@ -0,0 +1,144 @@ +/* + * Copyright 2015-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.ftp; + +import java.io.File; +import java.util.function.Supplier; + +import org.apache.commons.net.ftp.FTPFile; +import org.reactivestreams.Publisher; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.file.FileConsumerProperties; +import org.springframework.cloud.fn.common.file.FileReadingMode; +import org.springframework.cloud.fn.common.file.FileUtils; +import org.springframework.cloud.fn.common.ftp.FtpSessionFactoryConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.context.annotation.Lazy; +import org.springframework.integration.core.MessageSource; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.file.filters.ChainFileListFilter; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.ftp.dsl.Ftp; +import org.springframework.integration.ftp.dsl.FtpInboundChannelAdapterSpec; +import org.springframework.integration.ftp.filters.FtpPersistentAcceptOnceFileListFilter; +import org.springframework.integration.ftp.filters.FtpRegexPatternFileListFilter; +import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilter; +import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizingMessageSource; +import org.springframework.integration.metadata.ConcurrentMetadataStore; +import org.springframework.integration.util.IntegrationReactiveUtils; +import org.springframework.messaging.Message; +import org.springframework.util.StringUtils; + +/** + * @author David Turanski + * @author Marius Bogoevici + * @author Gary Russell + * @author Artem Bilan + * @author Christian Tzolov + */ +@Configuration +@EnableConfigurationProperties({FtpSupplierProperties.class, FileConsumerProperties.class}) +@Import(FtpSessionFactoryConfiguration.class) +public class FtpSupplierConfiguration { + + private final FtpSupplierProperties ftpSupplierProperties; + + private final FileConsumerProperties fileConsumerProperties; + + private ConcurrentMetadataStore metadataStore; + + SessionFactory ftpSessionFactory; + + @Autowired + @Lazy + @Qualifier("ftpMessageSource") + private FtpInboundFileSynchronizingMessageSource ftpMessageSource; + + public FtpSupplierConfiguration(FtpSupplierProperties ftpSupplierProperties, + FileConsumerProperties fileConsumerProperties, + ConcurrentMetadataStore metadataStore, + SessionFactory ftpSessionFactory) { + this.ftpSupplierProperties = ftpSupplierProperties; + this.fileConsumerProperties = fileConsumerProperties; + this.metadataStore = metadataStore; + this.ftpSessionFactory = ftpSessionFactory; + } + + @Bean + public MessageSource ftpMessageSource() { + FtpInboundChannelAdapterSpec messageSourceBuilder = Ftp.inboundAdapter(ftpSessionFactory) + .preserveTimestamp(this.ftpSupplierProperties.isPreserveTimestamp()) + .remoteDirectory(this.ftpSupplierProperties.getRemoteDir()) + .remoteFileSeparator(this.ftpSupplierProperties.getRemoteFileSeparator()) + .localDirectory(this.ftpSupplierProperties.getLocalDir()) + .autoCreateLocalDirectory(this.ftpSupplierProperties.isAutoCreateLocalDir()) + .temporaryFileSuffix(this.ftpSupplierProperties.getTmpFileSuffix()) + .deleteRemoteFiles(this.ftpSupplierProperties.isDeleteRemoteFiles()); + + ChainFileListFilter chainFileListFilter = new ChainFileListFilter<>(); + + if (StringUtils.hasText(this.ftpSupplierProperties.getFilenamePattern())) { + chainFileListFilter.addFilter(new FtpSimplePatternFileListFilter(this.ftpSupplierProperties.getFilenamePattern())); + } + else if (this.ftpSupplierProperties.getFilenameRegex() != null) { + chainFileListFilter.addFilter(new FtpRegexPatternFileListFilter(this.ftpSupplierProperties.getFilenameRegex())); + } + + chainFileListFilter.addFilter(new FtpPersistentAcceptOnceFileListFilter(this.metadataStore, "ftpSource/")); + + messageSourceBuilder.filter(chainFileListFilter); + return messageSourceBuilder.get(); + } + + @Bean + public Flux> ftpMessageFlux() { + return Mono.>create(monoSink -> + monoSink.onRequest(value -> + monoSink.success(this.ftpMessageSource.receive()))) + .subscribeOn(Schedulers.boundedElastic()) + .repeatWhenEmpty(it -> it.delayElements(this.ftpSupplierProperties.getDelayWhenEmpty())) + .repeat(); + } + + @Bean + @ConditionalOnExpression("environment['file.consumer.mode'] != 'ref'") + public Publisher> ftpReadingFlow() { + return FileUtils.enhanceFlowForReadingMode(IntegrationFlows + .from(IntegrationReactiveUtils.messageSourceToFlux(ftpMessageSource())), fileConsumerProperties) + .toReactivePublisher(); + } + + @Bean + public Supplier>> ftpSupplier() { + if (this.fileConsumerProperties.getMode() == FileReadingMode.ref) { + return this::ftpMessageFlux; + } + else { + return () -> Flux.from(ftpReadingFlow()); + } + } + +} diff --git a/functions/supplier/ftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierProperties.java b/functions/supplier/ftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierProperties.java new file mode 100644 index 00000000..3a71bdcb --- /dev/null +++ b/functions/supplier/ftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierProperties.java @@ -0,0 +1,178 @@ +/* + * Copyright 2015-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.ftp; + +import java.io.File; +import java.time.Duration; +import java.util.regex.Pattern; + +import javax.validation.constraints.AssertTrue; +import javax.validation.constraints.NotBlank; +import javax.validation.constraints.NotNull; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.validation.annotation.Validated; + +/** + * @author David Turanski + * @author Gary Russell + * @author Artem Bilan + */ +@ConfigurationProperties("ftp.supplier") +@Validated +public class FtpSupplierProperties { + + /** + * 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"), "ftp-source"); + + /** + * 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; + + /** + * Duration of delay when no new files are detected. + */ + private Duration delayWhenEmpty = Duration.ofSeconds(1); + + public boolean isAutoCreateLocalDir() { + return this.autoCreateLocalDir; + } + + public void setAutoCreateLocalDir(boolean autoCreateLocalDir) { + this.autoCreateLocalDir = autoCreateLocalDir; + } + + public boolean isDeleteRemoteFiles() { + return this.deleteRemoteFiles; + } + + public void setDeleteRemoteFiles(boolean deleteRemoteFiles) { + this.deleteRemoteFiles = deleteRemoteFiles; + } + + @NotNull + public File getLocalDir() { + return this.localDir; + } + + public final void setLocalDir(File localDir) { + this.localDir = localDir; + } + + public String getFilenamePattern() { + return this.filenamePattern; + } + + public void setFilenamePattern(String filenamePattern) { + this.filenamePattern = filenamePattern; + } + + public Pattern getFilenameRegex() { + return this.filenameRegex; + } + + public void setFilenameRegex(Pattern filenameRegex) { + this.filenameRegex = filenameRegex; + } + + public boolean isPreserveTimestamp() { + return this.preserveTimestamp; + } + + public void setPreserveTimestamp(boolean preserveTimestamp) { + this.preserveTimestamp = preserveTimestamp; + } + + @NotBlank + public String getRemoteDir() { + return this.remoteDir; + } + + public final void setRemoteDir(String remoteDir) { + this.remoteDir = remoteDir; + } + + @NotBlank + public String getTmpFileSuffix() { + return this.tmpFileSuffix; + } + + public void setTmpFileSuffix(String tmpFileSuffix) { + this.tmpFileSuffix = tmpFileSuffix; + } + + @NotBlank + public String getRemoteFileSeparator() { + return this.remoteFileSeparator; + } + + public void setRemoteFileSeparator(String remoteFileSeparator) { + this.remoteFileSeparator = remoteFileSeparator; + } + + @AssertTrue(message = "filenamePattern and filenameRegex are mutually exclusive") + public boolean isExclusivePatterns() { + return !(this.filenamePattern != null && this.filenameRegex != null); + } + + public Duration getDelayWhenEmpty() { + return delayWhenEmpty; + } + + public void setDelayWhenEmpty(Duration delayWhenEmpty) { + this.delayWhenEmpty = delayWhenEmpty; + } + +} diff --git a/functions/supplier/ftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierPropertiesTests.java b/functions/supplier/ftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierPropertiesTests.java new file mode 100644 index 00000000..efe08e72 --- /dev/null +++ b/functions/supplier/ftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierPropertiesTests.java @@ -0,0 +1,139 @@ +/* + * Copyright 2015-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.ftp; + +import java.io.File; + +import org.junit.jupiter.api.Test; + +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.boot.test.util.TestPropertyValues; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Configuration; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author David Turanski + * @author Gary Russell + * @author Artem Bilan + */ +public class FtpSupplierPropertiesTests { + + @Test + public void localDirCanBeCustomized() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + TestPropertyValues.of("ftp.supplier.localDir:local") + .applyTo(context); + context.register(Conf.class); + context.refresh(); + FtpSupplierProperties properties = context.getBean(FtpSupplierProperties.class); + assertThat(properties.getLocalDir()).isEqualTo(new File("local")); + context.close(); + } + + @Test + public void remoteDirCanBeCustomized() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + TestPropertyValues.of("ftp.supplier.remoteDir:/remote") + .applyTo(context); + context.register(Conf.class); + context.refresh(); + FtpSupplierProperties properties = context.getBean(FtpSupplierProperties.class); + assertThat(properties.getRemoteDir()).isEqualTo("/remote"); + context.close(); + } + + @Test + public void deleteRemoteFilesCanBeEnabled() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + TestPropertyValues.of("ftp.supplier.deleteRemoteFiles:true") + .applyTo(context); + context.register(Conf.class); + context.refresh(); + FtpSupplierProperties properties = context.getBean(FtpSupplierProperties.class); + assertThat(properties.isDeleteRemoteFiles()).isTrue(); + context.close(); + } + + @Test + public void autoCreateLocalDirCanBeDisabled() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + TestPropertyValues.of("ftp.supplier.autoCreateLocalDir:false") + .applyTo(context); + context.register(Conf.class); + context.refresh(); + FtpSupplierProperties properties = context.getBean(FtpSupplierProperties.class); + assertThat(!properties.isAutoCreateLocalDir()).isTrue(); + context.close(); + } + + @Test + public void tmpFileSuffixCanBeCustomized() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + TestPropertyValues.of("ftp.supplier.tmpFileSuffix:.foo") + .applyTo(context); + context.register(Conf.class); + context.refresh(); + FtpSupplierProperties properties = context.getBean(FtpSupplierProperties.class); + assertThat(properties.getTmpFileSuffix()).isEqualTo(".foo"); + context.close(); + } + + @Test + public void filenamePatternCanBeCustomized() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + TestPropertyValues.of("ftp.supplier.filenamePattern:*.foo") + .applyTo(context); + context.register(Conf.class); + context.refresh(); + FtpSupplierProperties properties = context.getBean(FtpSupplierProperties.class); + assertThat(properties.getFilenamePattern()).isEqualTo("*.foo"); + context.close(); + } + + @Test + public void remoteFileSeparatorCanBeCustomized() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + TestPropertyValues.of("ftp.supplier.remoteFileSeparator:\\") + .applyTo(context); + context.register(Conf.class); + context.refresh(); + FtpSupplierProperties properties = context.getBean(FtpSupplierProperties.class); + assertThat(properties.getRemoteFileSeparator()).isEqualTo("\\"); + context.close(); + } + + + @Test + public void preserveTimestampDirCanBeDisabled() { + AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); + TestPropertyValues.of("ftp.supplier.preserveTimestamp:false") + .applyTo(context); + context.register(Conf.class); + context.refresh(); + FtpSupplierProperties properties = context.getBean(FtpSupplierProperties.class); + assertThat(!properties.isPreserveTimestamp()).isTrue(); + context.close(); + } + + @Configuration + @EnableConfigurationProperties(FtpSupplierProperties.class) + static class Conf { + + } +} diff --git a/functions/supplier/ftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierTests.java b/functions/supplier/ftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierTests.java new file mode 100644 index 00000000..6fbcfe16 --- /dev/null +++ b/functions/supplier/ftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierTests.java @@ -0,0 +1,99 @@ +/* + * Copyright 2015-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.ftp; + +import java.io.File; +import java.util.function.Supplier; + +import org.apache.commons.net.ftp.FTPFile; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.cloud.fn.test.support.ftp.FtpTestSupport; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Primary; +import org.springframework.integration.file.remote.session.CachingSessionFactory; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.metadata.ConcurrentMetadataStore; +import org.springframework.messaging.Message; +import org.springframework.test.annotation.DirtiesContext; + +import static org.assertj.core.api.Assertions.assertThat; + +@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE, + properties = { + "debug=true", + "ftp.factory.username = foo", + "ftp.factory.password = foo", + "file.consumer.mode = ref", + "ftp.factory.cacheSessions = true" + }) +@DirtiesContext +public class FtpSupplierTests extends FtpTestSupport { + + @Autowired + Supplier>> ftpSupplier; + + @Autowired + private ConcurrentMetadataStore metadataStore; + + @Autowired + FtpSupplierProperties config; + + @Autowired + SessionFactory sessionFactory; + + @Test + public void testSourceFileAsRef() { + final Flux> messageFlux = ftpSupplier.get(); + assertThat(this.sessionFactory).isInstanceOf(CachingSessionFactory.class); + StepVerifier stepVerifier = + StepVerifier.create(messageFlux) + .assertNext((message) -> { + assertThat(new File(message.getPayload().toString().replaceAll("\"", ""))).isEqualTo( + new File(this.config.getLocalDir(), "ftpSource1.txt")); + } + ) + .assertNext((message) -> { + assertThat(new File(message.getPayload().toString().replaceAll("\"", ""))).isEqualTo( + new File(this.config.getLocalDir(), "ftpSource2.txt")); + }) + .thenCancel() + .verifyLater(); + stepVerifier.verify(); + } + + @SpringBootApplication + static class TestApplication { + + // These properties can be moved into the SpringBootApplication annotation, but providing here + // as a way to demonstrate how we can provide ConfigurationProperties as a bean in the application itself + // This way, it is less error-prone from typos. + @Bean + @Primary + public FtpSupplierProperties ftpSupplierProperties() { + final FtpSupplierProperties ftpSupplierProperties = new FtpSupplierProperties(); + ftpSupplierProperties.setRemoteDir("ftpSource"); + ftpSupplierProperties.setFilenamePattern("*"); + return ftpSupplierProperties; + } + } +}