SpringCloud/spring-functions-catalog
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Implement SFTP Source

Browse Source 
Address review comments

* Code style clean up - generics warnings
* Fix `SftpSupplierApplicationTests` for compatibility with Windows
This commit is contained in:
David Turanski
2020-06-24 17:29:31 -04:00
committed by Artem Bilan
parent c5e78e2f4b
commit aa7152fc18
15 changed files with 1549 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
common/file-common/src/main/java/org/springframework/cloud/fn/common/file/remote/RemoteFileDeletingAdvice.java Normal file
View File

@@ -0,0 +1,55 @@
/*
* Copyright 2020-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.common.file.remote;
import org.springframework.integration.aop.MessageSourceMutator;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.file.remote.RemoteFileTemplate;
import org.springframework.messaging.Message;
/**
* A {@link MessageSourceMutator} that deletes a remote file on success.
*
* @author David Turanski
*
*/
public class RemoteFileDeletingAdvice implements MessageSourceMutator {
private final RemoteFileTemplate<?> template;
private final String remoteFileSeparator;
/**
* Construct an instance with the provided template and separator.
* @param template the template.
* @param remoteFileSeparator the separator.
*/
public RemoteFileDeletingAdvice(RemoteFileTemplate<?> template,
String remoteFileSeparator) {
this.template = template;
this.remoteFileSeparator = remoteFileSeparator;
}
@Override
public Message<?> afterReceive(Message<?> result, MessageSource<?> source) {
String remoteDir = (String) result.getHeaders().get(FileHeaders.REMOTE_DIRECTORY);
String remoteFile = (String) result.getHeaders().get(FileHeaders.REMOTE_FILE);
this.template.remove(remoteDir + this.remoteFileSeparator + remoteFile);
return result;
}
}

2
common/ftp-common/pom.xml
View File

@@ -4,7 +4,7 @@
<artifactId>ftp-common</artifactId>
<version>1.0.0-SNAPSHOT</version>
<name>ftp-common</name>
<description>file consumer</description>
<description>ftp common</description>
<parent>
<groupId>org.springframework.cloud.fn</groupId>

6
common/function-test-support/pom.xml
View File

@@ -18,6 +18,7 @@
<test-containers.version> 1.14.2</test-containers.version>
<apache-ftpserver.version>1.1.1</apache-ftpserver.version>
<geode-starter.version>1.2.7.RELEASE</geode-starter.version>
<awaitility.version>4.0.3</awaitility.version>
</properties>
<dependencies>
@@ -61,6 +62,11 @@
<groupId>org.springframework</groupId>
<artifactId>spring-websocket</artifactId>
</dependency>
<dependency>
<groupId>org.awaitility</groupId>
<artifactId>awaitility</artifactId>
<version>${awaitility.version}</version>
</dependency>
</dependencies>
</project>

6
common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/sftp/SftpTestSupport.java
View File

@@ -60,13 +60,13 @@ public class SftpTestSupport extends RemoteFileTestSupport {
server = SshServer.setUpDefaultServer();
server.setPasswordAuthenticator((username, password, session) ->
StringUtils.hasText(password) && !"badPassword".equals(password)); // fail if pub key validation failed
server.setPublickeyAuthenticator((username, key, session) -> key.equals(decodePublicKey("id_rsa_pp.pub.rename2")));
server.setPublickeyAuthenticator((username, key, session) -> key.equals(decodePublicKey("id_rsa_pp.pub")));
server.setPort(0);
server.setKeyPairProvider(new SimpleGeneratorHostKeyProvider(new File("hostkey.ser")));
server.setSubsystemFactories(Collections.singletonList(new SftpSubsystemFactory()));
server.setFileSystemFactory(new VirtualFileSystemFactory(remoteTemporaryFolder));
server.start();
System.setProperty("sftp.consumer.factory.port", String.valueOf(server.getPort()));
System.setProperty("sftp.factory.port", String.valueOf(server.getPort()));
System.setProperty("sftp.consumer.localDir",
localTemporaryFolder + File.separator + "localTarget");
}
@@ -78,7 +78,7 @@ public class SftpTestSupport extends RemoteFileTestSupport {
if (hostkey.exists()) {
hostkey.delete();
}
System.clearProperty("sftp.consumer.factory.port");
System.clearProperty("sftp.factory.port");
System.clearProperty("sftp.consumer.localDir");
}

30
common/function-test-support/src/main/resources/id_rsa_pp Normal file
View File

@@ -0,0 +1,30 @@
-----BEGIN RSA PRIVATE KEY-----
Proc-Type: 4,ENCRYPTED
DEK-Info: AES-128-CBC,26D1755B05980BA01B1E8D3B65EF98D3
J5fMQDf0HrwcnfYujq+q05GEOSEMVWMU0vr0hBtz2WvUeaFBBVAvUWbJo7PHTdDV
vEdSv+k8FazOkIZpeOW26pKNaLSuFfN+lgtAW4p4yqGQhbL8byh+Ka5uGaPH1xQj
o4exMFqbwFaHG10LJoPp5NnK+T64w8McJEihBPxv/qwtsz0YhhDVl/1eSwKsGa5y
Usfdv0QbjNDtpV7+sy7OunpaaKjb8iQ/PbFDsX0TSiy6jJflPwCYVUoh3jCEJze3
OUKwoQu7AiHmUrsnLtDuL49Q5hV+f9+IPJlzSqU5Fu8PlfCowH+e8WWLVQkh7Hht
iwOQs5UIWH+Nzbguu3Gbph5lqMtgkQwK6/PSFLQXuJmf9l6eo7E+Wh5mFWwoIENT
mKUlgVU0ymajPHcA7uOVXl5XX+Mt0DiMdGiB4N6RJ1OjWgqlRSJbVmGPrnZWmjdW
VltNux6JrO1bOgwtApCouDZhnP7/JDhM2PCa/F2+XCc6s8Hnmmt+gBw6IM4fb+YH
2gf1jx5Onp1yuBGe6tvbrfPAdXYU1R3kd2+Yf5mdsu2+xiC4wBKSfjleoHN4xJ/b
wopjQvGV5dj/VjdUt99lMHGWPr+p4NmdEBoriCgjYuRjRDvCiTV2qKZe2MU+0CVE
/jCA8iMRQnx3LMjnmJXP/96j2fyXHMowClTF43Cvdc5jh622WITOOFAIq9RfIXdS
47G6FAN54V+Qt0pXEgIOmvG+B2C24A041fo3jUPZxFRSYYuv9vG6QJby3kCsaqqt
ngYs2JSKx9CfOfGfyPbNt2/CO/bBsXgYzLR7REx5My1Mp9YsDmeIgbcd2V5hw/Me
rlSXrG4Eqs9gRDrBUvsydUOJFC1PlnIXH3VSBc6X9o4n2H6XECOsJQSeXLCaeMav
deKb0r1HvdbAYrdqw6mRM8Ok3fpSoD3mUsZQ8fp3luO3tHL1lddxHb7EsKzt/ubh
B9lEzDTcILlINlCl41X0OZKr/c+Ec8EdaSvITYJj0fvaZmDF7Wcs/dDWNft2XUaD
VcptOKbQVE1ufbrc2s1BdKOriAC6dSVKVDrAUQD/MlhT3p/YwbjSY2MqwRKztWRv
sA1Kqjg1IdUQzQizRKuHa322qnduLjHy0rb0ElrMpFe3B+OcIPs1E7Gvo4BVQ5jU
5GqHm83iaFIQaXmEsrVtCOBygVf00+WNRV3WFTOP9UEWFgtxsaHAU8UZhsDbKcZ5
l/w6kQdNElQuJA+1n1OxRZ5wIpfrRIMaxBQg2plUTVb9Tgz1qTZxHIEYPAAePfvX
tDn9SBiktj/8dhy1T0ko89yXCekpGxkU9rbmAd4Lrp6Wbc6+bKt4KzAqw319qq9U
Pslq7EKNM0Zq5pfwn1MRjzvmyHxz2sYHeij0CmuZY7xuOa4NzXt1vr6nSlIpX77W
ng/Rnd7qpyG2IWYihi6ztFHyj9h3FEcBHMk4JINLcdYOxXICz3KsRMfntrwQ4E6O
NFJ3fPpVNkk4GZcxy6idNkUBz1s8ixVWC8yi36byxE+TTRtcvqXQnJVgs63vMUlC
HVwaGau4YeUp4Nj2ZO44Srd/kQRy8yuCOPDdlJEiO6eDD4+XKedJQg1LuGeoXMVn
-----END RSA PRIVATE KEY-----

