SpringCloud/stream-applications
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-156: Fix NPE in the RemoteFileDeletingAdvice

Browse Source 
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<Boolean> 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
This commit is contained in:
Artem Bilan
2021-04-08 16:23:20 -04:00
committed by David Turanski
parent 9faa2a6a38
commit 04beb9ecaf
4 changed files with 42 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
applications/source/sftp-source/src/test/java/org/springframework/cloud/stream/app/source/sftp/SftpSourceTests.java
View File

@@ -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);

15
functions/common/file-common/src/main/java/org/springframework/cloud/fn/common/file/remote/RemoteFileDeletingAdvice.java
View File

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

47
functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java
View File

@@ -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<Boolean> subscriptionBarrier() {
return MonoProcessor.create();
}
@Bean
public Supplier<Flux<? extends Message<?>>> sftpSupplier(MessageSource<?> sftpMessageSource,
@Nullable Publisher<Message<Object>> sftpReadingFlow,
MonoProcessor<Boolean> subscriptionBarrier,
SftpSupplierProperties sftpSupplierProperties) {
Flux<? extends Message<?>> 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<? extends Message<?>> 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<Message<Object>> 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<Message<Object>> 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<Object>) 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<LsEntry> stream = Stream.of(this.sessionFactory.getSession().list(this.remoteDirectory))
.map(x -> (LsEntry) x)

1
functions/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java
View File

@@ -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")