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

SFTP Consumer/Sink

Browse Source 
Resolves https://github.com/spring-cloud/stream-applications/issues/32
This commit is contained in:
Soby Chacko
2020-05-12 20:32:16 -04:00
committed by Gary Russell
parent 566d411a61
commit 5ba3d7ac66
13 changed files with 1004 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
applications/sink/pom.xml
View File

@@ -20,5 +20,6 @@
<module>redis-sink</module>
<module>rabbit-sink</module>
<module>router-sink</module>
<module>sftp-sink</module>
</modules>
</project>

58
applications/sink/sftp-sink/README.adoc Normal file
View File

@@ -0,0 +1,58 @@
//tag::ref-doc[]
= SFTP Sink
SFTP sink is a simple option to push files to an SFTP server from incoming messages.
It uses an `sftp-outbound-adapter`, therefore incoming messages can be either a `java.io.File` object, a `String` (content of the file)
or an array of `bytes` (file content as well).
To use this sink, you need a username and a password to login.
NOTE: By default Spring Integration will use `o.s.i.file.DefaultFileNameGenerator` if none is specified. `DefaultFileNameGenerator` will determine the file name
based on the value of the `file_name` header (if it exists) in the `MessageHeaders`, or if the payload of the `Message` is already a `java.io.File`, then it will
use the original name of that file.
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'`.
== Input
=== Headers
* `file_name` (See note above)
=== Payload
* `java.io.File`
* `java.io.InputStream`
* `byte[]`
* `String`
== Output
N/A (writes to the SFTP server).
== Options
The **$$sftp$$** $$sink$$ has the following options:
//tag::configuration-properties[]
$$sftp.consumer.auto-create-dir$$:: $$Whether or not to create the remote directory.$$ *($$Boolean$$, default: `$$true$$`)*
$$sftp.consumer.factory.allow-unknown-keys$$:: $$True to allow an unknown or changed key.$$ *($$Boolean$$, default: `$$false$$`)*
$$sftp.consumer.factory.cache-sessions$$:: $$Cache sessions.$$ *($$Boolean$$, default: `$$<none>$$`)*
$$sftp.consumer.factory.host$$:: $$The host name of the server.$$ *($$String$$, default: `$$localhost$$`)*
$$sftp.consumer.factory.known-hosts-expression$$:: $$A SpEL expression resolving to the location of the known hosts file.$$ *($$Expression$$, default: `$$<none>$$`)*
$$sftp.consumer.factory.pass-phrase$$:: $$Passphrase for user's private key.$$ *($$String$$, default: `$$<empty string>$$`)*
$$sftp.consumer.factory.password$$:: $$The password to use to connect to the server.$$ *($$String$$, default: `$$<none>$$`)*
$$sftp.consumer.factory.port$$:: $$The port of the server.$$ *($$Integer$$, default: `$$22$$`)*
$$sftp.consumer.factory.private-key$$:: $$Resource location of user's private key.$$ *($$Resource$$, default: `$$<none>$$`)*
$$sftp.consumer.factory.username$$:: $$The username to use to connect to the server.$$ *($$String$$, default: `$$<none>$$`)*
$$sftp.consumer.filename-expression$$:: $$A SpEL expression to generate the remote file name.$$ *($$String$$, default: `$$<none>$$`)*
$$sftp.consumer.mode$$:: $$Action to take if the remote file already exists.$$ *($$FileExistsMode$$, default: `$$<none>$$`, possible values: `APPEND`,`APPEND_NO_FLUSH`,`FAIL`,`IGNORE`,`REPLACE`,`REPLACE_IF_MODIFIED`)*
$$sftp.consumer.remote-dir$$:: $$The remote FTP directory.$$ *($$String$$, default: `$$/$$`)*
$$sftp.consumer.remote-file-separator$$:: $$The remote file separator.$$ *($$String$$, default: `$$/$$`)*
$$sftp.consumer.temporary-remote-dir$$:: $$A temporary directory where the file will be written if 'isUseTemporaryFilename()' is true.$$ *($$String$$, default: `$$/$$`)*
$$sftp.consumer.tmp-file-suffix$$:: $$The suffix to use while the transfer is in progress.$$ *($$String$$, default: `$$.tmp$$`)*
$$sftp.consumer.use-temporary-filename$$:: $$Whether or not to write to a temporary file and rename.$$ *($$Boolean$$, default: `$$true$$`)*
//end::configuration-properties[]
//end::ref-doc[]

78
applications/sink/sftp-sink/pom.xml Normal file
View File

