GH-167: rename remote files in sftp-supplier

Fixes https://github.com/spring-cloud/stream-applications/issues/167

* Support for renaming remote files after processing via 'sftp.supplier.rename-remote-files-to' property.

common/file-common/src/main/java/org/springframework/cloud/fn/common/file/remote/RemoteFileRenamingAdvice.java

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

supplier/sftp-supplier/pom.xml

@@ -30,6 +30,11 @@
             <artifactId>spring-integration-sftp</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.springframework.cloud.fn</groupId>
+            <artifactId>config-common</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.springframework.cloud.fn</groupId>
             <artifactId>file-common</artifactId>

supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java

@@ -38,6 +38,7 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.cloud.fn.common.file.FileConsumerProperties;
 import org.springframework.cloud.fn.common.file.FileUtils;
 import org.springframework.cloud.fn.common.file.remote.RemoteFileDeletingAdvice;
+import org.springframework.cloud.fn.common.file.remote.RemoteFileRenamingAdvice;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
@@ -48,16 +49,19 @@ import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.core.GenericSelector;
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.dsl.IntegrationFlow;
+import org.springframework.integration.dsl.IntegrationFlowBuilder;
 import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.filters.ChainFileListFilter;
 import org.springframework.integration.file.filters.FileListFilter;
+import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.handler.MessageProcessor;
 import org.springframework.integration.metadata.ConcurrentMetadataStore;
 import org.springframework.integration.sftp.dsl.Sftp;
 import org.springframework.integration.sftp.dsl.SftpInboundChannelAdapterSpec;
+import org.springframework.integration.sftp.dsl.SftpOutboundGatewaySpec;
 import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter;
 import org.springframework.integration.sftp.filters.SftpRegexPatternFileListFilter;
 import org.springframework.integration.sftp.filters.SftpSimplePatternFileListFilter;
@@ -66,6 +70,7 @@ import org.springframework.integration.util.IntegrationReactiveUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
+import org.springframework.messaging.MessageHandler;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
 import org.springframework.messaging.PollableChannel;
