SpringCloud/spring-functions-catalog
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Defer subscription to the S3 until downstream one

Browse Source 
It turns out that endpoint for S3 Source is starting to
consume the `Flux` earlier than an actual subscription
happens for the whole flow downstream

* Introduce a `MonoProcessor<Subscription> downstreamSubscription`
to fulfill from the downstream subscription and handle from the
`delaySubscription()` on the `Flux` for source data
This commit is contained in:
Artem Bilan
2020-09-25 13:27:13 -04:00
parent 730992c050
commit fd3253dc38
1 changed files with 16 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java
View File

@@ -24,8 +24,10 @@ import java.util.regex.Pattern;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.model.ListObjectsRequest;
import com.amazonaws.services.s3.model.S3ObjectSummary;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.MonoProcessor;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscription;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
@@ -86,9 +88,11 @@ public abstract class AwsS3SupplierConfiguration {
@ConditionalOnProperty(prefix = "s3.supplier", name = "list-only", havingValue = "false", matchIfMissing = true)
static class SynchronizingConfiguration extends AwsS3SupplierConfiguration {
private final MonoProcessor<Subscription> downstreamSubscription = MonoProcessor.create();
@Bean
public Supplier<Flux<Message<?>>> s3Supplier(Publisher<Message<Object>> s3SupplierFlow) {
return () -> Flux.from(s3SupplierFlow);
public Supplier<Flux<Message<?>>> s3Supplier(Publisher<Message<?>> s3SupplierFlow) {
return () -> Flux.from(s3SupplierFlow).doOnSubscribe(this.downstreamSubscription::onNext);
}
@Bean
@@ -118,8 +122,11 @@ public abstract class AwsS3SupplierConfiguration {
@Bean
public Publisher<Message<Object>> s3SupplierFlow(MessageSource<?> s3MessageSource) {
return FileUtils.enhanceFlowForReadingMode(IntegrationFlows
.from(IntegrationReactiveUtils.messageSourceToFlux(s3MessageSource)), fileConsumerProperties)
return FileUtils.enhanceFlowForReadingMode(
IntegrationFlows.from(
IntegrationReactiveUtils.messageSourceToFlux(s3MessageSource)
.delaySubscription(this.downstreamSubscription)),
fileConsumerProperties)
.toReactivePublisher();
}
@@ -146,11 +153,13 @@ public abstract class AwsS3SupplierConfiguration {
s3MessageSource.setAutoCreateLocalDirectory(this.awsS3SupplierProperties.isAutoCreateLocalDir());
return s3MessageSource;
}
}
@Configuration
@ConditionalOnProperty(prefix = "s3.supplier", name = "list-only", havingValue = "true")
static class ListOnlyConfiguration extends AwsS3SupplierConfiguration {
ListOnlyConfiguration(AwsS3SupplierProperties awsS3SupplierProperties,
FileConsumerProperties fileConsumerProperties,
AmazonS3 amazonS3,
@@ -209,5 +218,7 @@ public abstract class AwsS3SupplierConfiguration {
(MessageSource<Iterable<S3ObjectSummary>>) () -> new GenericMessage<>(
amazonS3.listObjects(listObjectsRequest).getObjectSummaries()));
}
}
}