@@ -0,0 +1,78 @@
<?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>sftp-sink</artifactId>
<version>3.0.0-SNAPSHOT</version>
<name>sftp-sink</name>
<description>sftp sink 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>sftp-consumer</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>sftp</name>
<type>sink</type>
<version>${project.version}</version>
<configClass>org.springframework.cloud.fn.consumer.sftp.SftpConsumerConfiguration.class</configClass>
</generatedApp>
<dependencies>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>sftp-consumer</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>

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

@@ -0,0 +1,2 @@
configuration-properties.classes=org.springframework.cloud.fn.consumer.sftp.SftpConsumerProperties, \
org.springframework.cloud.fn.consumer.sftp.SftpConsumerProperties$Factory

9
functions/common/function-test-support/pom.xml
View File

@@ -13,6 +13,10 @@
<relativePath>../../spring-functions-parent</relativePath>
</parent>
<properties>
<sshd-core.version>1.6.0</sshd-core.version>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.ftpserver</groupId>
@@ -35,6 +39,11 @@
<artifactId>spring-boot-starter-test</artifactId>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>org.apache.sshd</groupId>
<artifactId>sshd-core</artifactId>
<version>${sshd-core.version}</version>
</dependency>
</dependencies>
</project>

120
functions/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/sftp/SftpTestSupport.java Normal file
View File

@@ -0,0 +1,120 @@
/*
* 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.test.support.sftp;
import java.io.File;
import java.io.InputStream;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.security.KeyFactory;
import java.security.PublicKey;
import java.security.spec.RSAPublicKeySpec;
import java.util.Arrays;
import java.util.Collections;
import org.apache.sshd.common.file.virtualfs.VirtualFileSystemFactory;
import org.apache.sshd.server.SshServer;
import org.apache.sshd.server.keyprovider.SimpleGeneratorHostKeyProvider;
import org.apache.sshd.server.subsystem.sftp.SftpSubsystemFactory;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.springframework.cloud.fn.test.support.file.remote.RemoteFileTestSupport;
import org.springframework.core.io.ClassPathResource;
import org.springframework.util.Base64Utils;
import org.springframework.util.FileCopyUtils;
import org.springframework.util.StringUtils;
/**
* Provides an embedded SFTP Server for test cases.
*
* @author David Turanski
* @author Gary Russell
* @author Artem Bilan
*/
public class SftpTestSupport extends RemoteFileTestSupport {
private static SshServer server;
@Override
public String prefix() {
return "sftp";
}
@BeforeAll
public static void createServer() throws Exception {
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.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.consumer.localDir",
localTemporaryFolder + File.separator + "localTarget");
}
@AfterAll
public static void stopServer() throws Exception {
server.stop();
File hostkey = new File("hostkey.ser");
if (hostkey.exists()) {
hostkey.delete();
}
System.clearProperty("sftp.consumer.factory.port");
System.clearProperty("sftp.consumer.localDir");
}
private static PublicKey decodePublicKey(String key) {
try {
InputStream stream = new ClassPathResource(key).getInputStream();
byte[] keyBytes = FileCopyUtils.copyToByteArray(stream);
// strip any newline chars
while (keyBytes[keyBytes.length - 1] == 0x0a || keyBytes[keyBytes.length - 1] == 0x0d) {
keyBytes = Arrays.copyOf(keyBytes, keyBytes.length - 1);
}
byte[] decodeBuffer = Base64Utils.decode(keyBytes);
ByteBuffer bb = ByteBuffer.wrap(decodeBuffer);
int len = bb.getInt();
byte[] type = new byte[len];
bb.get(type);
if ("ssh-rsa".equals(new String(type))) {
BigInteger e = decodeBigInt(bb);
BigInteger m = decodeBigInt(bb);
RSAPublicKeySpec spec = new RSAPublicKeySpec(m, e);
return KeyFactory.getInstance("RSA").generatePublic(spec);
}
else {
throw new IllegalArgumentException("Only supports RSA");
}
}
catch (Exception e) {
throw new IllegalStateException("Failed to determine the test public key", e);
}
}
private static BigInteger decodeBigInt(ByteBuffer bb) {
int len = bb.getInt();
byte[] bytes = new byte[len];
bb.get(bytes);
return new BigInteger(bytes);
}
}

46
functions/consumer/sftp-consumer/pom.xml Normal file
View File