1
common/function-test-support/src/main/resources/id_rsa_pp.pub Normal file
View File

@@ -0,0 +1 @@
AAAAB3NzaC1yc2EAAAADAQABAAABAQC6MIzgyVi8G1+HRhFHPWRH+3w/8/uxtiuIfb4puVPjHI53Lvf5odzfhv0T6Z2/jSXmI3I6dpjbsgiptdCTX4kqUFLXxkuJR4LHatNtgO1w32aVIdAvfj7KtrL3SmP2XWqQGVcUWHEn2H1RHFHKdC6ArYFb1X8p5N/BHSQjuttaeVi9FsDxvC5euIbtDEEJmmvjjfWlI1m/6qCqMYxDWA9i9APU/rB0QwFNUQ6HuZ2QzEaU/hQMGmqgW5o1I/W8JR0bqis8wZQDLv1fwCkXpWG5BAuiJH+FJMxRAkfEMBpVwO7Sl0ufePVuSM2BMAAe+4a75sVp8ahbOId6y0GUTeJl

1
pom.xml
View File

@@ -91,6 +91,7 @@
<module>supplier/jms-supplier</module>
<module>supplier/mongodb-supplier</module>
<module>supplier/mqtt-supplier</module>
<module>supplier/sftp-supplier</module>
<module>supplier/tcp-supplier</module>
<module>supplier/time-supplier</module>
<module>supplier/rabbit-supplier</module>

86
supplier/sftp-supplier/README.adoc Normal file
View File

@@ -0,0 +1,86 @@
# SFTP Supplier
This module provides a SFTP supplier that can be reused and composed in other applications.
The `Supplier` uses various `Sftp` inbound adapters from Spring Integration to support a range of modes to consume data from an SFTP server.
These include:
* Synchronize remote files to a local directory and supplying the File contents, the File reference, or a message per line.
* Stream remote file contents as a byte array directly without copying to a local directory.
* List the remote file names only.
Messages emitted by the supplier are provided as a byte array by default. However, this can be customized using the `--mode` option:
- *ref* Provides a `java.io.File` reference
- *lines* Will split files line-by-line and emit a new message for each line
- *contents* The default. Provides the contents of a file as a byte array
NOTE: When using `--mode=lines`, you can provide an additional option `--withMarkers=true`.
If set to `true`, the underlying `FileSplitter` will emit additional _start-of-file_ and _end-of-file_ marker messages before and after the actual data.
The payload of these 2 additional marker messages is of type `FileSplitter.FileMarker`.
The option `withMarkers` defaults to `false` if not explicitly set.
When configuring the `sftp.factory.known-hosts-expression` option, the root object of the evaluation is the application context, an example might be `sftp.factory.known-hosts-expression = @systemProperties['user.home'] + '/.ssh/known_hosts'`.
## Idempotency
By default the supplier uses a https://docs.spring.io/spring-integration/api/org/springframework/integration/metadata/SimpleMetadataStore.html[SimpleMetadataStore], storing the last modified time to track files that have already been processed in memory.
If an application using this supplier is restarted, any existing files will be reprocessed. You can inject on of the persistent https://docs.spring.io/spring-integration/reference/html/meta-data-store.html[MetadataStore implementations] provided by Spring Integration, or your own of course, to maintain this state permanently.
See also link:../../common/metadata-store-common/README.adoc[`MetadataStore`] options for possible shared persistent store configuration for the `SftpPersistentAcceptOnceFileListFilter` used in the SFTP Source.
## Multiple SFTP Servers
This source supports consuming from multiple SFTP servers.
This requires configuring an SFTP Session Factory for each server.
The labels `one` and `two` shown below can be replaced by any names you want.
The following configuration will rotate between two SFTP servers (this can also be used for multiple directories on the same server), consuming files in a round-robin fashion:
```
sftp.supplier.factories.one.host=host1
sftp.supplier.factories.one.port=1234,
sftp.supplier.factories.one.username = user1,
sftp.supplier.factories.one.password = pass1,
...
sftp.supplier.factories.two.host=host2,
sftp.supplier.factories.two.port=2345,
sftp.supplier.factories.two.username = user2,
sftp.supplier.factories.two.password = pass2,
sftp.supplier.directories=one.sftpSource,two.sftpSecondSource,
sftp.supplier.max-fetch=1,
sftp.supplier.fair=true
```
---
`SFtpSupplier` is implemented as a `java.util.function.Supplier`.
This supplier gives you a reactive stream of objects from the provided directory(ies) as the supplier has a signature of `Supplier<Flux<Message<?>>>`.
Users have to subscribe to this `Flux` and receive the data.
## Beans for injection
You can import the `SftpSupplierConfiguration` in the application and then inject the following bean.
`sftpSupplier`
You need to inject this as `Supplier<Flux<Message<?>>>`.
You can use `sftpSupplier` as a qualifier when injecting.
Once injected, you can use the `get` method of the `Supplier` to invoke it and then subscribe to the returned `Flux`.
## Configuration Options
All configuration properties are prefixed with `ftp.supplier`.
There are also properties that need to be used with the prefix `file.consumer`.
For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierProperties.java[SftpSupplierProperties].
Also see link:src/main/java/org/springframework/cloud/fn/supplier/file/FileConsumerProperties.java[FileConsumerProperties].
## Examples
See this link:src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java[test suite] for the various ways, this supplier is used.
## Other usage
See this link:../../../applications/source/sftp-source/README.adoc[README] where this supplier is used to create a Spring Cloud Stream application that provides an Sftp Source.

