SpringCloud/stream-applications
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

FTP supplier/source

Browse Source 
* Addressing PR review comments
This commit is contained in:
Soby Chacko
2020-06-16 15:31:22 -04:00
committed by GitHub
parent 773b8c20f9
commit 50660a884e
30 changed files with 1016 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
applications/sink/ftp-sink/README.adoc
View File

@@ -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: `$$<none>$$`)*
$$ftp.factory.client-mode$$:: $$The client mode to use for the FTP session.$$ *($$ClientMode$$, default: `$$<none>$$`, 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: `$$<none>$$`)*
$$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: `$$<none>$$`)*
//end::configuration-properties[]
//end::ref-doc[]

3
applications/sink/ftp-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties
View File

@@ -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

2
applications/source/file-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties
View File

@@ -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

98
applications/source/ftp-source/README.adoc Normal file
View File

@@ -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: <java.io.File>`
* `file_name: <file name>`
==== Payload:
A `byte[]` filled with the file contents.
=== mode = lines
==== Headers:
* `Content-Type: text/plain`
* `file_orginalFile: <java.io.File>`
* `file_name: <file name>`
* `correlationId: <UUID>` (same for each line)
* `sequenceNumber: <n>`
* `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: `$$<none>$$`, 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: `$$<none>$$`)*
$$ftp.factory.cache-sessions$$:: $$Cache sessions.$$ *($$Boolean$$, default: `$$<none>$$`)*
$$ftp.factory.client-mode$$:: $$The client mode to use for the FTP session.$$ *($$ClientMode$$, default: `$$<none>$$`, 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: `$$<none>$$`)*
$$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: `$$<none>$$`)*
$$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: `$$<none>$$`)*
$$ftp.supplier.filename-regex$$:: $$A filter regex pattern to match the names of files to transfer.$$ *($$Pattern$$, default: `$$<none>$$`)*
$$ftp.supplier.local-dir$$:: $$The local directory to use for file transfers.$$ *($$File$$, default: `$$<none>$$`)*
$$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[]

83
applications/source/ftp-source/pom.xml Normal file
View File

@@ -0,0 +1,83 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>ftp-source</artifactId>
<version>3.0.0-SNAPSHOT</version>
<name>ftp-source</name>
<description>ftp source apps</description>
<packaging>jar</packaging>
<parent>
<groupId>org.springframework.cloud.stream.app</groupId>
<artifactId>stream-applications-core</artifactId>
<version>3.0.0-SNAPSHOT</version>
<relativePath/>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>ftp-supplier</artifactId>
<version>${java-functions.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>function-test-support</artifactId>
<version>${java-functions.version}</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-app-starter-doc-maven-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.springframework.cloud.stream.app.plugin</groupId>
<artifactId>spring-cloud-stream-app-maven-plugin</artifactId>
<configuration>
<generatedApp>
<name>ftp</name>
<type>source</type>
<version>${project.version}</version>
<configClass>org.springframework.cloud.fn.supplier.ftp.FtpSupplierConfiguration.class</configClass>
</generatedApp>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>ftp-supplier</artifactId>
<version>${java-functions.version}</version>
</dependency>
</dependencies>
</configuration>
</plugin>
</plugins>
</build>
<repositories>
<repository>
<snapshots>
<enabled>true</enabled>
</snapshots>
<id>spring-snapshots</id>
<name>Spring Snapshots</name>
<url>https://repo.spring.io/libs-snapshot-local</url>
</repository>
<repository>
<snapshots>
<enabled>false</enabled>
</snapshots>
<id>spring-milestones</id>
<name>Spring Milestones</name>
<url>https://repo.spring.io/libs-milestone-local</url>
</repository>
</repositories>
</project>

3
applications/source/ftp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties Normal file
View File

@@ -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

79
applications/source/ftp-source/src/test/java/org/springframework/cloud/stream/app/source/ftp/FtpSourceTests.java Normal file
View File

@@ -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<byte[]> 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;
}
}
}

