csf = new CachingSessionFactory<>(ftpSessionFactory);
+ return csf;
+ }
+ else {
+ return ftpSessionFactory;
+ }
+ }
+
+}
diff --git a/common/ftp-common/src/main/java/org/springframework/cloud/fn/common/ftp/FtpSessionFactoryProperties.java b/common/ftp-common/src/main/java/org/springframework/cloud/fn/common/ftp/FtpSessionFactoryProperties.java
new file mode 100644
index 00000000..80491c1e
--- /dev/null
+++ b/common/ftp-common/src/main/java/org/springframework/cloud/fn/common/ftp/FtpSessionFactoryProperties.java
@@ -0,0 +1,137 @@
+/*
+ * 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.common.ftp;
+
+import javax.validation.constraints.NotBlank;
+import javax.validation.constraints.NotNull;
+
+import org.apache.commons.net.ftp.FTPClient;
+import org.hibernate.validator.constraints.Range;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.validation.annotation.Validated;
+
+@ConfigurationProperties("ftp.factory")
+@Validated
+public class FtpSessionFactoryProperties {
+
+ /**
+ * The port of the server.
+ */
+ private int port = 21;
+
+ /**
+ * The client mode to use for the FTP session.
+ */
+ private ClientMode clientMode = ClientMode.PASSIVE;
+
+ /**
+ * 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;
+
+ @Range(min = 0, max = 65535)
+ public int getPort() {
+ return this.port;
+ }
+
+ public void setPort(int port) {
+ this.port = port;
+ }
+
+ @NotNull
+ public ClientMode getClientMode() {
+ return this.clientMode;
+ }
+
+ public void setClientMode(ClientMode clientMode) {
+ this.clientMode = clientMode;
+ }
+
+ public enum ClientMode {
+
+ /**
+ * Active client mode.
+ */
+ ACTIVE(FTPClient.ACTIVE_LOCAL_DATA_CONNECTION_MODE),
+ /**
+ * Passive client mode.
+ */
+ PASSIVE(FTPClient.PASSIVE_LOCAL_DATA_CONNECTION_MODE);
+
+ private final int mode;
+
+ ClientMode(int mode) {
+ this.mode = mode;
+ }
+
+ public int getMode() {
+ return mode;
+ }
+
+ }
+
+ @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;
+ }
+
+}
diff --git a/common/function-test-support/pom.xml b/common/function-test-support/pom.xml
new file mode 100644
index 00000000..2766569f
--- /dev/null
+++ b/common/function-test-support/pom.xml
@@ -0,0 +1,40 @@
+
+
+ 4.0.0
+ function-test-support
+ 1.0.0-SNAPSHOT
+ function-test-support
+ file consumer
+
+
+ org.springframework.cloud.fn
+ spring-functions-parent
+ 1.0.0-SNAPSHOT
+ ../../spring-functions-parent
+
+
+
+
+ org.apache.ftpserver
+ ftpserver-core
+ 1.1.1
+ compile
+
+
+ org.springframework.integration
+ spring-integration-ftp
+ true
+
+
+ org.springframework.integration
+ spring-integration-test
+ compile
+
+
+ org.springframework.boot
+ spring-boot-starter-test
+ compile
+
+
+
+
diff --git a/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/file/remote/RemoteFileTestSupport.java b/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/file/remote/RemoteFileTestSupport.java
new file mode 100644
index 00000000..adf735c6
--- /dev/null
+++ b/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/file/remote/RemoteFileTestSupport.java
@@ -0,0 +1,154 @@
+/*
+ * Copyright 2015-2016 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.file.remote;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.file.Path;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Abstract base class for tests requiring remote file servers, e.g. (S)FTP.
+ *
+ * @author Gary Russell
+ *
+ */
+public abstract class RemoteFileTestSupport {
+
+ protected static final int port = 0;
+
+ @TempDir
+ protected static Path remoteTemporaryFolder;
+
+ @TempDir
+ protected static Path localTemporaryFolder;
+
+ protected volatile File sourceRemoteDirectory;
+
+ protected volatile File targetRemoteDirectory;
+
+ protected volatile File sourceLocalDirectory;
+
+ protected volatile File targetLocalDirectory;
+
+ public File getSourceRemoteDirectory() {
+ return sourceRemoteDirectory;
+ }
+
+ public File getTargetRemoteDirectory() {
+ return targetRemoteDirectory;
+ }
+
+ public File getSourceLocalDirectory() {
+ return sourceLocalDirectory;
+ }
+
+ public File getTargetLocalDirectory() {
+ return targetLocalDirectory;
+ }
+
+ /**
+ * Default implementation creates the following folder structures:
+ *
+ *
+ * $ tree remoteSource/
+ * remoteSource/
+ * ├── remoteSource1.txt - contains 'source1'
+ * ├── remoteSource2.txt - contains 'source2'
+ * remoteTarget/
+ * $ tree localSource/
+ * localSource/
+ * ├── localSource1.txt - contains 'local1'
+ * ├── localSource2.txt - contains 'local2'
+ * localTarget/
+ *
+ *
+ * The intent is tests retrieve from remoteSource and verify arrival in localTarget or send from localSource and verify
+ * arrival in remoteTarget.
+ *
+ * Subclasses can change 'remote' in these names by overriding {@link #prefix()} or override this method completely to
+ * create a different structure.
+ *
+ * While a single server exists for all tests, the directory structure is rebuilt for each test.
+ * @throws IOException IO Exception.
+ */
+ @BeforeEach
+ public void setupFolders() throws IOException {
+ String prefix = prefix();
+ recursiveDelete(new File(remoteTemporaryFolder.toFile(), prefix + "Source"));
+
+ sourceRemoteDirectory = new File(remoteTemporaryFolder.toFile(), prefix + "Source");
+ sourceRemoteDirectory.mkdirs();
+ recursiveDelete(new File(remoteTemporaryFolder.toFile(), prefix + "Target"));
+ targetRemoteDirectory = new File(remoteTemporaryFolder.toFile(), prefix + "Target");
+ targetRemoteDirectory.mkdirs();
+ recursiveDelete(new File(localTemporaryFolder.toFile(), "localSource"));
+ sourceLocalDirectory = new File(localTemporaryFolder.toFile(), "localSource");
+ sourceLocalDirectory.mkdirs();
+ recursiveDelete(new File(localTemporaryFolder.toFile(), "localTarget"));
+ targetLocalDirectory = new File(localTemporaryFolder.toFile(), "localTarget");
+ targetLocalDirectory.mkdirs();
+ File file = new File(sourceRemoteDirectory, prefix + "Source1.txt");
+ file.createNewFile();
+ FileOutputStream fos = new FileOutputStream(file);
+ fos.write("source1".getBytes());
+ fos.close();
+ file = new File(sourceRemoteDirectory, prefix + "Source2.txt");
+ file.createNewFile();
+ fos = new FileOutputStream(file);
+ fos.write("source2".getBytes());
+ fos.close();
+ file = new File(sourceLocalDirectory, "localSource1.txt");
+ file.createNewFile();
+ fos = new FileOutputStream(file);
+ fos.write("local1".getBytes());
+ fos.close();
+ file = new File(sourceLocalDirectory, "localSource2.txt");
+ file.createNewFile();
+ fos = new FileOutputStream(file);
+ fos.write("local2".getBytes());
+ fos.close();
+ }
+
+ public static void recursiveDelete(File file) {
+ if (file != null && file.exists()) {
+ File[] files = file.listFiles();
+ if (files != null) {
+ for (File fyle : files) {
+ if (fyle.isDirectory()) {
+ recursiveDelete(fyle);
+ }
+ else {
+ fyle.delete();
+ }
+ }
+ }
+ file.delete();
+ }
+ }
+
+ /**
+ * Prefix for directory/file structure; default 'remote'.
+ * @return the prefix.
+ */
+ protected String prefix() {
+ return "remote";
+ }
+}
diff --git a/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/ftp/FtpTestSupport.java b/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/ftp/FtpTestSupport.java
new file mode 100644
index 00000000..0788f664
--- /dev/null
+++ b/common/function-test-support/src/main/java/org/springframework/cloud/fn/test/support/ftp/FtpTestSupport.java
@@ -0,0 +1,129 @@
+/*
+ * 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.ftp;
+
+import java.io.File;
+import java.util.Arrays;
+
+import org.apache.ftpserver.FtpServer;
+import org.apache.ftpserver.FtpServerFactory;
+import org.apache.ftpserver.ftplet.Authentication;
+import org.apache.ftpserver.ftplet.AuthenticationFailedException;
+import org.apache.ftpserver.ftplet.FtpException;
+import org.apache.ftpserver.ftplet.User;
+import org.apache.ftpserver.ftplet.UserManager;
+import org.apache.ftpserver.listener.ListenerFactory;
+import org.apache.ftpserver.usermanager.impl.BaseUser;
+import org.apache.ftpserver.usermanager.impl.ConcurrentLoginPermission;
+import org.apache.ftpserver.usermanager.impl.TransferRatePermission;
+import org.apache.ftpserver.usermanager.impl.WritePermission;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+
+import org.springframework.cloud.fn.test.support.file.remote.RemoteFileTestSupport;
+
+public class FtpTestSupport extends RemoteFileTestSupport {
+
+ private static final FtpServerFactory serverFactory = new FtpServerFactory();
+
+ private static volatile FtpServer server;
+
+ public String getTargetLocalDirectoryName() {
+ return targetLocalDirectory.getAbsolutePath() + File.separator;
+ }
+
+ @BeforeAll
+ public static void createServer() throws Exception {
+ serverFactory.setUserManager(new TestUserManager(remoteTemporaryFolder.toFile().getAbsolutePath()));
+
+ ListenerFactory factory = new ListenerFactory();
+ factory.setPort(0);
+ serverFactory.addListener("default", factory.createListener());
+
+ server = serverFactory.createServer();
+ server.start();
+ System.setProperty("ftp.factory.port", String.valueOf(serverFactory.getListener("default").getPort()));
+ System.setProperty("ftp.localDir",
+ localTemporaryFolder.toFile().getAbsolutePath() + File.separator + "localTarget");
+ }
+
+ @AfterAll
+ public static void stopServer() throws Exception {
+ server.stop();
+ System.clearProperty("ftp.factory.port");
+ System.clearProperty("ftp.localDir");
+ }
+
+ @Override
+ protected String prefix() {
+ return "ftp";
+ }
+
+ private static final class TestUserManager implements UserManager {
+
+ private final BaseUser testUser;
+
+ private TestUserManager(String homeDirectory) {
+ this.testUser = new BaseUser();
+ this.testUser.setAuthorities(Arrays.asList(new ConcurrentLoginPermission(1024, 1024),
+ new WritePermission(),
+ new TransferRatePermission(1024, 1024)));
+ this.testUser.setHomeDirectory(homeDirectory);
+ this.testUser.setName("TEST_USER");
+ }
+
+
+ @Override
+ public User getUserByName(String s) throws FtpException {
+ return this.testUser;
+ }
+
+ @Override
+ public String[] getAllUserNames() throws FtpException {
+ return new String[] { "TEST_USER" };
+ }
+
+ @Override
+ public void delete(String s) throws FtpException {
+ }
+
+ @Override
+ public void save(User user) throws FtpException {
+ }
+
+ @Override
+ public boolean doesExist(String s) throws FtpException {
+ return true;
+ }
+
+ @Override
+ public User authenticate(Authentication authentication) throws AuthenticationFailedException {
+ return this.testUser;
+ }
+
+ @Override
+ public String getAdminName() throws FtpException {
+ return "admin";
+ }
+
+ @Override
+ public boolean isAdmin(String s) throws FtpException {
+ return s.equals("admin");
+ }
+
+ }
+}
diff --git a/consumer/ftp-consumer/pom.xml b/consumer/ftp-consumer/pom.xml
new file mode 100644
index 00000000..5dc9a868
--- /dev/null
+++ b/consumer/ftp-consumer/pom.xml
@@ -0,0 +1,51 @@
+
+
+ 4.0.0
+ ftp-consumer
+ 1.0.0-SNAPSHOT
+ ftp-consumer
+ file consumer
+
+
+ org.springframework.cloud.fn
+ spring-functions-parent
+ 1.0.0-SNAPSHOT
+ ../../spring-functions-parent
+
+
+
+
+ org.springframework.cloud.fn
+ ftp-common
+ ${project.version}
+
+
+ org.springframework.integration
+ spring-integration-ftp
+
+
+ org.springframework.boot
+ spring-boot-starter-integration
+
+
+ org.springframework.boot
+ spring-boot-starter-validation
+
+
+ org.springframework.boot
+ spring-boot-configuration-processor
+ provided
+
+
+ org.springframework.boot
+ spring-boot-starter-test
+ test
+
+
+ org.springframework.cloud.fn
+ function-test-support
+ ${project.version}
+
+
+
+
diff --git a/consumer/ftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/ftp/FtpConsumerConfiguration.java b/consumer/ftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/ftp/FtpConsumerConfiguration.java
new file mode 100644
index 00000000..3a3bb67c
--- /dev/null
+++ b/consumer/ftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/ftp/FtpConsumerConfiguration.java
@@ -0,0 +1,74 @@
+/*
+ * 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.ftp;
+
+import java.util.function.Consumer;
+
+import org.apache.commons.net.ftp.FTPFile;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+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.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.ftp.dsl.Ftp;
+import org.springframework.integration.ftp.dsl.FtpMessageHandlerSpec;
+import org.springframework.integration.ftp.session.FtpRemoteFileTemplate;
+import org.springframework.messaging.Message;
+
+@Configuration
+@EnableConfigurationProperties(FtpConsumerProperties.class)
+@Import(FtpSessionFactoryConfiguration.class)
+public class FtpConsumerConfiguration {
+
+ private static final ExpressionParser EXPRESSION_PARSER = new SpelExpressionParser();
+
+ @Autowired
+ FtpConsumerProperties ftpConsumerProperties;
+
+ @Bean
+ public IntegrationFlow ftpInboundFlow(FtpConsumerProperties properties, SessionFactory ftpSessionFactory) {
+
+ IntegrationFlowBuilder integrationFlowBuilder =
+ IntegrationFlows.from(MessageConsumer.class, (gateway) -> gateway.beanName("ftpConsumer"));
+
+ FtpMessageHandlerSpec handlerSpec =
+ Ftp.outboundAdapter(new FtpRemoteFileTemplate(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> {
+
+ }
+
+}
diff --git a/consumer/ftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/ftp/FtpConsumerProperties.java b/consumer/ftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/ftp/FtpConsumerProperties.java
new file mode 100644
index 00000000..a303e2f5
--- /dev/null
+++ b/consumer/ftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/ftp/FtpConsumerProperties.java
@@ -0,0 +1,139 @@
+/*
+ * Copyright 2015-2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.cloud.fn.consumer.ftp;
+
+import javax.validation.constraints.NotBlank;
+import javax.validation.constraints.NotNull;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.integration.file.support.FileExistsMode;
+import org.springframework.validation.annotation.Validated;
+
+@ConfigurationProperties("ftp.consumer")
+@Validated
+public class FtpConsumerProperties {
+
+ /**
+ * 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 = "/";
+
+ /**
+ * 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;
+
+ @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 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;
+ }
+}
diff --git a/consumer/ftp-consumer/src/test/java/org/springframework/cloud/fn/consumer/ftp/FtpConsumerPropertiesTests.java b/consumer/ftp-consumer/src/test/java/org/springframework/cloud/fn/consumer/ftp/FtpConsumerPropertiesTests.java
new file mode 100644
index 00000000..4d449a9c
--- /dev/null
+++ b/consumer/ftp-consumer/src/test/java/org/springframework/cloud/fn/consumer/ftp/FtpConsumerPropertiesTests.java
@@ -0,0 +1,125 @@
+/*
+ * 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.ftp;
+
+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 org.springframework.integration.file.support.FileExistsMode;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * @author David Turanski
+ * @author Gary Russell
+ * @author Artem Bilan
+ */
+public class FtpConsumerPropertiesTests {
+
+ @Test
+ public void remoteDirCanBeCustomized() {
+ AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
+ TestPropertyValues.of("ftp.consumer.remoteDir:/remote")
+ .applyTo(context);
+ context.register(Conf.class);
+ context.refresh();
+ FtpConsumerProperties properties = context.getBean(FtpConsumerProperties.class);
+ assertThat(properties.getRemoteDir()).isEqualTo("/remote");
+ context.close();
+ }
+
+ @Test
+ public void autoCreateDirCanBeDisabled() {
+ AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
+ TestPropertyValues.of("ftp.consumer.autoCreateDir:false")
+ .applyTo(context);
+ context.register(Conf.class);
+ context.refresh();
+ FtpConsumerProperties properties = context.getBean(FtpConsumerProperties.class);
+ assertThat(!properties.isAutoCreateDir()).isTrue();
+ context.close();
+ }
+
+ @Test
+ public void tmpFileSuffixCanBeCustomized() {
+ AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
+ TestPropertyValues.of("ftp.consumer.tmpFileSuffix:.foo")
+ .applyTo(context);
+ context.register(Conf.class);
+ context.refresh();
+ FtpConsumerProperties properties = context.getBean(FtpConsumerProperties.class);
+ assertThat(properties.getTmpFileSuffix()).isEqualTo(".foo");
+ context.close();
+ }
+
+ @Test
+ public void tmpFileRemoteDirCanBeCustomized() {
+ AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
+ TestPropertyValues.of("ftp.consumer.temporaryRemoteDir:/foo")
+ .applyTo(context);
+ context.register(Conf.class);
+ context.refresh();
+ FtpConsumerProperties properties = context.getBean(FtpConsumerProperties.class);
+ assertThat(properties.getTemporaryRemoteDir()).isEqualTo("/foo");
+ context.close();
+ }
+
+ @Test
+ public void remoteFileSeparatorCanBeCustomized() {
+ AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
+ TestPropertyValues.of("ftp.consumer.remoteFileSeparator:\\")
+ .applyTo(context);
+ context.register(Conf.class);
+ context.refresh();
+ FtpConsumerProperties properties = context.getBean(FtpConsumerProperties.class);
+ assertThat(properties.getRemoteFileSeparator()).isEqualTo("\\");
+ context.close();
+ }
+
+ @Test
+ public void useTemporaryFileNameCanBeCustomized() {
+ AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
+ TestPropertyValues.of("ftp.consumer.useTemporaryFilename:false")
+ .applyTo(context);
+ context.register(Conf.class);
+ context.refresh();
+ FtpConsumerProperties properties = context.getBean(FtpConsumerProperties.class);
+ assertThat(properties.isUseTemporaryFilename()).isFalse();
+ context.close();
+ }
+
+ @Test
+ public void fileExistsModeCanBeCustomized() {
+ AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext();
+ TestPropertyValues.of("ftp.consumer.mode:FAIL")
+ .applyTo(context);
+ context.register(Conf.class);
+ context.refresh();
+ FtpConsumerProperties properties = context.getBean(FtpConsumerProperties.class);
+ assertThat(properties.getMode()).isEqualTo(FileExistsMode.FAIL);
+ context.close();
+ }
+
+ @Configuration
+ @EnableConfigurationProperties(FtpConsumerProperties.class)
+ static class Conf {
+
+ }
+}
diff --git a/consumer/ftp-consumer/src/test/java/org/springframework/cloud/fn/consumer/ftp/FtpConsumerTests.java b/consumer/ftp-consumer/src/test/java/org/springframework/cloud/fn/consumer/ftp/FtpConsumerTests.java
new file mode 100644
index 00000000..b2888656
--- /dev/null
+++ b/consumer/ftp-consumer/src/test/java/org/springframework/cloud/fn/consumer/ftp/FtpConsumerTests.java
@@ -0,0 +1,76 @@
+/*
+ * 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.ftp;
+
+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.ftp.FtpTestSupport;
+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 = {
+ "ftp.consumer.remoteDir = ftpTarget",
+ "ftp.factory.username = foo",
+ "ftp.factory.password = foo",
+ "ftp.consumer.mode = FAIL",
+ "ftp.consumer.filenameExpression = payload.name.toUpperCase()"
+ })
+public class FtpConsumerTests extends FtpTestSupport {
+
+ @Autowired
+ Consumer> ftpConsumer;
+
+ @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();
+ ftpConsumer.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 {
+ }
+}
diff --git a/pom.xml b/pom.xml
index 5007c6d4..fd22ecbe 100644
--- a/pom.xml
+++ b/pom.xml
@@ -40,9 +40,13 @@
+ common/ftp-common
+ common/function-test-support
+
consumer/cassandra-consumer
consumer/counter-consumer
consumer/file-consumer
+ consumer/ftp-consumer
consumer/jdbc-consumer
consumer/log-consumer
consumer/mongodb-consumer