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

GH-249: Add metadata-store to file-supplier (#251)

Browse Source 
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
This commit is contained in:
Artem Bilan
2022-03-30 11:41:34 -04:00
committed by GitHub
parent dfc000cd49
commit 321eeb391e
4 changed files with 69 additions and 34 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
supplier/file-supplier/pom.xml
View File

@@ -19,36 +19,29 @@
<artifactId>spring-integration-file</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-integration</artifactId>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>metadata-store-common</artifactId>
<version>${revision}</version>
</dependency>
<dependency>
<groupId>org.springframework.cloud.fn</groupId>
<artifactId>file-common</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-json</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-jdbc</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-jdbc</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.hsqldb</groupId>
<artifactId>hsqldb</artifactId>
<scope>test</scope>
</dependency>
</dependencies>

38
supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java
View File

@@ -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<File> filter(ConcurrentMetadataStore metadataStore) {
ChainFileListFilter<File> 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<File> 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());
}
}
}

30
supplier/file-supplier/src/test/java/org/springframework/cloud/fn/supplier/file/DefaultFileSupplierTests.java
View File

@@ -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<String> 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");
}
}

2
supplier/s3-supplier/README.adoc
View File

@@ -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].