76
supplier/sftp-supplier/pom.xml Normal file
View File

@@ -0,0 +1,76 @@
<?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>sftp-supplier</artifactId>
<version>1.0.0-SNAPSHOT</version>
<name>sftp-supplier</name>
<description>sftp 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-sftp</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-starter-logging</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>

364
supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java Normal file
View File

@@ -0,0 +1,364 @@
/*
* Copyright 2018-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.sftp;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.reactivestreams.Publisher;
import org.springframework.aop.framework.ProxyFactoryBean;
import org.springframework.aop.support.NameMatchMethodPointcutAdvisor;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.fn.common.file.FileConsumerProperties;
import org.springframework.cloud.fn.common.file.FileUtils;
import org.springframework.cloud.fn.common.file.remote.RemoteFileDeletingAdvice;
import org.springframework.context.Lifecycle;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.context.annotation.Primary;
import org.springframework.integration.aop.ReceiveMessageAdvice;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.file.filters.ChainFileListFilter;
import org.springframework.integration.file.filters.FileListFilter;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.handler.MessageProcessor;
import org.springframework.integration.metadata.ConcurrentMetadataStore;
import org.springframework.integration.sftp.dsl.Sftp;
import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter;
import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter;
import org.springframework.integration.sftp.session.SftpRemoteFileTemplate;
import org.springframework.integration.util.IntegrationReactiveUtils;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.MimeTypeUtils;
import org.springframework.util.StringUtils;
import com.jcraft.jsch.ChannelSftp.LsEntry;
import reactor.core.publisher.Flux;
/**
* @author Gary Russell
* @author Artem Bilan
* @author Chris Schaefer
* @author Christian Tzolov
* @author David Turanski
*/
@Configuration
@EnableConfigurationProperties({SftpSupplierProperties.class, FileConsumerProperties.class})
@Import({SftpSupplierFactoryConfiguration.class})
@SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
public class SftpSupplierConfiguration {
private static final String METADATA_STORE_PREFIX = "sftpSource/";
@Bean
public Supplier<Flux<? extends Message<?>>> sftpSupplier(MessageSource<?> sftpMessageSource,
@Nullable Publisher<Message<Object>> sftpReadingFlow,
SftpSupplierProperties sftpSupplierProperties) {
Flux<? extends Message<?>> flux = sftpReadingFlow == null
? sftpMessageFlux(sftpMessageSource, sftpSupplierProperties)
: Flux.from(sftpReadingFlow);
return () -> flux.doOnSubscribe(s -> {
if (sftpMessageSource instanceof Lifecycle) {
((Lifecycle) sftpMessageSource).start();
}
});
}
@Bean
@Primary
public MessageSource<?> sftpMessageSource(
MessageSource<?> messageSource,
BeanFactory beanFactory,
@Nullable List<ReceiveMessageAdvice> receiveMessageAdvice) {
if (CollectionUtils.isEmpty(receiveMessageAdvice)) {
return messageSource;
}
ProxyFactoryBean proxyFactoryBean = new ProxyFactoryBean();
proxyFactoryBean.setTarget(messageSource);
proxyFactoryBean.setBeanFactory(beanFactory);
receiveMessageAdvice.stream().map(advice -> {
NameMatchMethodPointcutAdvisor advisor = new NameMatchMethodPointcutAdvisor(advice);
advisor.addMethodName("receive");
return advisor;
}).forEach(proxyFactoryBean::addAdvisor);
return (MessageSource<?>) proxyFactoryBean.getObject();
}
/*
* Configure the standard filters for SFTP inbound adapters.
*/
@Bean
public ChainFileListFilter<LsEntry> chainFilter(SftpSupplierProperties sftpSupplierProperties,
ConcurrentMetadataStore metadataStore) {
ChainFileListFilter<LsEntry> chainFilter = new ChainFileListFilter<>();
if (StringUtils.hasText(sftpSupplierProperties.getFilenamePattern())) {
chainFilter
.addFilter(new SftpRegexPatternFileListFilter(sftpSupplierProperties.getFilenamePattern()));
}
else if (sftpSupplierProperties.getFilenameRegex() != null) {
chainFilter
.addFilter(new SftpRegexPatternFileListFilter(sftpSupplierProperties.getFilenameRegex()));
}
// TODO: Temporary work-around for
// https://github.com/spring-projects/spring-integration/issues/3315.
chainFilter.addFilter(Arrays::asList);
chainFilter.addFilter(new SftpPersistentAcceptOnceFileListFilter(metadataStore, METADATA_STORE_PREFIX));
return chainFilter;
}
/*
* Create a Flux from a MessageSource that will be used by the supplier.
*/
private Flux<? extends Message<?>> sftpMessageFlux(MessageSource<?> sftpMessageSource,
SftpSupplierProperties sftpSupplierProperties) {
return IntegrationReactiveUtils.messageSourceToFlux(sftpMessageSource)
.subscriberContext(context -> context.put(IntegrationReactiveUtils.DELAY_WHEN_EMPTY_KEY,
sftpSupplierProperties.getDelayWhenEmpty()));
}
private static String remoteDirectory(SftpSupplierProperties sftpSupplierProperties) {
return sftpSupplierProperties.isMultiSource()
? SftpSupplierProperties.keyDirectories(sftpSupplierProperties).get(0).getDirectory()
: sftpSupplierProperties.getRemoteDir();
}
@Configuration
@ConditionalOnProperty(prefix = "sftp.supplier", name = "stream")
static class StreamingConfiguration {
@Bean
public SftpRemoteFileTemplate sftpTemplate(SftpSupplierFactoryConfiguration.DelegatingFactoryWrapper wrapper) {
return new SftpRemoteFileTemplate(wrapper.getFactory());
}
/**
* Streaming {@link MessageSource} that provides an InputStream for each remote file. It
* does not synchronize files to a local directory.
* @return a {@link MessageSource}.
*/
@Bean
public MessageSource<?> targetMessageSource(SftpRemoteFileTemplate sftpTemplate,
SftpSupplierProperties sftpSupplierProperties,
FileListFilter<LsEntry> fileListFilter) {
return Sftp.inboundStreamingAdapter(sftpTemplate)
.remoteDirectory(remoteDirectory(sftpSupplierProperties))
.remoteFileSeparator(sftpSupplierProperties.getRemoteFileSeparator())
.filter(fileListFilter)
.maxFetchSize(sftpSupplierProperties.getMaxFetch()).get();
}
@Bean
public Publisher<Message<Object>> sftpReadingFlow(
MessageSource<?> sftpMessageSource,
SftpSupplierProperties sftpSupplierProperties,
FileConsumerProperties fileConsumerProperties) {
return FileUtils.enhanceStreamFlowForReadingMode(IntegrationFlows
.from(IntegrationReactiveUtils.messageSourceToFlux(sftpMessageSource)
.subscriberContext(
context -> (context.put(IntegrationReactiveUtils.DELAY_WHEN_EMPTY_KEY,
sftpSupplierProperties.getDelayWhenEmpty())))),
fileConsumerProperties)
.toReactivePublisher();
}
@Bean
@ConditionalOnProperty(prefix = "sftp.supplier", value = "delete-remote-files")
public RemoteFileDeletingAdvice remoteFileDeletingAdvice(SftpRemoteFileTemplate sftpTemplate,
SftpSupplierProperties sftpSupplierProperties) {
return new RemoteFileDeletingAdvice(sftpTemplate, sftpSupplierProperties.getRemoteFileSeparator());
}
}
@Configuration
@ConditionalOnExpression("environment['sftp.supplier.stream']!='true'")
static class NonStreamingConfiguration {
/**
* Enrich the flow to provide some standard headers, depending on
* {@link FileConsumerProperties}, when consuming file contents.
* @param sftpMessageSource the {@link MessageSource}.
* @param fileConsumerProperties the {@code FileConsumerProperties}.
* @return a {@code Publisher<Message>}.
*/
@Bean
@ConditionalOnExpression("environment['file.consumer.mode']!='ref' && environment['sftp.supplier.list-only']!='true'")
public Publisher<Message<Object>> sftpReadingFlow(
MessageSource<?> sftpMessageSource,
FileConsumerProperties fileConsumerProperties) {
return FileUtils.enhanceFlowForReadingMode(IntegrationFlows
.from(IntegrationReactiveUtils.messageSourceToFlux(sftpMessageSource)),
fileConsumerProperties)
.toReactivePublisher();
}
/**
* A {@link MessageSource} that synchronizes files to a local directory.
* @return the {code MessageSource}.
*/
@ConditionalOnExpression("environment['sftp.supplier.list-only'] != 'true'")
@Bean
public MessageSource targetMessageSource(SftpSupplierProperties sftpSupplierProperties,
SftpSupplierFactoryConfiguration.DelegatingFactoryWrapper delegatingFactoryWrapper,
FileListFilter<LsEntry> fileListFilter) {
return Sftp
.inboundAdapter(delegatingFactoryWrapper.getFactory())
.preserveTimestamp(sftpSupplierProperties.isPreserveTimestamp())
.autoCreateLocalDirectory(sftpSupplierProperties.isAutoCreateLocalDir())
.deleteRemoteFiles(sftpSupplierProperties.isDeleteRemoteFiles())
.localDirectory(sftpSupplierProperties.getLocalDir())
.remoteDirectory(remoteDirectory(sftpSupplierProperties))
.remoteFileSeparator(sftpSupplierProperties.getRemoteFileSeparator())
.temporaryFileSuffix(sftpSupplierProperties.getTmpFileSuffix())
.metadataStorePrefix(METADATA_STORE_PREFIX)
.maxFetchSize(sftpSupplierProperties.getMaxFetch())
.filter(fileListFilter).get();
}
}
/*
* List only configuration
*/
@Configuration
@ConditionalOnProperty(prefix = "sftp.supplier", name = "list-only")
static class ListingOnlyConfiguration {
@Bean
PollableChannel listingChannel() {
return new QueueChannel();
}
@Bean
@SuppressWarnings("unchecked")
public MessageSource<?> targetMessageSource(PollableChannel listingChannel,
SftpListingMessageProducer sftpListingMessageProducer) {
return () -> {
sftpListingMessageProducer.listNames();
return (Message<Object>) listingChannel.receive();
};
}
@Bean
public SftpListingMessageProducer sftpListingMessageProducer(SftpSupplierProperties sftpSupplierProperties,
SftpSupplierFactoryConfiguration.DelegatingFactoryWrapper delegatingFactoryWrapper) {
return new SftpListingMessageProducer(delegatingFactoryWrapper.getFactory(),
remoteDirectory(sftpSupplierProperties),
sftpSupplierProperties.getRemoteFileSeparator());
}
@Bean
public IntegrationFlow listingFlow(MessageProducerSupport messageProducerSupport,
MessageChannel listingChannel, MessageProcessor<?> metadataWriter) {
return IntegrationFlows.from(messageProducerSupport)
.split()
.transform(metadataWriter)
.channel(listingChannel)
.get();
}
@Bean
public MessageProcessor<?> metadataWriter(ConcurrentMetadataStore metadataStore) {
return message -> {
MessageHeaders messageHeaders = message.getHeaders();
Assert.notNull(messageHeaders, "Cannot transform message with null headers");
Assert.isTrue(messageHeaders.containsKey(FileHeaders.REMOTE_DIRECTORY),
"Remote directory header not found");
Assert.hasText((String) message.getPayload(), "Payload must not be empty.");
metadataStore.putIfAbsent(
message.getHeaders().get(FileHeaders.REMOTE_DIRECTORY).toString() + message.getPayload(),
String.valueOf(message.getHeaders().getTimestamp()));
return message;
};
}
static class SftpListingMessageProducer extends MessageProducerSupport {
private final String remoteDirectory;
private final SessionFactory<?> sessionFactory;
private final String remoteFileSeparator;
SftpListingMessageProducer(SessionFactory<?> sessionFactory, String remoteDirectory,
String remoteFileSeparator) {
this.sessionFactory = sessionFactory;
this.remoteDirectory = remoteDirectory;
this.remoteFileSeparator = remoteFileSeparator;
}
public void listNames() {
String[] names = {};
try {
names = Stream.of(this.sessionFactory.getSession().listNames(this.remoteDirectory))
.map(name -> String.join(this.remoteFileSeparator, this.remoteDirectory, name))
.collect(Collectors.toList()).toArray(names);
}
catch (IOException e) {
throw new MessagingException(e.getMessage(), e);
}
sendMessage(MessageBuilder.withPayload(names)
.setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN)
.setHeader(FileHeaders.REMOTE_DIRECTORY, this.remoteDirectory + this.remoteFileSeparator)
.build());
}
}
}
}

