diff --git a/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/remote/RemoteFileRenamingAdvice.java b/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/remote/RemoteFileRenamingAdvice.java new file mode 100644 index 00000000..06181683 --- /dev/null +++ b/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; + } +} diff --git a/supplier/sftp-supplier/pom.xml b/supplier/sftp-supplier/pom.xml index 259195a7..9b42a8c9 100644 --- a/supplier/sftp-supplier/pom.xml +++ b/supplier/sftp-supplier/pom.xml @@ -30,6 +30,11 @@ spring-integration-sftp + + org.springframework.cloud.fn + config-common + ${project.version} + org.springframework.cloud.fn file-common diff --git a/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java b/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java index 0b0f600f..a574971f 100644 --- a/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java +++ b/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>> sftpSupplier(MessageSource sftpMessageSource, - @Nullable Publisher> sftpReadingFlow, - MonoProcessor subscriptionBarrier, - SftpSupplierProperties sftpSupplierProperties) { + @Nullable Publisher> sftpReadingFlow, + MonoProcessor subscriptionBarrier, + SftpSupplierProperties sftpSupplierProperties) { Flux> 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()); + } } /* diff --git a/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierProperties.java b/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierProperties.java index e6c197de..d28c24e1 100644 --- a/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierProperties.java +++ b/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 { /** diff --git a/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java b/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java index 0a966682..5cfd6ffb 100644 --- a/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java +++ b/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>> sftpSupplier = context.getBean("sftpSupplier", + Supplier.class); + + final Set 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