Ftp consumer/sink

Resolves https://github.com/spring-cloud/stream-applications/issues/28

common/ftp-common/pom.xml

@@ -0,0 +1,41 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>ftp-common</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>ftp-common</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-ftp</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>
+    </dependencies>
+
+</project>

common/ftp-common/src/main/java/org/springframework/cloud/fn/common/ftp/FtpSessionFactoryConfiguration.java

@@ -0,0 +1,51 @@
+/*
+ * 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 org.apache.commons.net.ftp.FTPFile;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.integration.file.remote.session.CachingSessionFactory;
+import org.springframework.integration.file.remote.session.SessionFactory;
+import org.springframework.integration.ftp.session.DefaultFtpSessionFactory;
+
+@Configuration
+@EnableConfigurationProperties(FtpSessionFactoryProperties.class)
+public class FtpSessionFactoryConfiguration {
+
+	@Bean
+	@ConditionalOnMissingBean
+	public SessionFactory<FTPFile> ftpSessionFactory(FtpSessionFactoryProperties properties) {
+		DefaultFtpSessionFactory ftpSessionFactory = new DefaultFtpSessionFactory();
+		ftpSessionFactory.setHost(properties.getHost());
+		ftpSessionFactory.setPort(properties.getPort());
+		ftpSessionFactory.setUsername(properties.getUsername());
+		ftpSessionFactory.setPassword(properties.getPassword());
+		ftpSessionFactory.setClientMode(properties.getClientMode().getMode());
+		if (properties.getCacheSessions() != null) {
+			CachingSessionFactory<FTPFile> csf = new CachingSessionFactory<>(ftpSessionFactory);
+			return csf;
+		}
+		else {
+			return ftpSessionFactory;
+		}
+	}
+
+}

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

common/function-test-support/pom.xml

@@ -0,0 +1,40 @@
+<?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>function-test-support</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>function-test-support</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.apache.ftpserver</groupId>
+            <artifactId>ftpserver-core</artifactId>
+            <version>1.1.1</version>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-ftp</artifactId>
+            <optional>true</optional>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-test</artifactId>
+            <scope>compile</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>compile</scope>
+        </dependency>
+    </dependencies>
+
+</project>

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:
+	 *
+	 * <pre class="code">
+	 *  $ tree remoteSource/
+	 *  remoteSource/
+	 *  ├── remoteSource1.txt - contains 'source1'
+	 *  ├── remoteSource2.txt - contains 'source2'
+	 *  remoteTarget/
+	 *  $ tree localSource/
+	 *  localSource/
+	 *  ├── localSource1.txt - contains 'local1'
+	 *  ├── localSource2.txt - contains 'local2'
+	 *  localTarget/
+	 * </pre>
+	 *
+	 * The intent is tests retrieve from remoteSource and verify arrival in localTarget or send from localSource and verify
+	 * arrival in remoteTarget.
+	 * <p>
+	 * Subclasses can change 'remote' in these names by overriding {@link #prefix()} or override this method completely to
+	 * create a different structure.
+	 * <p>
+	 * 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";
+	}
+}

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");
+		}
+
+	}
+}

consumer/ftp-consumer/pom.xml

@@ -0,0 +1,51 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <artifactId>ftp-consumer</artifactId>
+    <version>1.0.0-SNAPSHOT</version>
+    <name>ftp-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.cloud.fn</groupId>
+            <artifactId>ftp-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.integration</groupId>
+            <artifactId>spring-integration-ftp</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-integration</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.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>

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<FTPFile> 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<Message<?>> {
+
+	}
+
+}

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

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

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<Message<?>> 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 {
+	}
+}

pom.xml

@@ -40,9 +40,13 @@
 	</properties>
 
   	<modules>
+		<module>common/ftp-common</module>
+		<module>common/function-test-support</module>
+
 		<module>consumer/cassandra-consumer</module>
 		<module>consumer/counter-consumer</module>
 		<module>consumer/file-consumer</module>
+		<module>consumer/ftp-consumer</module>
 		<module>consumer/jdbc-consumer</module>
 		<module>consumer/log-consumer</module>
 		<module>consumer/mongodb-consumer</module>