127
supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierFactoryConfiguration.java Normal file
View File

@@ -0,0 +1,127 @@
/*
* Copyright 2018-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.sftp;
import java.util.HashMap;
import java.util.Map;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.core.io.FileSystemResource;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.file.remote.aop.StandardRotationPolicy;
import org.springframework.integration.file.remote.session.DelegatingSessionFactory;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
import org.springframework.lang.Nullable;
import com.jcraft.jsch.ChannelSftp.LsEntry;
/**
* Session factory configuration.
*
* @author Gary Russell
* @author Artem Bilan
* @author David Turanski
*
*/
public class SftpSupplierFactoryConfiguration {
@Bean
@ConditionalOnMissingBean
public SessionFactory<LsEntry> sftpSessionFactory(SftpSupplierProperties properties, BeanFactory beanFactory) {
return buildFactory(beanFactory, properties.getFactory());
}
@Bean
public DelegatingFactoryWrapper delegatingFactoryWrapper(SftpSupplierProperties properties,
SessionFactory<LsEntry> defaultFactory, BeanFactory beanFactory) {
return new DelegatingFactoryWrapper(properties, defaultFactory, beanFactory);
}
@Bean
StandardRotationPolicy rotationPolicy(SftpSupplierProperties properties, DelegatingFactoryWrapper factory) {
return properties.isMultiSource()
? new StandardRotationPolicy(factory.getFactory(),
SftpSupplierProperties.keyDirectories(properties), properties.isFair())
: null;
}
@Bean
public SftpSupplierRotator rotatingAdvice(SftpSupplierProperties properties,
@Nullable StandardRotationPolicy rotationPolicy) {
return properties.isMultiSource()
? new SftpSupplierRotator(properties, rotationPolicy)
: null;
}
static SessionFactory<LsEntry> buildFactory(BeanFactory beanFactory, SftpSupplierProperties.Factory factory) {
DefaultSftpSessionFactory sftpSessionFactory = new DefaultSftpSessionFactory(true);
sftpSessionFactory.setHost(factory.getHost());
sftpSessionFactory.setPort(factory.getPort());
sftpSessionFactory.setUser(factory.getUsername());
sftpSessionFactory.setPassword(factory.getPassword());
sftpSessionFactory.setPrivateKey(factory.getPrivateKey());
sftpSessionFactory.setPrivateKeyPassphrase(factory.getPassPhrase());
sftpSessionFactory.setAllowUnknownKeys(factory.isAllowUnknownKeys());
if (factory.getKnownHostsExpression() != null) {
String path = factory.getKnownHostsExpression()
.getValue(IntegrationContextUtils.getEvaluationContext(beanFactory), String.class);
sftpSessionFactory.setKnownHostsResource(new FileSystemResource(path));
}
return sftpSessionFactory;
}
public final static class DelegatingFactoryWrapper implements DisposableBean {
private final DelegatingSessionFactory<LsEntry> delegatingSessionFactory;
private final Map<Object, SessionFactory<LsEntry>> factories = new HashMap<>();
DelegatingFactoryWrapper(SftpSupplierProperties properties, SessionFactory<LsEntry> defaultFactory,
BeanFactory beanFactory) {
properties.getFactories().forEach((key, factory) -> {
this.factories.put(key, SftpSupplierFactoryConfiguration.buildFactory(beanFactory, factory));
});
this.delegatingSessionFactory = new DelegatingSessionFactory<>(this.factories, defaultFactory);
}
public DelegatingSessionFactory<LsEntry> getFactory() {
return this.delegatingSessionFactory;
}
@Override
public void destroy() {
this.factories.values().forEach(f -> {
if (f instanceof DisposableBean) {
try {
((DisposableBean) f).destroy();
}
catch (Exception e) {
// empty
}
}
});
}
}
}