1
applications/source/pom.xml
View File

@@ -12,6 +12,7 @@
<modules>
<module>geode-source</module>
<module>file-source</module>
<module>ftp-source</module>
<module>jdbc-source</module>
<module>http-source</module>
<module>mqtt-source</module>

2
applications/stream-applications-core/common/stream-applications-metadata-store-common/src/main/resources/META-INF/spring.factories
View File

@@ -1,2 +0,0 @@
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.stream.app.metadata.MetadataStoreAutoConfiguration

1
applications/stream-applications-core/pom.xml
View File

@@ -56,7 +56,6 @@
<module>common/stream-applications-test-support</module>
<module>common/stream-applications-postprocessor-common</module>
<module>common/stream-applications-micrometer-common</module>
<module>common/stream-applications-metadata-store-common</module>
<module>common/stream-applications-task-launch-request-common</module>
<module>common/stream-applications-security-common</module>
</modules>

36
functions/common/file-common/pom.xml Normal file
View File

@@ -0,0 +1,36 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>file-common</artifactId>
<version>1.0.0-SNAPSHOT</version>
<name>file-common</name>
<description>file common</description>
<parent>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>spring-functions-parent</artifactId>
<version>1.0.0-SNAPSHOT</version>
<relativePath>../../spring-functions-parent</relativePath>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-file</artifactId>
</dependency>
</dependencies>
</project>

5
functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileConsumerProperties.java → functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/FileConsumerProperties.java
View File

@@ -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")

2
functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileReadingMode.java → functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/FileReadingMode.java
View File

@@ -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.

2
functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileUtils.java → functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/FileUtils.java
View File

@@ -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;

0
applications/stream-applications-core/common/stream-applications-metadata-store-common/README.adoc → functions/common/metadata-store-common/README.adoc
View File

29
applications/stream-applications-core/common/stream-applications-metadata-store-common/pom.xml → functions/common/metadata-store-common/pom.xml
View File

@@ -1,19 +1,18 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns="http://maven.apache.org/POM/4.0.0"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>metadata-store-common</artifactId>
<version>1.0.0-SNAPSHOT</version>
<name>metadata-store-common</name>
<description>metadata-store common</description>
<parent>
<artifactId>stream-applications-core</artifactId>
<groupId>org.springframework.cloud.stream.app</groupId>
<version>3.0.0-SNAPSHOT</version>
<relativePath>../..</relativePath>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>spring-functions-parent</artifactId>
<version>1.0.0-SNAPSHOT</version>
<relativePath>../../spring-functions-parent</relativePath>
</parent>
<artifactId>stream-applications-metadata-store-common</artifactId>
<name>stream-applications-metadata-store-common</name>
<properties>
<aws-java-sdk.version>1.11.439</aws-java-sdk.version>
<spring-integration-aws.version>2.0.0.RELEASE</spring-integration-aws.version>
@@ -27,6 +26,16 @@
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<!--Redis-->
<dependency>

2
applications/stream-applications-core/common/stream-applications-metadata-store-common/src/main/java/org/springframework/cloud/stream/app/metadata/ClientCacheAutoConfiguration.java → functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/ClientCacheAutoConfiguration.java
View File

@@ -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;

2
applications/stream-applications-core/common/stream-applications-metadata-store-common/src/main/java/org/springframework/cloud/stream/app/metadata/MetadataStoreAutoConfiguration.java → functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java
View File

@@ -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;

2
applications/stream-applications-core/common/stream-applications-metadata-store-common/src/main/java/org/springframework/cloud/stream/app/metadata/MetadataStoreProperties.java → functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java
View File

@@ -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;

2
functions/common/metadata-store-common/src/main/resources/META-INF/spring.factories Normal file
View File

@@ -0,0 +1,2 @@
org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.springframework.cloud.fn.common.metadata.store.MetadataStoreAutoConfiguration

