diff --git a/common/metadata-store-common/README.adoc b/common/metadata-store-common/README.adoc index d1a32b96..1adaa540 100644 --- a/common/metadata-store-common/README.adoc +++ b/common/metadata-store-common/README.adoc @@ -5,7 +5,7 @@ See Spring Integration "`https://docs.spring.io/spring-integration/docs/5.0.6.RE In addition to the standard Spring Boot configuration properties this module exposes a `MetadataStoreProperties` with the `metadata.store` prefix. -To auto-configure particular `MetadataStore` you just need to bring respective dependencies into the target app starter: +To auto-configure particular `MetadataStore` you need to set `metadata.store.type` and include the respective dependencies into the target app starter: ==== Redis diff --git a/common/metadata-store-common/pom.xml b/common/metadata-store-common/pom.xml index 7bd5cd07..dddcd4c5 100644 --- a/common/metadata-store-common/pom.xml +++ b/common/metadata-store-common/pom.xml @@ -31,6 +31,12 @@ spring-boot-starter + + org.springframework.boot + spring-boot-configuration-processor + provided + + org.springframework.boot spring-boot-starter-test diff --git a/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java b/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java index e185d9e3..d3e58ee3 100644 --- a/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java +++ b/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java @@ -27,9 +27,9 @@ import org.apache.geode.cache.GemFireCache; import org.apache.geode.cache.Region; import org.springframework.beans.factory.ObjectProvider; -import org.springframework.boot.autoconfigure.condition.ConditionalOnBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnClass; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.aws.core.region.RegionProvider; import org.springframework.context.annotation.Bean; @@ -52,6 +52,7 @@ import org.springframework.jdbc.core.JdbcTemplate; /** * @author Artem Bilan + * @author David Turanski * @since 2.0.2 */ @Configuration @@ -60,47 +61,46 @@ import org.springframework.jdbc.core.JdbcTemplate; public class MetadataStoreAutoConfiguration { @Bean + @ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "memory", matchIfMissing = true) @ConditionalOnMissingBean public ConcurrentMetadataStore simpleMetadataStore() { return new SimpleMetadataStore(); } - @ConditionalOnClass(RedisMetadataStore.class) - @ConditionalOnBean(RedisTemplate.class) + @ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "redis") static class Redis { @Bean @ConditionalOnMissingBean public ConcurrentMetadataStore redisMetadataStore(RedisTemplate redisTemplate, - MetadataStoreProperties metadataStoreProperties) { + MetadataStoreProperties metadataStoreProperties) { return new RedisMetadataStore(redisTemplate, metadataStoreProperties.getRedis().getKey()); } } - @ConditionalOnClass(MongoDbMetadataStore.class) - @ConditionalOnBean(MongoTemplate.class) + @ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "mongodb") static class Mongo { @Bean @ConditionalOnMissingBean public ConcurrentMetadataStore mongoDbMetadataStore(MongoTemplate mongoTemplate, - MetadataStoreProperties metadataStoreProperties) { + MetadataStoreProperties metadataStoreProperties) { return new MongoDbMetadataStore(mongoTemplate, metadataStoreProperties.getMongoDb().getCollection()); } } - @ConditionalOnClass(GemfireMetadataStore.class) + @ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "gemfire") @Import(ClientCacheAutoConfiguration.class) static class Gemfire { @Bean @ConditionalOnMissingBean public ClientRegionFactoryBean gemfireRegion(GemFireCache cache, - MetadataStoreProperties metadataStoreProperties) { + MetadataStoreProperties metadataStoreProperties) { ClientRegionFactoryBean clientRegionFactoryBean = new ClientRegionFactoryBean<>(); clientRegionFactoryBean.setCache(cache); @@ -111,7 +111,7 @@ public class MetadataStoreAutoConfiguration { @Bean @ConditionalOnMissingBean public ConcurrentMetadataStore gemfireMetadataStore(Region region, - ObjectProvider metadataStoreListenerObjectProvider) { + ObjectProvider metadataStoreListenerObjectProvider) { @SuppressWarnings("unchecked") GemfireMetadataStore gemfireMetadataStore = new GemfireMetadataStore((Region) region); @@ -122,7 +122,7 @@ public class MetadataStoreAutoConfiguration { } - @ConditionalOnClass(HazelcastMetadataStore.class) + @ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "hazelcast") static class Hazelcast { @Bean @@ -134,7 +134,7 @@ public class MetadataStoreAutoConfiguration { @Bean @ConditionalOnMissingBean public ConcurrentMetadataStore hazelcastMetadataStore(HazelcastInstance hazelcastInstance, - ObjectProvider metadataStoreListenerObjectProvider) { + ObjectProvider metadataStoreListenerObjectProvider) { HazelcastMetadataStore hazelcastMetadataStore = new HazelcastMetadataStore(hazelcastInstance); metadataStoreListenerObjectProvider.ifAvailable(hazelcastMetadataStore::addListener); @@ -143,7 +143,7 @@ public class MetadataStoreAutoConfiguration { } - @ConditionalOnClass({ZookeeperMetadataStore.class, CuratorFramework.class}) + @ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "zookeeper") static class Zookeeper { @Bean(initMethod = "start") @@ -157,8 +157,8 @@ public class MetadataStoreAutoConfiguration { @Bean @ConditionalOnMissingBean public ConcurrentMetadataStore zookeeperMetadataStore(CuratorFramework curatorFramework, - MetadataStoreProperties metadataStoreProperties, - ObjectProvider metadataStoreListenerObjectProvider) { + MetadataStoreProperties metadataStoreProperties, + ObjectProvider metadataStoreListenerObjectProvider) { MetadataStoreProperties.Zookeeper zookeeperProperties = metadataStoreProperties.getZookeeper(); ZookeeperMetadataStore zookeeperMetadataStore = new ZookeeperMetadataStore(curatorFramework); @@ -170,14 +170,13 @@ public class MetadataStoreAutoConfiguration { } - @ConditionalOnClass(DynamoDbMetadataStore.class) - @ConditionalOnBean({AWSCredentialsProvider.class, RegionProvider.class}) + @ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "dynamodb") static class DynamoDb { @Bean @ConditionalOnMissingBean public AmazonDynamoDBAsync dynamoDB(AWSCredentialsProvider awsCredentialsProvider, - RegionProvider regionProvider) { + RegionProvider regionProvider) { return AmazonDynamoDBAsyncClientBuilder.standard() .withCredentials(awsCredentialsProvider) @@ -190,12 +189,12 @@ public class MetadataStoreAutoConfiguration { @Bean @ConditionalOnMissingBean public ConcurrentMetadataStore dynamoDbMetadataStore(AmazonDynamoDBAsync dynamoDB, - MetadataStoreProperties metadataStoreProperties) { + MetadataStoreProperties metadataStoreProperties) { MetadataStoreProperties.DynamoDb dynamoDbProperties = metadataStoreProperties.getDynamoDb(); - DynamoDbMetadataStore dynamoDbMetadataStore = - new DynamoDbMetadataStore(dynamoDB, dynamoDbProperties.getTable()); + DynamoDbMetadataStore dynamoDbMetadataStore = new DynamoDbMetadataStore(dynamoDB, + dynamoDbProperties.getTable()); dynamoDbMetadataStore.setReadCapacity(dynamoDbProperties.getReadCapacity()); dynamoDbMetadataStore.setWriteCapacity(dynamoDbProperties.getWriteCapacity()); @@ -210,14 +209,12 @@ public class MetadataStoreAutoConfiguration { } - @ConditionalOnClass(JdbcMetadataStore.class) - @ConditionalOnBean(JdbcTemplate.class) + @ConditionalOnProperty(prefix = "metadata.store", name = "type", havingValue = "jdbc") static class Jdbc { - @Bean @ConditionalOnMissingBean public ConcurrentMetadataStore jdbcMetadataStore(JdbcTemplate jdbcTemplate, - MetadataStoreProperties metadataStoreProperties) { + MetadataStoreProperties metadataStoreProperties) { MetadataStoreProperties.Jdbc jdbcProperties = metadataStoreProperties.getJdbc(); diff --git a/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java b/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java index 39b898b1..ac030cd7 100644 --- a/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java +++ b/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java @@ -27,10 +27,27 @@ import org.springframework.integration.redis.metadata.RedisMetadataStore; /** * @author Artem Bilan + * @author David Turanski * @since 2.0.2 */ @ConfigurationProperties("metadata.store") public class MetadataStoreProperties { + enum StoreType { + mongodb, + gemfire, + redis, + dynamodb, + jdbc, + zookeeper, + hazelcast, + memory + } + + /** + * Indicates the type of metadata store to configure (default is 'memory'). + * You must include the corresponding Spring Integration dependency to use a persistent store. + */ + private StoreType type = StoreType.memory; private final Mongo mongoDb = new Mongo(); @@ -44,6 +61,14 @@ public class MetadataStoreProperties { private final Zookeeper zookeeper = new Zookeeper(); + public StoreType getType() { + return this.type; + } + + public void setType(StoreType type) { + this.type = type; + } + public Mongo getMongoDb() { return this.mongoDb; } diff --git a/supplier/s3-supplier/pom.xml b/supplier/s3-supplier/pom.xml index 045dbd58..54203f31 100644 --- a/supplier/s3-supplier/pom.xml +++ b/supplier/s3-supplier/pom.xml @@ -28,6 +28,11 @@ org.springframework.boot spring-boot-starter-validation + + org.springframework.cloud.fn + metadata-store-common + ${project.version} + org.springframework.boot spring-boot-configuration-processor diff --git a/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java b/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java index 3e9ca2cf..ed01b4c0 100644 --- a/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java +++ b/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java @@ -17,14 +17,17 @@ package org.springframework.cloud.fn.supplier.s3; import java.io.File; -import java.util.Arrays; +import java.util.function.Predicate; import java.util.function.Supplier; +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 org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.aws.core.env.ResourceIdResolver; import org.springframework.cloud.fn.common.file.FileConsumerProperties; @@ -37,80 +40,174 @@ import org.springframework.integration.aws.support.S3SessionFactory; import org.springframework.integration.aws.support.filters.S3PersistentAcceptOnceFileListFilter; import org.springframework.integration.aws.support.filters.S3RegexPatternFileListFilter; import org.springframework.integration.aws.support.filters.S3SimplePatternFileListFilter; +import org.springframework.integration.core.GenericSelector; import org.springframework.integration.core.MessageSource; import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.endpoint.ReactiveMessageSourceProducer; import org.springframework.integration.file.filters.ChainFileListFilter; -import org.springframework.integration.file.filters.FileListFilter; -import org.springframework.integration.metadata.SimpleMetadataStore; +import org.springframework.integration.metadata.ConcurrentMetadataStore; import org.springframework.integration.util.IntegrationReactiveUtils; import org.springframework.messaging.Message; +import org.springframework.messaging.support.GenericMessage; import org.springframework.util.StringUtils; /** * @author Artem Bilan + * @author David Turanski */ @Configuration -@EnableConfigurationProperties({AwsS3SupplierProperties.class, FileConsumerProperties.class}) -public class AwsS3SupplierConfiguration { +@EnableConfigurationProperties({ AwsS3SupplierProperties.class, FileConsumerProperties.class }) +public abstract class AwsS3SupplierConfiguration { - private final AwsS3SupplierProperties awsS3SupplierProperties; - private final FileConsumerProperties fileConsumerProperties; - private final AmazonS3 amazonS3; - private final ResourceIdResolver resourceIdResolver; + protected static final String METADATA_STORE_PREFIX = "s3-metadata-"; + + protected final AwsS3SupplierProperties awsS3SupplierProperties; + + protected final FileConsumerProperties fileConsumerProperties; + + protected final AmazonS3 amazonS3; + + protected final ResourceIdResolver resourceIdResolver; + + protected final ConcurrentMetadataStore metadataStore; public AwsS3SupplierConfiguration(AwsS3SupplierProperties awsS3SupplierProperties, - FileConsumerProperties fileConsumerProperties, - AmazonS3 amazonS3, - ResourceIdResolver resourceIdResolver) { + FileConsumerProperties fileConsumerProperties, + AmazonS3 amazonS3, + ResourceIdResolver resourceIdResolver, ConcurrentMetadataStore metadataStore) { this.awsS3SupplierProperties = awsS3SupplierProperties; this.fileConsumerProperties = fileConsumerProperties; this.amazonS3 = amazonS3; this.resourceIdResolver = resourceIdResolver; + this.metadataStore = metadataStore; } - @Bean - public S3InboundFileSynchronizer s3InboundFileSynchronizer() { - S3SessionFactory s3SessionFactory = new S3SessionFactory(this.amazonS3, this.resourceIdResolver); - S3InboundFileSynchronizer synchronizer = new S3InboundFileSynchronizer(s3SessionFactory); - synchronizer.setDeleteRemoteFiles(this.awsS3SupplierProperties.isDeleteRemoteFiles()); - synchronizer.setPreserveTimestamp(this.awsS3SupplierProperties.isPreserveTimestamp()); - String remoteDir = this.awsS3SupplierProperties.getRemoteDir(); - synchronizer.setRemoteDirectory(remoteDir); - synchronizer.setRemoteFileSeparator(this.awsS3SupplierProperties.getRemoteFileSeparator()); - synchronizer.setTemporaryFileSuffix(this.awsS3SupplierProperties.getTmpFileSuffix()); + @Configuration + @ConditionalOnProperty(prefix = "s3.supplier", name = "list-only", havingValue = "false", matchIfMissing = true) + static class SynchronizingConfiguration extends AwsS3SupplierConfiguration { - FileListFilter filter = null; - if (StringUtils.hasText(this.awsS3SupplierProperties.getFilenamePattern())) { - filter = new S3SimplePatternFileListFilter(this.awsS3SupplierProperties.getFilenamePattern()); + @Bean + public Supplier>> s3Supplier(Publisher> s3SupplierFlow) { + return () -> Flux.from(s3SupplierFlow); } - else if (this.awsS3SupplierProperties.getFilenameRegex() != null) { - filter = new S3RegexPatternFileListFilter(this.awsS3SupplierProperties.getFilenameRegex()); + + @Bean + public ChainFileListFilter filter(ConcurrentMetadataStore metadataStore) { + ChainFileListFilter chainFilter = new ChainFileListFilter<>(); + if (StringUtils.hasText(this.awsS3SupplierProperties.getFilenamePattern())) { + chainFilter.addFilter( + new S3SimplePatternFileListFilter(this.awsS3SupplierProperties.getFilenamePattern())); + } + else if (this.awsS3SupplierProperties.getFilenameRegex() != null) { + chainFilter + .addFilter(new S3RegexPatternFileListFilter(this.awsS3SupplierProperties.getFilenameRegex())); + } + + chainFilter.addFilter(new S3PersistentAcceptOnceFileListFilter(metadataStore, METADATA_STORE_PREFIX)); + return chainFilter; } - if (filter != null) { - synchronizer.setFilter(new ChainFileListFilter<>(Arrays.asList(filter, - new S3PersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "s3-metadata-")))); + + SynchronizingConfiguration(AwsS3SupplierProperties awsS3SupplierProperties, + FileConsumerProperties fileConsumerProperties, + AmazonS3 amazonS3, + ResourceIdResolver resourceIdResolver, + ConcurrentMetadataStore concurrentMetadataStore) { + super(awsS3SupplierProperties, fileConsumerProperties, amazonS3, resourceIdResolver, + concurrentMetadataStore); + } + + @Bean + public Publisher> s3SupplierFlow(MessageSource s3MessageSource) { + return FileUtils.enhanceFlowForReadingMode(IntegrationFlows + .from(IntegrationReactiveUtils.messageSourceToFlux(s3MessageSource)), fileConsumerProperties) + .toReactivePublisher(); + } + + @Bean + public S3InboundFileSynchronizer s3InboundFileSynchronizer(ChainFileListFilter filter) { + S3SessionFactory s3SessionFactory = new S3SessionFactory(this.amazonS3, this.resourceIdResolver); + S3InboundFileSynchronizer synchronizer = new S3InboundFileSynchronizer(s3SessionFactory); + synchronizer.setDeleteRemoteFiles(this.awsS3SupplierProperties.isDeleteRemoteFiles()); + synchronizer.setPreserveTimestamp(this.awsS3SupplierProperties.isPreserveTimestamp()); + String remoteDir = this.awsS3SupplierProperties.getRemoteDir(); + synchronizer.setRemoteDirectory(remoteDir); + synchronizer.setRemoteFileSeparator(this.awsS3SupplierProperties.getRemoteFileSeparator()); + synchronizer.setTemporaryFileSuffix(this.awsS3SupplierProperties.getTmpFileSuffix()); + synchronizer.setFilter(filter); + + return synchronizer; + } + + @Bean + public MessageSource s3MessageSource(S3InboundFileSynchronizer s3InboundFileSynchronizer) { + S3InboundFileSynchronizingMessageSource s3MessageSource = new S3InboundFileSynchronizingMessageSource( + s3InboundFileSynchronizer); + s3MessageSource.setLocalDirectory(this.awsS3SupplierProperties.getLocalDir()); + s3MessageSource.setAutoCreateLocalDirectory(this.awsS3SupplierProperties.isAutoCreateLocalDir()); + return s3MessageSource; } - return synchronizer; } - @Bean - public MessageSource s3MessageSource() { - S3InboundFileSynchronizingMessageSource s3MessageSource = - new S3InboundFileSynchronizingMessageSource(s3InboundFileSynchronizer()); - s3MessageSource.setLocalDirectory(this.awsS3SupplierProperties.getLocalDir()); - 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, + ResourceIdResolver resourceIdResolver, ConcurrentMetadataStore metadataStore) { + super(awsS3SupplierProperties, fileConsumerProperties, amazonS3, resourceIdResolver, metadataStore); + } - @Bean - public Publisher> s3SupplierFlow() { - return FileUtils.enhanceFlowForReadingMode(IntegrationFlows - .from(IntegrationReactiveUtils.messageSourceToFlux(s3MessageSource())), fileConsumerProperties) - .toReactivePublisher(); - } + @Bean + public Supplier>> s3Supplier(Publisher> s3SupplierFlow) { + return () -> Flux.from(s3SupplierFlow); + } - @Bean - public Supplier>> s3Supplier() { - return () -> Flux.from(s3SupplierFlow()); + @Bean + public Publisher> s3SupplierFlow(ReactiveMessageSourceProducer s3ListingProducer, + GenericSelector listOnlyFilter) { + return IntegrationFlows + .from(s3ListingProducer) + .split() + .filter(listOnlyFilter) + .toReactivePublisher(); + } + + @Bean + GenericSelector listOnlyFilter() { + Predicate predicate = s -> true; + if (StringUtils.hasText(this.awsS3SupplierProperties.getFilenamePattern())) { + Pattern pattern = Pattern.compile(this.awsS3SupplierProperties.getFilenamePattern()); + predicate = (S3ObjectSummary summary) -> pattern.matcher(summary.getKey()).matches(); + } + else if (this.awsS3SupplierProperties.getFilenameRegex() != null) { + predicate = (S3ObjectSummary summary) -> this.awsS3SupplierProperties.getFilenameRegex() + .matcher(summary.getKey()).matches(); + } + predicate = predicate.and((S3ObjectSummary summary) -> { + final String key = METADATA_STORE_PREFIX + summary.getBucketName() + "-" + summary.getKey(); + final String lastModified = String.valueOf(summary.getLastModified().getTime()); + final String storedLastModified = this.metadataStore.get(key); + boolean result = !lastModified.equals(storedLastModified); + if (result) { + metadataStore.put(key, lastModified); + } + return result; + }); + + GenericSelector selector = predicate::test; + + return selector; + } + + @Bean + ReactiveMessageSourceProducer s3ListingMessageProducer(AmazonS3 amazonS3, + AwsS3SupplierProperties awsS3SupplierProperties) { + ListObjectsRequest listObjectsRequest = new ListObjectsRequest(); + listObjectsRequest.setBucketName(awsS3SupplierProperties.getRemoteDir()); + return new ReactiveMessageSourceProducer( + (MessageSource>) () -> new GenericMessage<>( + amazonS3.listObjects(listObjectsRequest).getObjectSummaries())); + } } } diff --git a/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierProperties.java b/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierProperties.java index 5bf24501..7b50ede1 100644 --- a/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierProperties.java +++ b/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierProperties.java @@ -80,6 +80,11 @@ public class AwsS3SupplierProperties { */ private boolean preserveTimestamp = true; + /** + * Set to true to return s3 object metadata without copying file to a local directory. + */ + private boolean listOnly = false; + @Length(min = 3) public String getRemoteDir() { return this.remoteDir; diff --git a/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AbstractAwsS3SupplierMockTests.java b/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AbstractAwsS3SupplierMockTests.java index 161c7a2e..ea42103a 100644 --- a/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AbstractAwsS3SupplierMockTests.java +++ b/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AbstractAwsS3SupplierMockTests.java @@ -130,16 +130,17 @@ public abstract class AbstractAwsS3SupplierMockTests { Calendar calendar = Calendar.getInstance(); calendar.add(Calendar.DATE, 1); + ObjectListing objectListing = new ObjectListing(); + List objectSummaries = objectListing.getObjectSummaries(); + for (S3Object s3Object : S3_OBJECTS) { + S3ObjectSummary s3ObjectSummary = new S3ObjectSummary(); + s3ObjectSummary.setBucketName(S3_BUCKET); + s3ObjectSummary.setKey(s3Object.getKey()); + s3ObjectSummary.setLastModified(calendar.getTime()); + objectSummaries.add(s3ObjectSummary); + } + willAnswer(invocation -> { - ObjectListing objectListing = new ObjectListing(); - List objectSummaries = objectListing.getObjectSummaries(); - for (S3Object s3Object : S3_OBJECTS) { - S3ObjectSummary s3ObjectSummary = new S3ObjectSummary(); - s3ObjectSummary.setBucketName(S3_BUCKET); - s3ObjectSummary.setKey(s3Object.getKey()); - s3ObjectSummary.setLastModified(calendar.getTime()); - objectSummaries.add(s3ObjectSummary); - } return objectListing; }).given(amazonS3).listObjects(any(ListObjectsRequest.class)); diff --git a/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3ListOnlyTests.java b/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3ListOnlyTests.java new file mode 100644 index 00000000..207c3fd6 --- /dev/null +++ b/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3ListOnlyTests.java @@ -0,0 +1,69 @@ +/* + * Copyright 2020-2020 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. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.cloud.fn.supplier.s3; + +import java.time.Duration; +import java.util.HashSet; + +import com.amazonaws.services.s3.model.S3ObjectSummary; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; + +import org.springframework.messaging.Message; +import org.springframework.test.context.TestPropertySource; + +import static org.assertj.core.api.Assertions.assertThat; + +@TestPropertySource(properties = { + "s3.supplier.list-only=true" +}) +public class AmazonS3ListOnlyTests extends AbstractAwsS3SupplierMockTests { + + @Test + public void test() { + final Flux> messageFlux = s3Supplier.get(); + final HashSet keys = new HashSet<>(); + keys.add("1.test"); + keys.add("2.test"); + keys.add("otherFile"); + StepVerifier stepVerifier = StepVerifier.create(messageFlux) + .assertNext(message -> { + S3ObjectSummary summary = (S3ObjectSummary) message.getPayload(); + assertThat(summary.getBucketName()).isEqualTo(S3_BUCKET); + assertThat(keys.contains(summary.getKey())); + keys.remove(summary.getKey()); + }) + .assertNext(message -> { + S3ObjectSummary summary = (S3ObjectSummary) message.getPayload(); + assertThat(summary.getBucketName()).isEqualTo(S3_BUCKET); + assertThat(keys.contains(summary.getKey())); + keys.remove(summary.getKey()); + }) + .assertNext(message -> { + S3ObjectSummary summary = (S3ObjectSummary) message.getPayload(); + assertThat(summary.getBucketName()).isEqualTo(S3_BUCKET); + assertThat(keys.contains(summary.getKey())); + keys.remove(summary.getKey()); + }) + .expectTimeout(Duration.ofSeconds(1)) + .verifyLater(); + standardIntegrationFlow.start(); + stepVerifier.verify(Duration.ofSeconds(10)); + standardIntegrationFlow.stop(); + } +} diff --git a/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java b/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java index a9242a38..5e37ce0a 100644 --- a/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java +++ b/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java @@ -17,9 +17,10 @@ package org.springframework.cloud.fn.supplier.sftp; import java.io.IOException; -import java.util.Arrays; import java.util.List; +import java.util.function.Predicate; import java.util.function.Supplier; +import java.util.regex.Pattern; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -41,8 +42,10 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; import org.springframework.context.annotation.Primary; +import org.springframework.http.MediaType; import org.springframework.integration.aop.ReceiveMessageAdvice; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.GenericSelector; import org.springframework.integration.core.MessageSource; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; @@ -66,9 +69,7 @@ import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.MessageBuilder; -import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; -import org.springframework.util.MimeTypeUtils; import org.springframework.util.StringUtils; /** @@ -87,6 +88,8 @@ public class SftpSupplierConfiguration { private static final String METADATA_STORE_PREFIX = "sftpSource/"; + private static final String FILE_MODIFIED_TIME_HEADER = "FILE_MODIFIED_TIME"; + @Bean public Supplier>> sftpSupplier(MessageSource sftpMessageSource, @Nullable Publisher> sftpReadingFlow, @@ -130,7 +133,7 @@ public class SftpSupplierConfiguration { * Configure the standard filters for SFTP inbound adapters. */ @Bean - public ChainFileListFilter chainFilter(SftpSupplierProperties sftpSupplierProperties, + public FileListFilter chainFilter(SftpSupplierProperties sftpSupplierProperties, ConcurrentMetadataStore metadataStore) { ChainFileListFilter chainFilter = new ChainFileListFilter<>(); @@ -142,9 +145,7 @@ public class SftpSupplierConfiguration { chainFilter .addFilter(new SftpRegexPatternFileListFilter(sftpSupplierProperties.getFilenameRegex())); } - // TODO: Temporary work-around for - // https://github.com/spring-projects/spring-integration/issues/3315. - chainFilter.addFilter(Arrays::asList); + chainFilter.addFilter(new SftpPersistentAcceptOnceFileListFilter(metadataStore, METADATA_STORE_PREFIX)); return chainFilter; } @@ -279,7 +280,6 @@ public class SftpSupplierConfiguration { } @Bean - @SuppressWarnings("unchecked") public MessageSource targetMessageSource(PollableChannel listingChannel, SftpListingMessageProducer sftpListingMessageProducer) { return () -> { @@ -299,29 +299,70 @@ public class SftpSupplierConfiguration { } @Bean - public IntegrationFlow listingFlow(MessageProducerSupport messageProducerSupport, - MessageChannel listingChannel, MessageProcessor metadataWriter) { + GenericSelector listOnlyFilter(SftpSupplierProperties sftpSupplierProperties) { + Predicate predicate = s -> true; + if (StringUtils.hasText(sftpSupplierProperties.getFilenamePattern())) { + predicate = Pattern.compile(sftpSupplierProperties.getFilenamePattern()).asPredicate(); + } + else if (sftpSupplierProperties.getFilenameRegex() != null) { + predicate = sftpSupplierProperties.getFilenameRegex().asPredicate(); + } - return IntegrationFlows.from(messageProducerSupport) + GenericSelector selector = predicate::test; + + return selector; + } + + @Bean + public IntegrationFlow listingFlow(MessageProducerSupport listingMessageProducer, + MessageChannel listingChannel, MessageProcessor lsEntryToStringTransformer, + GenericSelector> duplicateFilter, + GenericSelector listOnlyFilter) { + + return IntegrationFlows.from(listingMessageProducer) .split() - .transform(metadataWriter) + .transform(lsEntryToStringTransformer) + .filter(duplicateFilter) + .filter(listOnlyFilter) .channel(listingChannel) .get(); } @Bean - public MessageProcessor metadataWriter(ConcurrentMetadataStore metadataStore) { - return message -> { - MessageHeaders messageHeaders = message.getHeaders(); - Assert.notNull(messageHeaders, "Cannot transform message with null headers"); - Assert.isTrue(messageHeaders.containsKey(FileHeaders.REMOTE_DIRECTORY), - "Remote directory header not found"); - Assert.hasText((String) message.getPayload(), "Payload must not be empty."); + public MessageProcessor> lsEntryToStringTransformer() { + return (Message message) -> { - metadataStore.putIfAbsent( - message.getHeaders().get(FileHeaders.REMOTE_DIRECTORY).toString() + message.getPayload(), - String.valueOf(message.getHeaders().getTimestamp())); - return message; + LsEntry lsEntry = (LsEntry) message.getPayload(); + + String fileName = message.getHeaders().get(FileHeaders.REMOTE_DIRECTORY) + lsEntry.getFilename(); + + return MessageBuilder.withPayload(fileName) + .copyHeaders(message.getHeaders()) + .setHeader(FILE_MODIFIED_TIME_HEADER, String.valueOf(lsEntry.getAttrs().getMTime())) + .setHeader(MessageHeaders.CONTENT_TYPE, MediaType.TEXT_PLAIN) + .build(); + }; + + } + + @Bean + GenericSelector> duplicateFilter(ConcurrentMetadataStore metadataStore) { + return new GenericSelector>() { + @Override + public boolean accept(Message message) { + + String lastModifiedTime = (String) message.getHeaders().get(FILE_MODIFIED_TIME_HEADER); + String storedLastModifiedTime = metadataStore.get(METADATA_STORE_PREFIX + message.getPayload()); + + boolean result = !lastModifiedTime.equals(storedLastModifiedTime); + + if (result) { + metadataStore.put( + METADATA_STORE_PREFIX + message.getPayload(), + message.getHeaders().get(FILE_MODIFIED_TIME_HEADER).toString()); + } + return result; + } }; } @@ -342,17 +383,19 @@ public class SftpSupplierConfiguration { } public void listNames() { - String[] names = {}; + LsEntry[] entries = {}; try { - names = Stream.of(this.sessionFactory.getSession().listNames(this.remoteDirectory)) - .map(name -> String.join(this.remoteFileSeparator, this.remoteDirectory, name)) - .collect(Collectors.toList()).toArray(names); + entries = Stream.of(this.sessionFactory.getSession().list(this.remoteDirectory)) + .filter(o -> { + LsEntry lsEntry = (LsEntry) o; + return !(lsEntry.getAttrs().isDir() || lsEntry.getAttrs().isLink()); + }) + .collect(Collectors.toList()).toArray(entries); } catch (IOException e) { throw new MessagingException(e.getMessage(), e); } - sendMessage(MessageBuilder.withPayload(names) - .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.TEXT_PLAIN) + sendMessage(MessageBuilder.withPayload(entries) .setHeader(FileHeaders.REMOTE_DIRECTORY, this.remoteDirectory + this.remoteFileSeparator) .build()); } diff --git a/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java b/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java index 3dc9c71a..6020a261 100644 --- a/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java +++ b/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java @@ -91,7 +91,28 @@ public class SftpSupplierApplicationTests extends SftpTestSupport { assertThat(message.getHeaders().get(MessageHeaders.CONTENT_TYPE)) .isEqualTo(MediaType.TEXT_PLAIN); }) - .thenCancel() + .expectTimeout(Duration.ofMillis(1000)) + .verify(Duration.ofSeconds(10)); + + }); + } + + @Test + void supplierForListOnlyWithPatternFilter() { + defaultApplicationContextRunner + .withPropertyValues("sftp.supplier.listOnly=true", "sftp.supplier.file-name-pattern=.*1.txt") + .run(context -> { + Supplier>> sftpSupplier = context.getBean("sftpSupplier", + Supplier.class); + SftpSupplierProperties properties = context.getBean(SftpSupplierProperties.class); + + final AtomicReference expectedFileName = new AtomicReference<>( + properties.getRemoteDir() + File.separator + "sftpSource1.txt"); + StepVerifier.create(sftpSupplier.get()) + .assertNext(message -> { + assertThat(expectedFileName.get()).contains(message.getPayload()); + }) + .expectTimeout(Duration.ofMillis(1000)) .verify(Duration.ofSeconds(10)); }); @@ -114,7 +135,7 @@ public class SftpSupplierApplicationTests extends SftpTestSupport { final AtomicReference> expectedFileNames = new AtomicReference<>(fileNames); StepVerifier.create(sftpSupplier.get()) .assertNext(message -> { - File file = (File) message.getPayload(); + File file = message.getPayload(); assertThat(expectedFileNames.get()).contains(file.getAbsolutePath()); expectedFileNames.get().remove(file.getAbsolutePath()); })