@@ -0,0 +1,46 @@
<?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-consumer</artifactId>
<version>1.0.0-SNAPSHOT</version>
<name>sftp-consumer</name>
<description>file consumer</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.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>org.springframework.cloud.fn</groupId>
<artifactId>function-test-support</artifactId>
<version>${project.version}</version>
</dependency>
</dependencies>
</project>

69
functions/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerConfiguration.java Normal file
View File

@@ -0,0 +1,69 @@
/*
* 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.consumer.sftp;
import java.util.function.Consumer;
import com.jcraft.jsch.ChannelSftp;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.expression.ExpressionParser;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlowBuilder;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.sftp.dsl.Sftp;
import org.springframework.integration.sftp.dsl.SftpMessageHandlerSpec;
import org.springframework.integration.sftp.session.SftpRemoteFileTemplate;
import org.springframework.messaging.Message;
@Configuration
@EnableConfigurationProperties(SftpConsumerProperties.class)
@Import(SftpConsumerSessionFactoryConfiguration.class)
public class SftpConsumerConfiguration {
private static final ExpressionParser EXPRESSION_PARSER = new SpelExpressionParser();
@Bean
public IntegrationFlow ftpInboundFlow(SftpConsumerProperties properties, SessionFactory<ChannelSftp.LsEntry> ftpSessionFactory) {
IntegrationFlowBuilder integrationFlowBuilder =
IntegrationFlows.from(MessageConsumer.class, (gateway) -> gateway.beanName("sftpConsumer"));
SftpMessageHandlerSpec handlerSpec =
Sftp.outboundAdapter(new SftpRemoteFileTemplate(ftpSessionFactory), properties.getMode())
.remoteDirectory(properties.getRemoteDir())
.remoteFileSeparator(properties.getRemoteFileSeparator())
.autoCreateDirectory(properties.isAutoCreateDir())
.temporaryFileSuffix(properties.getTmpFileSuffix());
if (properties.getFilenameExpression() != null) {
handlerSpec.fileNameExpression(EXPRESSION_PARSER.parseExpression(properties.getFilenameExpression())
.getExpressionString());
}
return integrationFlowBuilder
.handle(handlerSpec)
.get();
}
private interface MessageConsumer extends Consumer<Message<?>> {
}
}

278
functions/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerProperties.java Normal file
View File

@@ -0,0 +1,278 @@
/*
* 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.consumer.sftp;
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.support.FileExistsMode;
import org.springframework.validation.annotation.Validated;
/**
* @author Gary Russell
* @author Artem Bilan
*/
@ConfigurationProperties("sftp.consumer")
@Validated
public class SftpConsumerProperties {
private final Factory factory = new Factory();
/**
* A temporary directory where the file will be written if 'isUseTemporaryFilename()' is true.
*/
private String temporaryRemoteDir = "/";
/**
* Whether or not to create the remote directory.
*/
private boolean autoCreateDir = true;
/**
* Action to take if the remote file already exists.
*/
private FileExistsMode mode = FileExistsMode.REPLACE;
/**
* Whether or not to write to a temporary file and rename.
*/
private boolean useTemporaryFilename = true;
/**
* A SpEL expression to generate the remote file name.
*/
private String filenameExpression;
/**
* 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 = "/";
@NotBlank
public String getTemporaryRemoteDir() {
return this.temporaryRemoteDir;
}
public void setTemporaryRemoteDir(String temporaryRemoteDir) {
this.temporaryRemoteDir = temporaryRemoteDir;
}
public boolean isAutoCreateDir() {
return this.autoCreateDir;
}
public void setAutoCreateDir(boolean autoCreateDir) {
this.autoCreateDir = autoCreateDir;
}
@NotNull
public FileExistsMode getMode() {
return this.mode;
}
public void setMode(FileExistsMode mode) {
this.mode = mode;
}
public boolean isUseTemporaryFilename() {
return this.useTemporaryFilename;
}
public void setUseTemporaryFilename(boolean useTemporaryFilename) {
this.useTemporaryFilename = useTemporaryFilename;
}
public String getFilenameExpression() {
return this.filenameExpression;
}
public void setFilenameExpression(String filenameExpression) {
this.filenameExpression = filenameExpression;
}
@NotBlank
public String getRemoteDir() {
return remoteDir;
}
public final 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 Factory getFactory() {
return this.factory;
}
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;
/**
* Cache sessions.
*/
private Boolean cacheSessions;
/**
* 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;
}
public Boolean getCacheSessions() {
return this.cacheSessions;
}
public void setCacheSessions(Boolean cacheSessions) {
this.cacheSessions = cacheSessions;
}
@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;
}
}
}

61
functions/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerSessionFactoryConfiguration.java Normal file
View File

@@ -0,0 +1,61 @@
/*
* 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.consumer.sftp;
import com.jcraft.jsch.ChannelSftp;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.file.remote.session.CachingSessionFactory;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.sftp.session.DefaultSftpSessionFactory;
/**
* Session factory configuration.
*
* @author Gary Russell
*
*/
public class SftpConsumerSessionFactoryConfiguration {
@Bean
@ConditionalOnMissingBean
public SessionFactory<ChannelSftp.LsEntry> sftpSessionFactory(SftpConsumerProperties properties, BeanFactory beanFactory) {
DefaultSftpSessionFactory sftpSessionFactory = new DefaultSftpSessionFactory();
SftpConsumerProperties.Factory factory = properties.getFactory();
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) {
sftpSessionFactory.setKnownHosts(factory.getKnownHostsExpression()
.getValue(IntegrationContextUtils.getEvaluationContext(beanFactory), String.class));
}
if (factory.getCacheSessions() != null) {
CachingSessionFactory<ChannelSftp.LsEntry> csf = new CachingSessionFactory<>(sftpSessionFactory);
return csf;
}
else {
return sftpSessionFactory;
}
}
}