406
supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierProperties.java Normal file
View File

@@ -0,0 +1,406 @@
/*
* Copyright 2018-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.sftp;
import java.io.File;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;
import javax.validation.constraints.AssertTrue;
import javax.validation.constraints.NotBlank;
import javax.validation.constraints.NotNull;
import org.hibernate.validator.constraints.Range;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.core.io.Resource;
import org.springframework.expression.Expression;
import org.springframework.integration.file.remote.aop.RotationPolicy;
import org.springframework.util.Assert;
import org.springframework.validation.annotation.Validated;
/**
* @author Gary Russell
* @author Artem Bilan
* @author Chris Schaefer
* @author David Turanski
*/
@ConfigurationProperties("sftp.supplier")
@Validated
public class SftpSupplierProperties {
/**
* Session factory properties.
*/
private final Factory factory = new Factory();
/**
* The remote FTP directory.
*/
private String remoteDir = "/";
/**
* The suffix to use while the transfer is in progress.
*/
private String tmpFileSuffix = ".tmp";
/**
* The remote file separator.
*/
private String remoteFileSeparator = "/";
/**
* Set to true to delete remote files after successful transfer.
*/
private boolean deleteRemoteFiles = false;
/**
* The local directory to use for file transfers.
*/
private File localDir = new File(System.getProperty("java.io.tmpdir"), "sftp-supplier");
/**
* Set to true to create the local directory if it does not exist.
*/
private boolean autoCreateLocalDir = true;
/**
* A filter pattern to match the names of files to transfer.
*/
private String filenamePattern;
/**
* A filter regex pattern to match the names of files to transfer.
*/
private Pattern filenameRegex;
/**
* Set to true to preserve the original timestamp.
*/
private boolean preserveTimestamp = true;
/**
* Set to true to stream the file rather than copy to a local directory.
*/
private boolean stream = false;
/**
* Set to true to return file metadata without the entire payload.
*/
private boolean listOnly = false;
/**
* Duration of delay when no new files are detected.
*/
private Duration delayWhenEmpty = Duration.ofSeconds(1);
/**
* The maximum number of remote files to fetch per poll; default unlimited. Does not apply
* when listing files or building task launch requests.
*/
private int maxFetch = Integer.MIN_VALUE;
/**
* True for fair rotation of multiple servers/directories. This is false by default so if
* a source has more than one entry, these will be received before the other sources are
* visited.
*/
private boolean fair;
/**
* A map of factory names to factories.
*/
private Map<String, Factory> factories = Collections.emptyMap();
/**
* A list of factory "name.directory" pairs.
*/
private String[] directories;
@NotBlank
public String getRemoteDir() {
return remoteDir;
}
public void setRemoteDir(String remoteDir) {
this.remoteDir = remoteDir;
}
@NotBlank
public String getTmpFileSuffix() {
return tmpFileSuffix;
}
public void setTmpFileSuffix(String tmpFileSuffix) {
this.tmpFileSuffix = tmpFileSuffix;
}
@NotBlank
public String getRemoteFileSeparator() {
return remoteFileSeparator;
}
public void setRemoteFileSeparator(String remoteFileSeparator) {
this.remoteFileSeparator = remoteFileSeparator;
}
public boolean isAutoCreateLocalDir() {
return autoCreateLocalDir;
}
public void setAutoCreateLocalDir(boolean autoCreateLocalDir) {
this.autoCreateLocalDir = autoCreateLocalDir;
}
public boolean isDeleteRemoteFiles() {
return deleteRemoteFiles;
}
public void setDeleteRemoteFiles(boolean deleteRemoteFiles) {
this.deleteRemoteFiles = deleteRemoteFiles;
}
@NotNull
public File getLocalDir() {
return localDir;
}
public final void setLocalDir(File localDir) {
this.localDir = localDir;
}
public String getFilenamePattern() {
return filenamePattern;
}
public void setFilenamePattern(String filenamePattern) {
this.filenamePattern = filenamePattern;
}
public Pattern getFilenameRegex() {
return filenameRegex;
}
public void setFilenameRegex(Pattern filenameRegex) {
this.filenameRegex = filenameRegex;
}
public boolean isPreserveTimestamp() {
return preserveTimestamp;
}
public void setPreserveTimestamp(boolean preserveTimestamp) {
this.preserveTimestamp = preserveTimestamp;
}
@AssertTrue(message = "filenamePattern and filenameRegex are mutually exclusive")
public boolean isExclusivePatterns() {
return !(this.filenamePattern != null && this.filenameRegex != null);
}
public boolean isListOnly() {
return listOnly;
}
public void setListOnly(boolean listOnly) {
this.listOnly = listOnly;
}
public boolean isMultiSource() {
return this.directories != null && this.directories.length > 0;
}
public int getMaxFetch() {
return maxFetch;
}
public void setMaxFetch(int maxFetch) {
this.maxFetch = maxFetch;
}
public boolean isFair() {
return this.fair;
}
public void setFair(boolean fair) {
this.fair = fair;
}
public Map<String, Factory> getFactories() {
return this.factories;
}
public void setFactories(Map<String, Factory> factories) {
this.factories = factories;
}
public String[] getDirectories() {
return this.directories;
}
public void setDirectories(String[] directories) {
this.directories = directories;
}
public boolean isStream() {
return stream;
}
public void setStream(boolean stream) {
this.stream = stream;
}
public Factory getFactory() {
return factory;
}
public Duration getDelayWhenEmpty() {
return delayWhenEmpty;
}
public void setDelayWhenEmpty(Duration delayWhenEmpty) {
this.delayWhenEmpty = delayWhenEmpty;
}
static List<RotationPolicy.KeyDirectory> keyDirectories(SftpSupplierProperties properties) {
List<RotationPolicy.KeyDirectory> keyDirs = new ArrayList<>();
Assert.isTrue(properties.getDirectories().length > 0, "At least one key.directory required");
for (String keyDir : properties.getDirectories()) {
String[] split = keyDir.split("\\.");
Assert.isTrue(split.length == 2, () -> "key/directory can only have one '.': " + keyDir);
keyDirs.add(new RotationPolicy.KeyDirectory(split[0], split[1]));
}
return keyDirs;
}
public static class Factory {
/**
* The host name of the server.
*/
private String host = "localhost";
/**
* The username to use to connect to the server.
*/
private String username;
/**
* The password to use to connect to the server.
*/
private String password;
/**
* The port of the server.
*/
private int port = 22;
/**
* Resource location of user's private key.
*/
private Resource privateKey;
/**
* Passphrase for user's private key.
*/
private String passPhrase = "";
/**
* True to allow an unknown or changed key.
*/
private boolean allowUnknownKeys = false;
/**
* A SpEL expression resolving to the location of the known hosts file.
*/
private Expression knownHostsExpression = null;
@NotBlank
public String getHost() {
return this.host;
}
public void setHost(String host) {
this.host = host;
}
@NotBlank
public String getUsername() {
return this.username;
}
public void setUsername(String username) {
this.username = username;
}
public String getPassword() {
return this.password;
}
public void setPassword(String password) {
this.password = password;
}
@Range(min = 0, max = 65535)
public int getPort() {
return this.port;
}
public void setPort(int port) {
this.port = port;
}
public Resource getPrivateKey() {
return this.privateKey;
}
public void setPrivateKey(Resource privateKey) {
this.privateKey = privateKey;
}
public String getPassPhrase() {
return this.passPhrase;
}
public void setPassPhrase(String passPhrase) {
this.passPhrase = passPhrase;
}
public boolean isAllowUnknownKeys() {
return this.allowUnknownKeys;
}
public void setAllowUnknownKeys(boolean allowUnknownKeys) {
this.allowUnknownKeys = allowUnknownKeys;
}
public Expression getKnownHostsExpression() {
return this.knownHostsExpression;
}
public void setKnownHostsExpression(Expression knownHosts) {
this.knownHostsExpression = knownHosts;
}
}
}