@@ -97,9 +102,9 @@ public class SftpSupplierConfiguration {
 
 	@Bean
 	public Supplier<Flux<? extends Message<?>>> sftpSupplier(MessageSource<?> sftpMessageSource,
-			@Nullable Publisher<Message<Object>> sftpReadingFlow,
-			MonoProcessor<Boolean> subscriptionBarrier,
-			SftpSupplierProperties sftpSupplierProperties) {
+															@Nullable Publisher<Message<Object>> sftpReadingFlow,
+															MonoProcessor<Boolean> subscriptionBarrier,
+															SftpSupplierProperties sftpSupplierProperties) {
 
 		Flux<? extends Message<?>> flux = sftpReadingFlow == null
 				? sftpMessageFlux(sftpMessageSource, sftpSupplierProperties, subscriptionBarrier)
@@ -221,6 +226,12 @@ public class SftpSupplierConfiguration {
 			return new RemoteFileDeletingAdvice(sftpTemplate, sftpSupplierProperties.getRemoteFileSeparator());
 		}
 
+		@Bean
+		@ConditionalOnProperty(prefix = "sftp.supplier", value = "rename-remote-files-to")
+		public RemoteFileRenamingAdvice remoteFileRenamingAdvice(SftpRemoteFileTemplate sftpTemplate,
+																SftpSupplierProperties sftpSupplierProperties) {
+			return new RemoteFileRenamingAdvice(sftpTemplate, sftpSupplierProperties.getRemoteFileSeparator(), sftpSupplierProperties.getRenameRemoteFilesTo());
+		}
 	}
 
 	@Configuration
@@ -240,16 +251,24 @@ public class SftpSupplierConfiguration {
 				MessageSource<?> sftpMessageSource,
 				MonoProcessor<?> subscriptionBarrier,
 				SftpSupplierProperties sftpSupplierProperties,
-				FileConsumerProperties fileConsumerProperties) {
+				FileConsumerProperties fileConsumerProperties,
+				@Nullable MessageHandler renameRemoteFileHandler) {
 
-			return FileUtils.enhanceFlowForReadingMode(IntegrationFlows
+			IntegrationFlowBuilder flowBuilder = FileUtils.enhanceFlowForReadingMode(IntegrationFlows
 							.from(IntegrationReactiveUtils.messageSourceToFlux(sftpMessageSource)
 									.delaySubscription(subscriptionBarrier)
 									.subscriberContext(
 											context -> (context.put(IntegrationReactiveUtils.DELAY_WHEN_EMPTY_KEY,
 													sftpSupplierProperties.getDelayWhenEmpty())))),
-					fileConsumerProperties)
-					.toReactivePublisher();
+					fileConsumerProperties);
+
+			if (renameRemoteFileHandler != null) {
+				flowBuilder.publishSubscribeChannel(pubsub ->
+						pubsub.subscribe(subFlow -> subFlow.handle(renameRemoteFileHandler).nullChannel())
+				);
+			}
+
+			return flowBuilder.toReactivePublisher();
 		}
 
 		/**
@@ -276,6 +295,12 @@ public class SftpSupplierConfiguration {
 					.filter(fileListFilter);
 		}
 
+		@Bean
+		@ConditionalOnProperty(prefix = "sftp.supplier", value = "rename-remote-files-to")
+		public SftpOutboundGatewaySpec renameRemoteFileHandler(SftpSupplierFactoryConfiguration.DelegatingFactoryWrapper delegatingFactoryWrapper, SftpSupplierProperties sftpSupplierProperties) {
+			return Sftp.outboundGateway(delegatingFactoryWrapper.getFactory(), AbstractRemoteFileOutboundGateway.Command.MV.getCommand(), String.format("headers.get('%s') + '%s' + headers.get('%s')", FileHeaders.REMOTE_DIRECTORY, sftpSupplierProperties.getRemoteFileSeparator(), FileHeaders.REMOTE_FILE))
+					.renameExpression(sftpSupplierProperties.getRenameRemoteFilesTo());
+		}
 	}
 
 	/*

supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierProperties.java

@@ -49,7 +49,6 @@ import org.springframework.validation.annotation.Validated;
 @ConfigurationProperties("sftp.supplier")
 @Validated
 public class SftpSupplierProperties {
-
 	/**
 	 * Session factory properties.
 	 */
@@ -75,6 +74,11 @@ public class SftpSupplierProperties {
 	 */
 	private boolean deleteRemoteFiles = false;
 
+	/**
+	 * A SpEL expression resolving to the new name remote files must be renamed to after successful transfer.
+	 */
+	private Expression renameRemoteFilesTo = null;
+
 	/**
 	 * The local directory to use for file transfers.
 	 */
@@ -187,6 +191,14 @@ public class SftpSupplierProperties {
 		this.deleteRemoteFiles = deleteRemoteFiles;
 	}
 
+	public Expression getRenameRemoteFilesTo() {
+		return renameRemoteFilesTo;
+	}
+
+	public void setRenameRemoteFilesTo(Expression renameRemoteFilesTo) {
+		this.renameRemoteFilesTo = renameRemoteFilesTo;
+	}
+
 	@NotNull
 	public File getLocalDir() {
 		return localDir;
@@ -309,6 +321,11 @@ public class SftpSupplierProperties {
 		this.sortBy = sortBy;
 	}
 
+	@AssertTrue(message = "deleteRemoteFiles must be 'false' when renameRemoteFilesTo is set")
+	public boolean isRenameRemoteFilesValid() {
+		return renameRemoteFilesTo == null || !deleteRemoteFiles;
+	}
+
 	public static class Factory {
 
 		/**

supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java

@@ -23,11 +23,13 @@ import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -35,6 +37,7 @@ import reactor.core.publisher.Flux;
 import reactor.test.StepVerifier;
 
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.test.context.assertj.AssertableApplicationContext;
 import org.springframework.boot.test.context.runner.ApplicationContextRunner;
 import org.springframework.cloud.fn.test.support.sftp.SftpTestSupport;
 import org.springframework.http.MediaType;
@@ -235,6 +238,46 @@ public class SftpSupplierApplicationTests extends SftpTestSupport {
 
 	}
 
+	@Test
+	void renameRemoteFilesStream() {
+		defaultApplicationContextRunner
+				.withPropertyValues(
+						"sftp.supplier.stream=true",
+						"sftp.supplier.delete-remote-files=false",
+						"sftp.supplier.rename-remote-files-to='/sftpTarget/' + headers.file_remoteFile")
+				.run(this::doTestRenameRemoteFiles);
+	}
+
+	@Test
+	void renameRemoteFiles() {
+		defaultApplicationContextRunner
+				.withPropertyValues(
+						"sftp.supplier.stream=false",
+						"sftp.supplier.delete-remote-files=false",
+						"sftp.supplier.rename-remote-files-to='/sftpTarget/' + headers.file_remoteFile")
+				.run(this::doTestRenameRemoteFiles);
+	}
+
+	private void doTestRenameRemoteFiles(AssertableApplicationContext context) {
+		Supplier<Flux<Message<byte[]>>> sftpSupplier = context.getBean("sftpSupplier",
+				Supplier.class);
+
+		final Set<String> expectedTargetFiles = Arrays.stream(getSourceRemoteDirectory().list())
+				.collect(Collectors.toSet());
+
+		StepVerifier.create(sftpSupplier.get())
+				.expectNextMatches(message -> message.getPayload().length > 0)
+				.expectNextMatches(message -> message.getPayload().length > 0)
+				.thenCancel()
+				.verify(Duration.ofSeconds(30));
+		await().atMost(Duration.ofSeconds(30))
+				.until(() ->
+						expectedTargetFiles.equals(
+								Arrays.stream(getTargetRemoteDirectory().list())
+								.collect(Collectors.toSet()))
+				);
+	}
+
 	@Test
 	public void streamSourceFilesInLineMode() {
 		defaultApplicationContextRunner