203
functions/consumer/sftp-consumer/src/test/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerPropertiesTests.java Normal file
View File

@@ -0,0 +1,203 @@
/*
* 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.consumer.sftp;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.ConfigurationPropertiesBinding;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.test.util.TestPropertyValues;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
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.core.convert.converter.Converter;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.expression.ParseException;
import org.springframework.expression.spel.standard.SpelExpression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.config.IntegrationConverter;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.file.support.FileExistsMode;
import org.springframework.integration.test.util.TestUtils;
import static org.assertj.core.api.Assertions.assertThat;
/**
* @author David Turanski
* @author Gary Russell
* @author Artem Bilan
* @author Chris Schaefer
*/
public class SftpConsumerPropertiesTests {
@Test
public void remoteDirCanBeCustomized() {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
testPropertyValues(context, "sftp.consumer.remoteDir:/remote");
context.register(Conf.class);
context.refresh();
SftpConsumerProperties properties = context.getBean(SftpConsumerProperties.class);
assertThat(properties.getRemoteDir()).isEqualTo("/remote");
context.close();
}
@Test
public void autoCreateDirCanBeDisabled() {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
testPropertyValues(context, "sftp.consumer.autoCreateDir:false");
context.register(Conf.class);
context.refresh();
SftpConsumerProperties properties = context.getBean(SftpConsumerProperties.class);
assertThat(!properties.isAutoCreateDir()).isTrue();
context.close();
}
@Test
public void tmpFileSuffixCanBeCustomized() {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
testPropertyValues(context, "sftp.consumer.tmpFileSuffix:.foo");
context.register(Conf.class);
context.refresh();
SftpConsumerProperties properties = context.getBean(SftpConsumerProperties.class);
assertThat(properties.getTmpFileSuffix()).isEqualTo(".foo");
context.close();
}
@Test
public void tmpFileRemoteDirCanBeCustomized() {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
testPropertyValues(context, "sftp.consumer.temporaryRemoteDir:/foo");
context.register(Conf.class);
context.refresh();
SftpConsumerProperties properties = context.getBean(SftpConsumerProperties.class);
assertThat(properties.getTemporaryRemoteDir()).isEqualTo("/foo");
context.close();
}
@Test
public void remoteFileSeparatorCanBeCustomized() {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
testPropertyValues(context, "sftp.consumer.remoteFileSeparator:\\");
context.register(Conf.class);
context.refresh();
SftpConsumerProperties properties = context.getBean(SftpConsumerProperties.class);
assertThat(properties.getRemoteFileSeparator()).isEqualTo("\\");
context.close();
}
@Test
public void useTemporaryFileNameCanBeCustomized() {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
testPropertyValues(context, "sftp.consumer.useTemporaryFilename:false");
context.register(Conf.class);
context.refresh();
SftpConsumerProperties properties = context.getBean(SftpConsumerProperties.class);
assertThat(properties.isUseTemporaryFilename()).isFalse();
context.close();
}
@Test
public void fileExistsModeCanBeCustomized() {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
testPropertyValues(context, "sftp.consumer.mode:FAIL");
context.register(Conf.class);
context.refresh();
SftpConsumerProperties properties = context.getBean(SftpConsumerProperties.class);
assertThat(properties.getMode()).isEqualTo(FileExistsMode.FAIL);
context.close();
}
@Test
public void knownHostsExpression() {
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
testPropertyValues(context,
"sftp.consumer.factory.known-hosts-expression = @systemProperties[\"user.home\"] + \"/.ssh/known_hosts\"",
"sftp.consumer.factory.cache-sessions = true");
context.register(Factory.class);
context.refresh();
SessionFactory<?> sessionFactory = context.getBean(SessionFactory.class);
assertThat(TestUtils.getPropertyValue(sessionFactory, "sessionFactory.knownHosts").toString().endsWith(
"/.ssh/known_hosts]")).isTrue();
context.close();
}
private void testPropertyValues(ConfigurableApplicationContext context, String... props) {
TestPropertyValues.of("sftp.consumer.factory.username=foo").and(props).applyTo(context);
}
@Configuration
@EnableConfigurationProperties(SftpConsumerProperties.class)
static class Conf {
}
@Configuration
@EnableConfigurationProperties(SftpConsumerProperties.class)
@EnableIntegration
@Import(SftpConsumerSessionFactoryConfiguration.class)
static class Factory {
@Bean
@ConfigurationPropertiesBinding
@IntegrationConverter
public Converter<String, Expression> spelConverter() {
return new SpelConverter();
}
/**
* TODO: This needs to be refactored into a generic place for any functions to use.
*
* A simple converter from String to Expression.
*
* @author Eric Bottard
*/
public static class SpelConverter implements Converter<String, Expression> {
private SpelExpressionParser parser = new SpelExpressionParser();
@Autowired
@Qualifier(IntegrationContextUtils.INTEGRATION_EVALUATION_CONTEXT_BEAN_NAME)
@Lazy
private EvaluationContext evaluationContext;
@Override
public Expression convert(String source) {
try {
Expression expression = this.parser.parseExpression(source);
if (expression instanceof SpelExpression) {
((SpelExpression) expression)
.setEvaluationContext(this.evaluationContext);
}
return expression;
}
catch (ParseException e) {
throw new IllegalArgumentException(String.format(
"Could not convert '%s' into a SpEL expression", source), e);
}
}
}
}
}

