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

GH-431: Support sub-dirs in S3 Source

Browse Source 
Fixes https://github.com/spring-cloud/stream-applications/issues/431

* Use `WatchService` for local file `MessageSource` in `AwsS3SupplierConfiguration`
for `S3InboundFileSynchronizingMessageSource`.
This way we are able to scan sub-dirs in the local file store and skip those
sub-dirs as entities.

The `S3InboundFileSynchronizingMessageSource` is able to properly create a local sub-dir
according to the complex key of S3 object, for example if we have a remove entry as `myDir/myFile.txt`,
so in the `/local` dir it will be stored as `/local/myDir/myFile.txt`.
The mentioned `WatchService` can react to entries of entire file tree.
This commit is contained in:
abilan
2023-02-21 14:20:34 -05:00
committed by Corneil du Plessis
parent a0cc610a06
commit f6939fcc25
5 changed files with 29 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2022 the original author or authors.
* Copyright 2016-2023 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.
@@ -16,7 +16,6 @@
package org.springframework.cloud.fn.supplier.s3;
import java.io.File;
import java.util.List;
import java.util.function.Predicate;
import java.util.function.Supplier;
@@ -121,9 +120,11 @@ public class AwsS3SupplierConfiguration {
}
@Bean
public Publisher<Message<Object>> s3SupplierFlow(MessageSource<?> s3MessageSource) {
public Publisher<Message<Object>> s3SupplierFlow(S3InboundFileSynchronizingMessageSource s3MessageSource) {
return FileUtils.enhanceFlowForReadingMode(
IntegrationFlow.from(IntegrationReactiveUtils.messageSourceToFlux(s3MessageSource)),
IntegrationFlow.from(
IntegrationReactiveUtils.messageSourceToFlux(s3MessageSource)
.doOnSubscribe((s) -> s3MessageSource.start())),
fileConsumerProperties)
.toReactivePublisher(true);
}
@@ -144,13 +145,15 @@ public class AwsS3SupplierConfiguration {
}
@Bean
public MessageSource<File> s3MessageSource(S3InboundFileSynchronizer s3InboundFileSynchronizer,
public S3InboundFileSynchronizingMessageSource s3MessageSource(
S3InboundFileSynchronizer s3InboundFileSynchronizer,
@Nullable ComponentCustomizer<S3InboundFileSynchronizingMessageSource> s3MessageSourceCustomizer) {
S3InboundFileSynchronizingMessageSource s3MessageSource = new S3InboundFileSynchronizingMessageSource(
s3InboundFileSynchronizer);
s3MessageSource.setLocalDirectory(this.awsS3SupplierProperties.getLocalDir());
s3MessageSource.setAutoCreateLocalDirectory(this.awsS3SupplierProperties.isAutoCreateLocalDir());
s3MessageSource.setUseWatchService(true);
if (s3MessageSourceCustomizer != null) {
s3MessageSourceCustomizer.customize(s3MessageSource);

9
functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AbstractAwsS3SupplierMockTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2021 the original author or authors.
* Copyright 2016-2023 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.
@@ -39,9 +39,14 @@ import reactor.core.publisher.Flux;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.cloud.fn.common.config.ComponentCustomizer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Primary;
import org.springframework.integration.aws.inbound.S3InboundFileSynchronizingMessageSource;
import org.springframework.integration.dsl.StandardIntegrationFlow;
import org.springframework.integration.file.RecursiveDirectoryScanner;
import org.springframework.integration.file.filters.AbstractFileListFilter;
import org.springframework.integration.file.filters.FileListFilter;
import org.springframework.integration.test.context.SpringIntegrationTest;
import org.springframework.messaging.Message;
import org.springframework.test.annotation.DirtiesContext;
@@ -104,7 +109,7 @@ public abstract class AbstractAwsS3SupplierMockTests {
for (File file : f.listFiles()) {
S3Object s3Object = new S3Object();
s3Object.setBucketName(S3_BUCKET);
s3Object.setKey(file.getName());
s3Object.setKey("subdir/" + file.getName());
s3Object.setObjectContent(new FileInputStream(file));
S3_OBJECTS.add(s3Object);
}

6
functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3FilesTransferredTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-2023 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.
@@ -39,12 +39,12 @@ public class AmazonS3FilesTransferredTests extends AbstractAwsS3SupplierMockTest
StepVerifier.create(messageFlux)
.assertNext((message) -> {
assertThat(new File(message.getPayload().toString().replaceAll("\"", "")))
.isEqualTo(new File(this.awsS3SupplierProperties.getLocalDir() + File.separator + "1.test"));
.isEqualTo(new File(this.awsS3SupplierProperties.getLocalDir() + File.separator + "subdir" + File.separator + "1.test"));
}
)
.assertNext((message) -> {
assertThat(new File(message.getPayload().toString().replaceAll("\"", "")))
.isEqualTo(new File(this.awsS3SupplierProperties.getLocalDir() + File.separator + "2.test"));
.isEqualTo(new File(this.awsS3SupplierProperties.getLocalDir() + File.separator + "subdir" + File.separator + "2.test"));
})
.thenCancel()
.verifyLater();

14
functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3LinesTransferredTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2020 the original author or authors.
* Copyright 2016-2023 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.
@@ -31,7 +31,7 @@ import static org.assertj.core.api.Assertions.assertThat;
@TestPropertySource(properties = {
"file.consumer.mode=lines",
"s3.supplier.filenamePattern=otherFile",
"s3.supplier.filenamePattern=*/otherFile",
"file.consumer.with-markers=false" })
public class AmazonS3LinesTransferredTests extends AbstractAwsS3SupplierMockTests {
@@ -41,10 +41,10 @@ public class AmazonS3LinesTransferredTests extends AbstractAwsS3SupplierMockTest
StepVerifier stepVerifier =
StepVerifier.create(messageFlux)
.assertNext((message) -> {
assertThat(message.getPayload().toString()).isEqualTo("Other");
assertThat(message.getHeaders().containsKey(FileHeaders.ORIGINAL_FILE)).isTrue();
assertThat(message.getHeaders().containsValue(
new File(this.awsS3SupplierProperties.getLocalDir(), "otherFile"))).isTrue();
assertThat(message.getPayload()).isEqualTo("Other");
assertThat(message.getHeaders()).containsKey(FileHeaders.ORIGINAL_FILE);
assertThat(message.getHeaders()).containsValue(
new File(this.awsS3SupplierProperties.getLocalDir(), "subdir/otherFile"));
}
)
.assertNext((message) -> {
@@ -55,6 +55,6 @@ public class AmazonS3LinesTransferredTests extends AbstractAwsS3SupplierMockTest
standardIntegrationFlow.start();
stepVerifier.verify(Duration.ofSeconds(10));
assertThat(this.awsS3SupplierProperties.getLocalDir().list().length).isEqualTo(1);
assertThat(this.awsS3SupplierProperties.getLocalDir().list()).hasSize(1);
}
}

8
functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3ListOnlyTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2020-2020 the original author or authors.
* Copyright 2020-2023 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.
@@ -38,9 +38,9 @@ public class AmazonS3ListOnlyTests extends AbstractAwsS3SupplierMockTests {
public void test() {
final Flux<Message<?>> messageFlux = s3Supplier.get();
final HashSet<String> keys = new HashSet<>();
keys.add("1.test");
keys.add("2.test");
keys.add("otherFile");
keys.add("subdir/1.test");
keys.add("subdir/2.test");
keys.add("subdir/otherFile");
StepVerifier stepVerifier = StepVerifier.create(messageFlux)
.assertNext(message -> {
S3ObjectSummary summary = (S3ObjectSummary) message.getPayload();