63
supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierRotator.java Normal file
View File

@@ -0,0 +1,63 @@
/*
* Copyright 2018-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.sftp;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.file.remote.aop.RotatingServerAdvice;
import org.springframework.integration.file.remote.aop.StandardRotationPolicy;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
/**
* An {@link RotatingServerAdvice} for listing files on multiple directories/servers.
*
* @author Gary Russell
* @author David Turanski
* @since 2.0
*/
public class SftpSupplierRotator extends RotatingServerAdvice {
private static String SFTP_SELECTED_SERVER_PROPERTY_KEY = "sftp_selectedServer";
private final SftpSupplierProperties properties;
private final StandardRotationPolicy rotationPolicy;
public SftpSupplierRotator(SftpSupplierProperties properties, StandardRotationPolicy rotationPolicy) {
super(rotationPolicy);
this.properties = properties;
this.rotationPolicy = rotationPolicy;
}
public String getCurrentKey() {
return this.rotationPolicy.getCurrent().getKey().toString();
}
public String getCurrentDirectory() {
return this.rotationPolicy.getCurrent().getDirectory();
}
@Override
public Message<?> afterReceive(Message<?> result, MessageSource<?> source) {
if (result != null) {
result = MessageBuilder.fromMessage(result)
.setHeader(SFTP_SELECTED_SERVER_PROPERTY_KEY, this.getCurrentKey()).build();
}
this.rotationPolicy.afterReceive(result != null, source);
return result;
}
}

