From 04beb9ecaf0b89542a1e7efbf43e848dfe6c2e6d Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 8 Apr 2021 16:23:20 -0400 Subject: [PATCH] GH-156: Fix NPE in the `RemoteFileDeletingAdvice` Fixes https://github.com/spring-cloud/stream-applications/issues/156 The `MessageSource.receive()` may produce `null`: The `MessageSourceMutator` impl must honor such an input * Add `MonoProcessor subscriptionBarrier` to delay subscription to the source `Flux` until subscription happens to the supplier's flux. This way we don't have unexpected interaction with the source when there are regular endpoints in the flow in between --- .../app/source/sftp/SftpSourceTests.java | 3 +- .../file/remote/RemoteFileDeletingAdvice.java | 15 ++++-- .../sftp/SftpSupplierConfiguration.java | 47 ++++++++++++------- .../sftp/SftpSupplierApplicationTests.java | 1 + 4 files changed, 42 insertions(+), 24 deletions(-) diff --git a/applications/source/sftp-source/src/test/java/org/springframework/cloud/stream/app/source/sftp/SftpSourceTests.java b/applications/source/sftp-source/src/test/java/org/springframework/cloud/stream/app/source/sftp/SftpSourceTests.java index e5092c34..d7ad9982 100644 --- a/applications/source/sftp-source/src/test/java/org/springframework/cloud/stream/app/source/sftp/SftpSourceTests.java +++ b/applications/source/sftp-source/src/test/java/org/springframework/cloud/stream/app/source/sftp/SftpSourceTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2020 the original author or authors. + * 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. @@ -47,6 +47,7 @@ public class SftpSourceTests extends SftpTestSupport { "sftp.supplier.factory.port =${sftp.factory.port}", "sftp.supplier.factory.allowUnknownKeys=true", "sftp.supplier.remoteDir=sftpSource", + "sftp.supplier.delete-remote-files=true", "spring.cloud.function.definition=sftpSupplier") .run(context -> { OutputDestination output = context.getBean(OutputDestination.class); diff --git a/functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/remote/RemoteFileDeletingAdvice.java b/functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/remote/RemoteFileDeletingAdvice.java index ec25de02..67fd7f81 100644 --- a/functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/remote/RemoteFileDeletingAdvice.java +++ b/functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/remote/RemoteFileDeletingAdvice.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2020 the original author or authors. + * 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. @@ -20,12 +20,14 @@ 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 deletes a remote file on success. * * @author David Turanski + * @author Artem Bilan * */ public class RemoteFileDeletingAdvice implements MessageSourceMutator { @@ -45,11 +47,14 @@ public class RemoteFileDeletingAdvice implements MessageSourceMutator { this.remoteFileSeparator = remoteFileSeparator; } + @Nullable @Override - public Message afterReceive(Message result, MessageSource source) { - String remoteDir = (String) result.getHeaders().get(FileHeaders.REMOTE_DIRECTORY); - String remoteFile = (String) result.getHeaders().get(FileHeaders.REMOTE_FILE); - this.template.remove(remoteDir + this.remoteFileSeparator + remoteFile); + 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); + this.template.remove(remoteDir + this.remoteFileSeparator + remoteFile); + } return result; } } diff --git a/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java b/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java index ae0d8915..0b0f600f 100644 --- a/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java +++ b/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java @@ -27,6 +27,7 @@ import java.util.stream.Stream; import com.jcraft.jsch.ChannelSftp.LsEntry; import org.reactivestreams.Publisher; import reactor.core.publisher.Flux; +import reactor.core.publisher.MonoProcessor; import org.springframework.aop.framework.ProxyFactoryBean; import org.springframework.aop.support.NameMatchMethodPointcutAdvisor; @@ -37,7 +38,6 @@ 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.context.Lifecycle; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; @@ -82,28 +82,30 @@ import org.springframework.util.StringUtils; */ @Configuration -@EnableConfigurationProperties({SftpSupplierProperties.class, FileConsumerProperties.class}) -@Import({SftpSupplierFactoryConfiguration.class}) +@EnableConfigurationProperties({ SftpSupplierProperties.class, FileConsumerProperties.class }) +@Import({ SftpSupplierFactoryConfiguration.class }) public class SftpSupplierConfiguration { private static final String METADATA_STORE_PREFIX = "sftpSource/"; private static final String FILE_MODIFIED_TIME_HEADER = "FILE_MODIFIED_TIME"; + @Bean + public MonoProcessor subscriptionBarrier() { + return MonoProcessor.create(); + } + @Bean public Supplier>> sftpSupplier(MessageSource sftpMessageSource, @Nullable Publisher> sftpReadingFlow, + MonoProcessor subscriptionBarrier, SftpSupplierProperties sftpSupplierProperties) { Flux> flux = sftpReadingFlow == null - ? sftpMessageFlux(sftpMessageSource, sftpSupplierProperties) + ? sftpMessageFlux(sftpMessageSource, sftpSupplierProperties, subscriptionBarrier) : Flux.from(sftpReadingFlow); - return () -> flux.doOnSubscribe(s -> { - if (sftpMessageSource instanceof Lifecycle) { - ((Lifecycle) sftpMessageSource).start(); - } - }); + return () -> flux.doOnSubscribe(s -> subscriptionBarrier.onNext(true)); } @Bean @@ -153,9 +155,10 @@ public class SftpSupplierConfiguration { * Create a Flux from a MessageSource that will be used by the supplier. */ private Flux> sftpMessageFlux(MessageSource sftpMessageSource, - SftpSupplierProperties sftpSupplierProperties) { + SftpSupplierProperties sftpSupplierProperties, MonoProcessor subscriptionBarrier) { return IntegrationReactiveUtils.messageSourceToFlux(sftpMessageSource) + .delaySubscription(subscriptionBarrier) .subscriberContext(context -> context.put(IntegrationReactiveUtils.DELAY_WHEN_EMPTY_KEY, sftpSupplierProperties.getDelayWhenEmpty())); @@ -196,14 +199,16 @@ public class SftpSupplierConfiguration { @Bean public Publisher> sftpReadingFlow( MessageSource sftpMessageSource, + MonoProcessor subscriptionBarrier, SftpSupplierProperties sftpSupplierProperties, FileConsumerProperties fileConsumerProperties) { return FileUtils.enhanceStreamFlowForReadingMode(IntegrationFlows - .from(IntegrationReactiveUtils.messageSourceToFlux(sftpMessageSource) - .subscriberContext( - context -> (context.put(IntegrationReactiveUtils.DELAY_WHEN_EMPTY_KEY, - sftpSupplierProperties.getDelayWhenEmpty())))), + .from(IntegrationReactiveUtils.messageSourceToFlux(sftpMessageSource) + .delaySubscription(subscriptionBarrier) + .subscriberContext( + context -> (context.put(IntegrationReactiveUtils.DELAY_WHEN_EMPTY_KEY, + sftpSupplierProperties.getDelayWhenEmpty())))), fileConsumerProperties) .toReactivePublisher(); } @@ -233,10 +238,16 @@ public class SftpSupplierConfiguration { @ConditionalOnExpression("environment['file.consumer.mode']!='ref' && environment['sftp.supplier.list-only']!='true'") public Publisher> sftpReadingFlow( MessageSource sftpMessageSource, + MonoProcessor subscriptionBarrier, + SftpSupplierProperties sftpSupplierProperties, FileConsumerProperties fileConsumerProperties) { return FileUtils.enhanceFlowForReadingMode(IntegrationFlows - .from(IntegrationReactiveUtils.messageSourceToFlux(sftpMessageSource)), + .from(IntegrationReactiveUtils.messageSourceToFlux(sftpMessageSource) + .delaySubscription(subscriptionBarrier) + .subscriberContext( + context -> (context.put(IntegrationReactiveUtils.DELAY_WHEN_EMPTY_KEY, + sftpSupplierProperties.getDelayWhenEmpty())))), fileConsumerProperties) .toReactivePublisher(); } @@ -282,7 +293,7 @@ public class SftpSupplierConfiguration { @Bean @SuppressWarnings("unchecked") public MessageSource targetMessageSource(PollableChannel listingChannel, - SftpListingMessageProducer sftpListingMessageProducer) { + SftpListingMessageProducer sftpListingMessageProducer) { return () -> { sftpListingMessageProducer.listNames(); return (Message) listingChannel.receive(); @@ -378,7 +389,7 @@ public class SftpSupplierConfiguration { private final SftpSupplierProperties.SortSpec sort; SftpListingMessageProducer(SessionFactory sessionFactory, String remoteDirectory, - String remoteFileSeparator, SftpSupplierProperties.SortSpec sort) { + String remoteFileSeparator, SftpSupplierProperties.SortSpec sort) { this.sessionFactory = sessionFactory; this.remoteDirectory = remoteDirectory; @@ -387,7 +398,7 @@ public class SftpSupplierConfiguration { } public void listNames() { - LsEntry[] entries = {}; + LsEntry[] entries = { }; try { Stream stream = Stream.of(this.sessionFactory.getSession().list(this.remoteDirectory)) .map(x -> (LsEntry) x) diff --git a/functions/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java b/functions/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java index bb4b465a..0a966682 100644 --- a/functions/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java +++ b/functions/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java @@ -243,6 +243,7 @@ public class SftpSupplierApplicationTests extends SftpTestSupport { "sftp.supplier.factory.private-key = classpath:id_rsa_pp", "sftp.supplier.factory.passphrase = secret", "sftp.supplier.factory.password = badPassword", // ensure public key was used + "sftp.supplier.delete-remote-files=true", "file.consumer.mode=lines", "file.consumer.with-markers=true", "file.consumer.markers-json=true")