From f6939fcc2551d64d1bd4815a7dedf5e6845db263 Mon Sep 17 00:00:00 2001 From: abilan Date: Tue, 21 Feb 2023 14:20:34 -0500 Subject: [PATCH] GH-431: Support sub-dirs in S3 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. --- .../fn/supplier/s3/AwsS3SupplierConfiguration.java | 13 ++++++++----- .../s3/AbstractAwsS3SupplierMockTests.java | 9 +++++++-- .../supplier/s3/AmazonS3FilesTransferredTests.java | 6 +++--- .../supplier/s3/AmazonS3LinesTransferredTests.java | 14 +++++++------- .../fn/supplier/s3/AmazonS3ListOnlyTests.java | 8 ++++---- 5 files changed, 29 insertions(+), 21 deletions(-) diff --git a/functions/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java b/functions/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java index 97dbcf40..bdfc585d 100644 --- a/functions/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java +++ b/functions/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java @@ -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> s3SupplierFlow(MessageSource s3MessageSource) { + public Publisher> 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 s3MessageSource(S3InboundFileSynchronizer s3InboundFileSynchronizer, + public S3InboundFileSynchronizingMessageSource s3MessageSource( + S3InboundFileSynchronizer s3InboundFileSynchronizer, @Nullable ComponentCustomizer 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); diff --git a/functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AbstractAwsS3SupplierMockTests.java b/functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AbstractAwsS3SupplierMockTests.java index 947558df..6516cc8c 100644 --- a/functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AbstractAwsS3SupplierMockTests.java +++ b/functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AbstractAwsS3SupplierMockTests.java @@ -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); } diff --git a/functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3FilesTransferredTests.java b/functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3FilesTransferredTests.java index 53722fe3..f5e4b197 100644 --- a/functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3FilesTransferredTests.java +++ b/functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3FilesTransferredTests.java @@ -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(); diff --git a/functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3LinesTransferredTests.java b/functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3LinesTransferredTests.java index 490e5fda..c6305d67 100644 --- a/functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3LinesTransferredTests.java +++ b/functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3LinesTransferredTests.java @@ -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); } } diff --git a/functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3ListOnlyTests.java b/functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3ListOnlyTests.java index 47ffa682..1f4adc69 100644 --- a/functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3ListOnlyTests.java +++ b/functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3ListOnlyTests.java @@ -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> messageFlux = s3Supplier.get(); final HashSet 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();