From 321eeb391edd1b231f36873aa258dfcbd60859b0 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Wed, 30 Mar 2022 11:41:34 -0400 Subject: [PATCH] GH-249: Add metadata-store to file-supplier (#251) Fixes https://github.com/spring-cloud/stream-applications/issues/249 Currently, the metadata store is supported by remote files source (S3, ftp and sftp), but not by file source. There are cases where it is convenient to be able to use the metadata store with file source as well. * Add externally configured `ConcurrentMetadataStore` into `FileSupplierConfiguration` which is based on the auto-configuration provided by the `metadata-store-common` artifact * Test with an embedded JDBC store * Remove redundant dependencies in the file-supplier pom - supplied by the parent * Add `MetadataStoreProperties` to allow-list for dataflow * Fix typo in the `s3-supplier` README --- supplier/file-supplier/pom.xml | 33 +++++++--------- .../file/FileSupplierConfiguration.java | 38 ++++++++++++++----- .../file/DefaultFileSupplierTests.java | 30 +++++++++++++-- supplier/s3-supplier/README.adoc | 2 +- 4 files changed, 69 insertions(+), 34 deletions(-) diff --git a/supplier/file-supplier/pom.xml b/supplier/file-supplier/pom.xml index b1ba628d..a2f89922 100644 --- a/supplier/file-supplier/pom.xml +++ b/supplier/file-supplier/pom.xml @@ -19,36 +19,29 @@ spring-integration-file - org.springframework.boot - spring-boot-starter-integration + org.springframework.cloud.fn + metadata-store-common + ${revision} org.springframework.cloud.fn file-common ${project.version} + - org.springframework.boot - spring-boot-starter-json - true - - - org.springframework.boot - spring-boot-starter-validation - - - org.springframework.boot - spring-boot-configuration-processor - provided - - - org.springframework.boot - spring-boot-starter-test + org.springframework.integration + spring-integration-jdbc test - io.projectreactor - reactor-test + org.springframework.boot + spring-boot-starter-jdbc + test + + + org.hsqldb + hsqldb test diff --git a/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java b/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java index 50b275e5..fe2b483b 100644 --- a/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java +++ b/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2020 the original author or authors. + * Copyright 2020-2022 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,6 +16,7 @@ package org.springframework.cloud.fn.supplier.file; +import java.io.File; import java.util.function.Supplier; import org.reactivestreams.Publisher; @@ -38,6 +39,12 @@ import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.file.FileReadingMessageSource; import org.springframework.integration.file.dsl.FileInboundChannelAdapterSpec; import org.springframework.integration.file.dsl.Files; +import org.springframework.integration.file.filters.ChainFileListFilter; +import org.springframework.integration.file.filters.FileListFilter; +import org.springframework.integration.file.filters.FileSystemPersistentAcceptOnceFileListFilter; +import org.springframework.integration.file.filters.RegexPatternFileListFilter; +import org.springframework.integration.file.filters.SimplePatternFileListFilter; +import org.springframework.integration.metadata.ConcurrentMetadataStore; import org.springframework.messaging.Message; import org.springframework.util.StringUtils; @@ -49,6 +56,8 @@ import org.springframework.util.StringUtils; @EnableConfigurationProperties({FileSupplierProperties.class, FileConsumerProperties.class}) public class FileSupplierConfiguration { + private static final String METADATA_STORE_PREFIX = "local-file-system-metadata-"; + private final FileSupplierProperties fileSupplierProperties; private final FileConsumerProperties fileConsumerProperties; @@ -60,22 +69,32 @@ public class FileSupplierConfiguration { public FileSupplierConfiguration(FileSupplierProperties fileSupplierProperties, FileConsumerProperties fileConsumerProperties) { + this.fileSupplierProperties = fileSupplierProperties; this.fileConsumerProperties = fileConsumerProperties; } @Bean - public FileInboundChannelAdapterSpec fileMessageSource() { - final FileInboundChannelAdapterSpec fileInboundChannelAdapterSpec = - Files.inboundAdapter(this.fileSupplierProperties.getDirectory()); + public ChainFileListFilter filter(ConcurrentMetadataStore metadataStore) { + ChainFileListFilter chainFilter = new ChainFileListFilter<>(); if (StringUtils.hasText(this.fileSupplierProperties.getFilenamePattern())) { - fileInboundChannelAdapterSpec.patternFilter(this.fileSupplierProperties.getFilenamePattern()); + chainFilter.addFilter(new SimplePatternFileListFilter(this.fileSupplierProperties.getFilenamePattern())); } else if (this.fileSupplierProperties.getFilenameRegex() != null) { - fileInboundChannelAdapterSpec.regexFilter(this.fileSupplierProperties.getFilenameRegex().pattern()); + chainFilter.addFilter(new RegexPatternFileListFilter(this.fileSupplierProperties.getFilenameRegex())); } - fileInboundChannelAdapterSpec.preventDuplicates(this.fileSupplierProperties.isPreventDuplicates()); - return fileInboundChannelAdapterSpec; + + if (this.fileSupplierProperties.isPreventDuplicates()) { + chainFilter.addFilter(new FileSystemPersistentAcceptOnceFileListFilter(metadataStore, METADATA_STORE_PREFIX)); + } + + return chainFilter; + } + + @Bean + public FileInboundChannelAdapterSpec fileMessageSource(FileListFilter fileListFilter) { + return Files.inboundAdapter(this.fileSupplierProperties.getDirectory()) + .filter(fileListFilter); } @Bean @@ -86,7 +105,7 @@ public class FileSupplierConfiguration { .subscribeOn(Schedulers.boundedElastic()) .repeatWhenEmpty(it -> it.delayElements(this.fileSupplierProperties.getDelayWhenEmpty())) .repeat() - .doOnSubscribe(s -> this.fileMessageSource.start()); + .doOnRequest(r -> this.fileMessageSource.start()); } @Bean @@ -106,4 +125,5 @@ public class FileSupplierConfiguration { return () -> Flux.from(fileReadingFlow()); } } + } diff --git a/supplier/file-supplier/src/test/java/org/springframework/cloud/fn/supplier/file/DefaultFileSupplierTests.java b/supplier/file-supplier/src/test/java/org/springframework/cloud/fn/supplier/file/DefaultFileSupplierTests.java index 234e5fd8..e1068243 100644 --- a/supplier/file-supplier/src/test/java/org/springframework/cloud/fn/supplier/file/DefaultFileSupplierTests.java +++ b/supplier/file-supplier/src/test/java/org/springframework/cloud/fn/supplier/file/DefaultFileSupplierTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2020 the original author or authors. + * Copyright 2020-2022 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. @@ -19,16 +19,22 @@ package org.springframework.cloud.fn.supplier.file; import java.io.IOException; import java.nio.file.Files; import java.nio.file.Path; +import java.util.List; import org.junit.jupiter.api.Test; -import reactor.core.publisher.Flux; -import reactor.test.StepVerifier; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.FileReadingMessageSource; +import org.springframework.integration.jdbc.metadata.JdbcMetadataStore; +import org.springframework.integration.metadata.ConcurrentMetadataStore; +import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.messaging.Message; +import org.springframework.test.context.TestPropertySource; + +import reactor.core.publisher.Flux; +import reactor.test.StepVerifier; import static org.assertj.core.api.Assertions.assertThat; @@ -37,15 +43,21 @@ import static org.assertj.core.api.Assertions.assertThat; * @author Artem Bilan * @author Soby Chacko */ +@TestPropertySource(properties = "metadata.store.type = jdbc") public class DefaultFileSupplierTests extends AbstractFileSupplierTests { @Autowired @Qualifier("fileMessageSource") private FileReadingMessageSource fileMessageSource; + @Autowired + private ConcurrentMetadataStore metadataStore; + + @Autowired + private JdbcTemplate jdbcTemplate; + @Test public void testBasicFlow() throws IOException { - Path firstFile = tempDir.resolve("first.file"); Files.write(firstFile, "first.file".getBytes()); @@ -83,6 +95,16 @@ public class DefaultFileSupplierTests extends AbstractFileSupplierTests { stepVerifier.verify(); assertThat(this.fileMessageSource.isRunning()).isTrue(); + + assertThat(this.metadataStore).isInstanceOf(JdbcMetadataStore.class); + + List metadataStoreContent = + jdbcTemplate.queryForList("SELECT metadata_key FROM int_metadata_store", String.class); + + assertThat(metadataStoreContent).hasSize(2); + assertThat(metadataStoreContent.get(0)).startsWith("local-file-system-metadata-"); + assertThat(metadataStoreContent.get(0)).endsWith("first.file"); + assertThat(metadataStoreContent.get(1)).endsWith("test.file"); } } diff --git a/supplier/s3-supplier/README.adoc b/supplier/s3-supplier/README.adoc index c01ccc61..973a03b4 100644 --- a/supplier/s3-supplier/README.adoc +++ b/supplier/s3-supplier/README.adoc @@ -23,7 +23,7 @@ Once injected, you can use the `get` method of the `Supplier` to invoke it and t All configuration properties are prefixed with `s3.supplier`. There are also properties that need to be used with the prefix `s3.common` and `file.consumer`. -For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierProperties.java[AwsS3upplierProperties], +For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierProperties.java[AwsS3SupplierProperties], link:../../common/file-common/src/main/java/org/springframework/cloud/fn/common/file/FileConsumerProperties.java[FileConsumerProperties], and link:../../common/aws-s3-common/src/main/java/org/springframework/cloud/fn/common/aws/s3/AmazonS3Properties.java[AmazonS3Properties].