78
functions/consumer/sftp-consumer/src/test/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerTests.java Normal file
View File

@@ -0,0 +1,78 @@
/*
* 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.consumer.sftp;
import java.io.File;
import java.util.function.Consumer;
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.test.support.sftp.SftpTestSupport;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import static org.assertj.core.api.Assertions.assertThat;
@DirtiesContext
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE,
properties = {
"sftp.consumer.remoteDir = sftpTarget",
"sftp.consumer.factory.username = foo",
"sftp.consumer.factory.password = foo",
"sftp.consumer.mode = FAIL",
"sftp.consumer.filenameExpression = payload.name.toUpperCase()",
"sftp.consumer.factory.allowUnknownKeys = true"
})
public class SftpConsumerTests extends SftpTestSupport {
@Autowired
Consumer<Message<?>> sftpConsumer;
@Test
public void sendFiles() {
for (int i = 1; i <= 2; i++) {
String pathname = "/localSource" + i + ".txt";
String upperPathname = pathname.toUpperCase();
new File(getTargetRemoteDirectory() + upperPathname).delete();
assertThat(new File(getTargetRemoteDirectory() + upperPathname).exists()).isFalse();
sftpConsumer.accept(new GenericMessage<>(new File(getSourceLocalDirectory() + pathname)));
File expected = new File(getTargetRemoteDirectory() + upperPathname);
assertThat(expected.exists()).isTrue();
// verify the uppercase on a case-insensitive file system
File[] files = getTargetRemoteDirectory().listFiles();
for (File file : files) {
assertThat(file.getName().startsWith("LOCALSOURCE")).isTrue();
}
}
}
@Test
public void serverRefreshed() { // noop test to test the dirs are refreshed properly
String pathname = "/LOCALSOURCE1.TXT";
assertThat(getTargetRemoteDirectory().exists()).isTrue();
assertThat(new File(getTargetRemoteDirectory() + pathname).exists()).isFalse();
}
@SpringBootApplication
static class TestApplication {
}
}

1
functions/pom.xml
View File

@@ -52,6 +52,7 @@
<module>consumer/mongodb-consumer</module>
<module>consumer/rabbit-consumer</module>
<module>consumer/redis-consumer</module>
<module>consumer/sftp-consumer</module>
<module>function/filter-function</module>
<module>function/header-enricher-function</module>