2
applications/stream-applications-core/common/stream-applications-metadata-store-common/src/test/java/org/springframework/cloud/stream/app/metadata/MetadataStoreAutoConfigurationTests.java → functions/common/metadata-store-common/src/test/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfigurationTests.java
View File

@@ -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;

3
functions/pom.xml
View File

@@ -42,9 +42,11 @@
<modules>
<module>common/config-common</module>
<module>common/file-common</module>
<module>common/ftp-common</module>
<module>common/function-test-support</module>
<module>common/geode-common</module>
<module>common/metadata-store-common</module>
<module>common/mqtt-common</module>
<module>common/tcp-common</module>
@@ -71,6 +73,7 @@
<module>function/splitter-function</module>
<module>supplier/file-supplier</module>
<module>supplier/ftp-supplier</module>
<module>supplier/geode-supplier</module>
<module>supplier/http-supplier</module>
<module>supplier/jdbc-supplier</module>

5
functions/supplier/file-supplier/pom.xml
View File

@@ -22,6 +22,11 @@
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>file-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-json</artifactId>

3
functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java
View File

@@ -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;

35
functions/supplier/ftp-supplier/README.adoc Normal file
View File

@@ -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<Flux<Message<?>>>`.
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<Flux<Message<?>>>`.
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.

72
functions/supplier/ftp-supplier/pom.xml Normal file
View File

@@ -0,0 +1,72 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<artifactId>ftp-supplier</artifactId>
<version>1.0.0-SNAPSHOT</version>
<name>ftp-supplier</name>
<description>ftp supplier</description>
<parent>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>spring-functions-parent</artifactId>
<version>1.0.0-SNAPSHOT</version>
<relativePath>../../spring-functions-parent</relativePath>
</parent>
<dependencies>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-ftp</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-integration</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>file-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>ftp-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>metadata-store-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-json</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>function-test-support</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
</dependencies>
</project>

144
functions/supplier/ftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierConfiguration.java Normal file
View File

@@ -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<FTPFile> ftpSessionFactory;
@Autowired
@Lazy
@Qualifier("ftpMessageSource")
private FtpInboundFileSynchronizingMessageSource ftpMessageSource;
public FtpSupplierConfiguration(FtpSupplierProperties ftpSupplierProperties,
FileConsumerProperties fileConsumerProperties,
ConcurrentMetadataStore metadataStore,
SessionFactory<FTPFile> ftpSessionFactory) {
this.ftpSupplierProperties = ftpSupplierProperties;
this.fileConsumerProperties = fileConsumerProperties;
this.metadataStore = metadataStore;
this.ftpSessionFactory = ftpSessionFactory;
}
@Bean
public MessageSource<File> 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<FTPFile> 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<Message<?>> ftpMessageFlux() {
return Mono.<Message<?>>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<Message<Object>> ftpReadingFlow() {
return FileUtils.enhanceFlowForReadingMode(IntegrationFlows
.from(IntegrationReactiveUtils.messageSourceToFlux(ftpMessageSource())), fileConsumerProperties)
.toReactivePublisher();
}
@Bean
public Supplier<Flux<Message<?>>> ftpSupplier() {
if (this.fileConsumerProperties.getMode() == FileReadingMode.ref) {
return this::ftpMessageFlux;
}
else {
return () -> Flux.from(ftpReadingFlow());
}
}
}

178
functions/supplier/ftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierProperties.java Normal file
View File

@@ -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;
}
}

139
functions/supplier/ftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierPropertiesTests.java Normal file
View File

@@ -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 {
}
}

99
functions/supplier/ftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierTests.java Normal file
View File

@@ -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<Flux<Message<?>>> ftpSupplier;
@Autowired
private ConcurrentMetadataStore metadataStore;
@Autowired
FtpSupplierProperties config;
@Autowired
SessionFactory<FTPFile> sessionFactory;
@Test
public void testSourceFileAsRef() {
final Flux<Message<?>> 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;
}
}
}