314
supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java Normal file
View File

@@ -0,0 +1,314 @@
/*
* Copyright 2020-2020 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.cloud.fn.supplier.sftp;
import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.time.Duration;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.runner.ApplicationContextRunner;
import org.springframework.cloud.fn.test.support.sftp.SftpTestSupport;
import org.springframework.http.MediaType;
import org.springframework.integration.file.splitter.FileSplitter;
import org.springframework.integration.json.JsonPathUtils;
import org.springframework.integration.metadata.MetadataStore;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.awaitility.Awaitility.await;
public class SftpSupplierApplicationTests extends SftpTestSupport {
ApplicationContextRunner defaultApplicationContextRunner;
@BeforeEach
void setUpDefaultProperties() {
defaultApplicationContextRunner = new ApplicationContextRunner()
.withUserConfiguration(TestApp.class)
.withPropertyValues(
"sftp.supplier.factory.host=localhost",
"sftp.supplier.factory.port=${sftp.factory.port}",
"sftp.supplier.factory.username=user",
"sftp.supplier.factory.password=pass",
"sftp.supplier.factory.cache-sessions=true",
"sftp.supplier.factory.allowUnknownKeys=true",
"sftp.supplier.localDir=" + this.targetLocalDirectory.getAbsolutePath(),
"sftp.supplier.remoteDir=sftpSource");
}
@Test
void supplierForListOnly() {
defaultApplicationContextRunner
.withPropertyValues("sftp.supplier.listOnly=true")
.run(context -> {
Supplier<Flux<Message<String>>> sftpSupplier = context.getBean("sftpSupplier",
Supplier.class);
SftpSupplierProperties properties = context.getBean(SftpSupplierProperties.class);
HashSet<String> fileNames = new HashSet<>();
fileNames.add(String.join(properties.getRemoteFileSeparator(), properties.getRemoteDir(),
"sftpSource1.txt"));
fileNames.add(String.join(properties.getRemoteFileSeparator(), properties.getRemoteDir(),
"sftpSource2.txt"));
final AtomicReference<Set<String>> expectedFileNames = new AtomicReference<>(fileNames);
StepVerifier.create(sftpSupplier.get())
.assertNext(message -> {
assertThat(expectedFileNames.get()).contains(message.getPayload());
expectedFileNames.get().remove(message.getPayload());
assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE))
.isEqualTo(MediaType.TEXT_PLAIN);
})
.assertNext(message -> {
assertThat(expectedFileNames.get()).contains(message.getPayload());
assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE))
.isEqualTo(MediaType.TEXT_PLAIN);
})
.thenCancel()
.verify(Duration.ofSeconds(10));
});
}
@Test
void supplierForFileRef() {
defaultApplicationContextRunner
.withPropertyValues(
"sftp.supplier.localDir=" + getTargetLocalDirectory().getAbsolutePath(),
"file.consumer.mode=ref")
.run(context -> {
Supplier<Flux<Message<File>>> sftpSupplier = context.getBean("sftpSupplier",
Supplier.class);
SftpSupplierProperties properties = context.getBean(SftpSupplierProperties.class);
MetadataStore metadataStore = context.getBean(MetadataStore.class);
HashSet<String> fileNames = new HashSet<>();
fileNames.add(properties.getLocalDir() + File.separator + "sftpSource1.txt");
fileNames.add(properties.getLocalDir() + File.separator + "sftpSource2.txt");
final AtomicReference<Set<String>> expectedFileNames = new AtomicReference<>(fileNames);
StepVerifier.create(sftpSupplier.get())
.assertNext(message -> {
File file = (File) message.getPayload();
assertThat(expectedFileNames.get()).contains(file.getAbsolutePath());
expectedFileNames.get().remove(file.getAbsolutePath());
})
.expectNextMatches(
message -> expectedFileNames.get().contains(message.getPayload().getAbsolutePath()))
.thenCancel()
.verify(Duration.ofSeconds(10));
assertThat(metadataStore.get("sftpSource/sftpSource1.txt")).isNotNull();
assertThat(metadataStore.get("sftpSource/sftpSource2.txt")).isNotNull();
assertThat(Files.exists(Paths.get(getTargetLocalDirectory().getAbsolutePath(), "sftpSource1.txt")))
.isTrue();
assertThat(Files.exists(Paths.get(getTargetLocalDirectory().getAbsolutePath(), "sftpSource2.txt")))
.isTrue();
});
}
@Test
void deleteRemoteFiles() {
defaultApplicationContextRunner
.withPropertyValues(
"sftp.supplier.stream=true",
"sftp.supplier.delete-remote-files=true")
.run(context -> {
Supplier<Flux<Message<byte[]>>> sftpSupplier = context.getBean("sftpSupplier",
Supplier.class);
StepVerifier.create(sftpSupplier.get())
.expectNextMatches(message -> message.getPayload().length > 0)
.expectNextMatches(message -> message.getPayload().length > 0)
.thenCancel()
.verify(Duration.ofSeconds(10));
await().atMost(Duration.ofSeconds(10))
.until(() -> getSourceRemoteDirectory().list().length == 0);
});
}
@Test
public void streamSourceFilesInLineMode() {
defaultApplicationContextRunner
.withPropertyValues(
"sftp.supplier.stream=true",
"sftp.supplier.factory.private-key = classpath:id_rsa_pp",
"sftp.supplier.factory.passphrase = secret",
"sftp.supplier.factory.password = badPassword", // ensure public key was used
"file.consumer.mode=lines",
"file.consumer.with-markers=true",
"file.consumer.markers-json=true")
.run(context -> {
Supplier<Flux<Message<String>>> sftpSupplier = context.getBean("sftpSupplier",
Supplier.class);
StepVerifier.create(sftpSupplier.get())
.assertNext(message -> {
final Object evaluate;
try {
evaluate = JsonPathUtils.evaluate(message.getPayload(), "$.mark");
assertThat(evaluate).isEqualTo(FileSplitter.FileMarker.Mark.START.name());
}
catch (IOException e) {
fail(e.getMessage());
}
})
.expectNextMatches(message -> message.getPayload().startsWith("source"))
.assertNext(message -> {
final Object evaluate;
try {
evaluate = JsonPathUtils.evaluate(message.getPayload(), "$.mark");
assertThat(evaluate).isEqualTo(FileSplitter.FileMarker.Mark.END.name());
}
catch (IOException e) {
fail(e.getMessage());
}
})
.thenCancel()
.verify(Duration.ofSeconds(10));
});
}
@Test
void supplierWithMultiSourceAndStreamContentsSource3ComesSecond() throws Exception {
Path newSource = createNewRemoteSource(
Paths.get(remoteTemporaryFolder.toString(), "sftpSecondSource", "doesNotMatter.txt"),
"source3");
new ApplicationContextRunner()
.withUserConfiguration(TestApp.class)
.withPropertyValues(
"sftp.supplier.stream=true",
"sftp.supplier.factories.one.host=localhost",
"sftp.supplier.factories.one.port=${sftp.factory.port}",
"sftp.supplier.factories.one.username=user",
"sftp.supplier.factories.one.password=pass",
"sftp.supplier.factories.one.cache-sessions=true",
"sftp.supplier.factories.one.allowUnknownKeys=true",
"sftp.supplier.factories.two.host=localhost",
"sftp.supplier.factories.two.port=${sftp.factory.port}",
"sftp.supplier.factories.two.username = user",
"sftp.supplier.factories.two.password = pass",
"sftp.supplier.factories.two.cache-sessions = true",
"sftp.supplier.factories.two.allowUnknownKeys = true",
"sftp.supplier.directories=one.sftpSource,two.sftpSecondSource",
"sftp.supplier.max-fetch=1",
"sftp.supplier.fair=true")
.run(context -> {
Supplier<Flux<Message<byte[]>>> sftpSupplier = context.getBean("sftpSupplier", Supplier.class);
HashSet<String> contents = new HashSet<>();
contents.add("source1");
contents.add("source2");
final AtomicReference<Set<String>> expectedContentsOfAllFiles = new AtomicReference<>(contents);
StepVerifier.create(sftpSupplier.get())
.assertNext(message -> {
String payload = new String(message.getPayload());
assertThat(expectedContentsOfAllFiles.get()).contains(payload);
expectedContentsOfAllFiles.get().remove(payload);
})
.expectNextMatches(message -> new String(message.getPayload()).equals("source3"))
.expectNextMatches(message -> expectedContentsOfAllFiles.get()
.contains(new String(message.getPayload())))
.thenCancel()
.verify(Duration.ofSeconds(10));
});
deleteNewSource(newSource);
}
@Test
void supplierMultiSourceRefTestsFor200Alex() throws Exception {
Path newSource = createNewRemoteSource(
Paths.get(remoteTemporaryFolder.toString(), "sftpSecondSource", "sftpSource3.txt"),
"doesNotMatter");
new ApplicationContextRunner()
.withUserConfiguration(TestApp.class)
.withPropertyValues(
"file.consumer.mode = ref",
"sftp.supplier.localDir=" + this.targetLocalDirectory.getAbsolutePath(),
"sftp.supplier.factories.one.host=localhost",
"sftp.supplier.factories.one.port=${sftp.factory.port}",
"sftp.supplier.factories.one.username = user",
"sftp.supplier.factories.one.password = pass",
"sftp.supplier.factories.one.cache-sessions = true",
"sftp.supplier.factories.one.allowUnknownKeys = true",
"sftp.supplier.factories.two.host=localhost",
"sftp.supplier.factories.two.port=${sftp.factory.port}",
"sftp.supplier.factories.two.username = user",
"sftp.supplier.factories.two.password = pass",
"sftp.supplier.factories.two.cache-sessions = true",
"sftp.supplier.factories.two.allowUnknownKeys = true",
"sftp.supplier.factories.empty.host=localhost",
"sftp.supplier.factories.empty.port=${sftp.factory.port}",
"sftp.supplier.factories.empty.username=user",
"sftp.supplier.factories.empty.password=pass",
"sftp.supplier.factories.empty.allowUnknownKeys = true",
"sftp.supplier.directories=one.sftpSource,two.sftpSecondSource,empty.sftpSource",
"sftp.supplier.max-fetch=1",
"sftp.supplier.fair=true")
.run(context -> {
Supplier<Flux<Message<File>>> sftpSupplier = context.getBean("sftpSupplier", Supplier.class);
SftpSupplierProperties properties = context.getBean(SftpSupplierProperties.class);
String localDir = properties.getLocalDir().getPath();
HashSet<String> firstSourceFiles = new HashSet<>();
firstSourceFiles.add(Paths.get(localDir, "sftpSource1.txt").toString());
firstSourceFiles.add(Paths.get(localDir, "sftpSource2.txt").toString());
final AtomicReference<Set<String>> expectedFirstSourcePaths = new AtomicReference<>(
firstSourceFiles);
StepVerifier.create(sftpSupplier.get())
.assertNext(message -> {
assertThat(expectedFirstSourcePaths.get()).contains(message.getPayload().getPath());
expectedFirstSourcePaths.get().remove(message.getPayload().getPath());
})
.expectNextMatches(message -> message.getPayload().getPath().equals(
Paths.get(localDir, "sftpSource3.txt").toString()))
.expectNextMatches(message -> expectedFirstSourcePaths.get()
.contains(message.getPayload().getPath()))
.thenCancel()
.verify(Duration.ofSeconds(10));
});
deleteNewSource(newSource);
}
private Path createNewRemoteSource(Path remotePath, String contents) throws Exception {
Files.createDirectory(remotePath.getParent());
Files.write(Files.createFile(remotePath), contents.getBytes());
return remotePath;
}
private void deleteNewSource(Path newSource) throws IOException {
Files.delete(newSource);
Files.delete(newSource.getParent());
}
@SpringBootApplication
static class TestApp {
}
}

16
supplier/sftp-supplier/src/test/resources/logback.xml Normal file
View File

@@ -0,0 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<layout class="ch.qos.logback.classic.PatternLayout">
<Pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</Pattern>
</layout>
</appender>
<root level="info">
<appender-ref ref="STDOUT" />
</root>
<!-- <logger name="org.springframework.integration.file.remote.aop" level="trace">-->
<!-- <appender-ref ref="STDOUT"/>-->
<!-- </logger>-->
</configuration>