From ddcc10363e708c93af010c1f3553ed7cca8a954d Mon Sep 17 00:00:00 2001 From: David Turanski Date: Thu, 17 Sep 2020 15:36:36 -0400 Subject: [PATCH] List only enhancements * Add metadata.store.type Property and idempotent SftpSupplier for list-only * Implement list-only for S3 source and Optimize metadastore access * Fixed build and READMEs * Change to ConditionalOnProperty * Change to ReactiveMessageProducer * Update cdc-debezium-source/README.adoc * Make all MetadataStoreProperties visible --- .../ImageRecognitionProcessorTests.java | 177 ++++++++-------- .../source/cdc-debezium-source/README.adoc | 19 +- ...onfiguration-metadata-whitelist.properties | 9 +- ...dataflow-configuration-metadata.properties | 12 ++ applications/source/ftp-source/README.adoc | 18 +- ...onfiguration-metadata-whitelist.properties | 9 +- ...dataflow-configuration-metadata.properties | 9 +- applications/source/s3-source/README.adoc | 19 ++ ...onfiguration-metadata-whitelist.properties | 9 +- ...dataflow-configuration-metadata.properties | 9 +- applications/source/sftp-source/README.adoc | 19 ++ ...onfiguration-metadata-whitelist.properties | 9 +- ...dataflow-configuration-metadata.properties | 9 +- .../app/source/time/TimeSourceTests.java | 1 - .../pom.xml | 31 ++- .../common/AbstractMicrometerTagTest.java | 12 +- .../common/metadata-store-common/README.adoc | 2 +- .../common/metadata-store-common/pom.xml | 6 + .../store/MetadataStoreAutoConfiguration.java | 47 ++--- .../store/MetadataStoreProperties.java | 25 +++ functions/supplier/s3-supplier/pom.xml | 5 + .../s3/AwsS3SupplierConfiguration.java | 193 +++++++++++++----- .../supplier/s3/AwsS3SupplierProperties.java | 5 + .../s3/AbstractAwsS3SupplierMockTests.java | 19 +- .../fn/supplier/s3/AmazonS3ListOnlyTests.java | 69 +++++++ .../sftp/SftpSupplierConfiguration.java | 101 ++++++--- .../sftp/SftpSupplierApplicationTests.java | 25 ++- 27 files changed, 655 insertions(+), 213 deletions(-) create mode 100644 applications/source/cdc-debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties create mode 100644 functions/supplier/s3-supplier/src/test/java/org/springframework/cloud/fn/supplier/s3/AmazonS3ListOnlyTests.java diff --git a/applications/processor/image-recognition-processor/src/test/java/org/springframework/cloud/stream/app/processor/image/recognition/ImageRecognitionProcessorTests.java b/applications/processor/image-recognition-processor/src/test/java/org/springframework/cloud/stream/app/processor/image/recognition/ImageRecognitionProcessorTests.java index 9364232c..5695ed11 100644 --- a/applications/processor/image-recognition-processor/src/test/java/org/springframework/cloud/stream/app/processor/image/recognition/ImageRecognitionProcessorTests.java +++ b/applications/processor/image-recognition-processor/src/test/java/org/springframework/cloud/stream/app/processor/image/recognition/ImageRecognitionProcessorTests.java @@ -17,11 +17,16 @@ package org.springframework.cloud.stream.app.processor.image.recognition; import java.io.IOException; +import java.math.BigDecimal; +import java.math.RoundingMode; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; import java.util.function.Consumer; +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.condition.EnabledOnOs; -import org.junit.jupiter.api.condition.OS; import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.SpringBootApplication; @@ -42,39 +47,29 @@ import static org.assertj.core.api.Assertions.assertThat; */ public class ImageRecognitionProcessorTests { - @Test - @EnabledOnOs(OS.MAC) - public void testImageRecognitionProcessorMobileNetV2Mac() throws IOException { - testImageRecognitionProcessorMobileNetV2(message -> { - String jsonRecognizedObjects = (String) message.getHeaders().get(ImageRecognitionProcessorConfiguration.RECOGNIZED_OBJECTS_HEADER); - assertThat(jsonRecognizedObjects) - .isEqualTo("[{\"label\":\"giant panda, panda, panda bear, coon bear, Ailuropoda melanoleuca\",\"probability\":0.962329626083374}," + - "{\"label\":\"badger\",\"probability\":0.006058811210095882}," + - "{\"label\":\"ram, tup\",\"probability\":0.0010668420000001788}]"); - }); - } + private ObjectMapper objectMapper = new ObjectMapper(); @Test - @EnabledOnOs(OS.LINUX) - public void testImageRecognitionProcessorMobileNetV2Linux() throws IOException { - testImageRecognitionProcessorMobileNetV2(message -> { - String jsonRecognizedObjects = (String) message.getHeaders().get(ImageRecognitionProcessorConfiguration.RECOGNIZED_OBJECTS_HEADER); - assertThat(jsonRecognizedObjects) - .isEqualTo("[{\"label\":\"giant panda, panda, panda bear, coon bear, Ailuropoda melanoleuca\",\"probability\":0.9623297452926636}," + - "{\"label\":\"badger\",\"probability\":0.006058800499886274}," + - "{\"label\":\"ram, tup\",\"probability\":0.0010668395552784204}]"); - }); + public void testImageRecognitionProcessorMobileNetV2() throws IOException { + List> expected = deserializeAndRoundToNPlaces( + "[{\"label\":\"giant panda, panda, panda bear, coon bear, Ailuropoda melanoleuca\",\"probability\":0.962329626083374}," + + + "{\"label\":\"badger\",\"probability\":0.006058811210095882}," + + "{\"label\":\"ram, tup\",\"probability\":0.0010668420000001788}]", + 6); + + imageRecognitionProcessorMobileNetV2(verify(expected)); } - private void testImageRecognitionProcessorMobileNetV2(Consumer> consumer) throws IOException { + private void imageRecognitionProcessorMobileNetV2(Consumer> consumer) throws IOException { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( TestChannelBinderConfiguration.getCompleteConfiguration(ImageRecognitionProcessorTestApplication.class)) - .web(WebApplicationType.NONE) - .run("--spring.cloud.function.definition=imageRecognitionFunction", - "--image.recognition.modelType=mobilenetv2", - "--image.recognition.responseSize=3", - "--image.recognition.debugOutput=true", - "--image.recognition.debugOutputPath=./target/image-recognition-mobilenetv2.png")) { + .web(WebApplicationType.NONE) + .run("--spring.cloud.function.definition=imageRecognitionFunction", + "--image.recognition.modelType=mobilenetv2", + "--image.recognition.responseSize=3", + "--image.recognition.debugOutput=true", + "--image.recognition.debugOutputPath=./target/image-recognition-mobilenetv2.png")) { InputDestination processorInput = context.getBean(InputDestination.class); OutputDestination processorOutput = context.getBean(OutputDestination.class); @@ -87,38 +82,26 @@ public class ImageRecognitionProcessorTests { } @Test - @EnabledOnOs(OS.MAC) - public void testImageRecognitionProcessorMobileNetV1Mac() throws IOException { - testImageRecognitionProcessorMobileNetV1(message -> { - String jsonRecognizedObjects = (String) message.getHeaders().get(ImageRecognitionProcessorConfiguration.RECOGNIZED_OBJECTS_HEADER); - assertThat(jsonRecognizedObjects) - .isEqualTo("[{\"label\":\"giant panda, panda, panda bear, coon bear, Ailuropoda melanoleuca\",\"probability\":0.984053909778595}," + - "{\"label\":\"ram, tup\",\"probability\":0.0019619385711848736}," + - "{\"label\":\"Staffordshire bullterrier, Staffordshire bull terrier\",\"probability\":0.0018697341438382864}]"); - }); + public void testImageRecognitionProcessorMobileNetV1() throws IOException { + List> expected = deserializeAndRoundToNPlaces( + "[{\"label\":\"giant panda, panda, panda bear, coon bear, Ailuropoda melanoleuca\",\"probability\":0.984053909778595}," + + + "{\"label\":\"ram, tup\",\"probability\":0.0019619385711848736}," + + "{\"label\":\"Staffordshire bullterrier, Staffordshire bull terrier\",\"probability\":0.0018697341438382864}]", + 6); + + imageRecognitionProcessorMobileNetV1(verify(expected)); } - @Test - @EnabledOnOs(OS.LINUX) - public void testImageRecognitionProcessorMobileNetV1Linux() throws IOException { - testImageRecognitionProcessorMobileNetV1(message -> { - String jsonRecognizedObjects = (String) message.getHeaders().get(ImageRecognitionProcessorConfiguration.RECOGNIZED_OBJECTS_HEADER); - assertThat(jsonRecognizedObjects) - .isEqualTo("[{\"label\":\"giant panda, panda, panda bear, coon bear, Ailuropoda melanoleuca\",\"probability\":0.9840537905693054}," + - "{\"label\":\"ram, tup\",\"probability\":0.0019619381055235863}," + - "{\"label\":\"Staffordshire bullterrier, Staffordshire bull terrier\",\"probability\":0.001869735773652792}]"); - }); - } - - private void testImageRecognitionProcessorMobileNetV1(Consumer> consumer) throws IOException { + private void imageRecognitionProcessorMobileNetV1(Consumer> consumer) throws IOException { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( TestChannelBinderConfiguration.getCompleteConfiguration(ImageRecognitionProcessorTestApplication.class)) - .web(WebApplicationType.NONE) - .run("--image.recognition.model=https://download.tensorflow.org/models/mobilenet_v1_2018_08_02/mobilenet_v1_1.0_224.tgz#mobilenet_v1_1.0_224_frozen.pb", - "--image.recognition.modelType=mobilenetv1", - "--image.recognition.responseSize=3", - "--image.recognition.debugOutput=true", - "--image.recognition.debugOutputPath=./target/image-recognition-mobilenetv1.png")) { + .web(WebApplicationType.NONE) + .run("--image.recognition.model=https://download.tensorflow.org/models/mobilenet_v1_2018_08_02/mobilenet_v1_1.0_224.tgz#mobilenet_v1_1.0_224_frozen.pb", + "--image.recognition.modelType=mobilenetv1", + "--image.recognition.responseSize=3", + "--image.recognition.debugOutput=true", + "--image.recognition.debugOutputPath=./target/image-recognition-mobilenetv1.png")) { InputDestination processorInput = context.getBean(InputDestination.class); OutputDestination processorOutput = context.getBean(OutputDestination.class); @@ -131,38 +114,24 @@ public class ImageRecognitionProcessorTests { } @Test - @EnabledOnOs(OS.MAC) - public void testImageRecognitionProcessorInceptionMac() throws IOException { - testImageRecognitionProcessorInception(message -> { - String jsonRecognizedObjects = (String) message.getHeaders().get(ImageRecognitionProcessorConfiguration.RECOGNIZED_OBJECTS_HEADER); - assertThat(jsonRecognizedObjects) - .isEqualTo("[{\"label\":\"giant panda\",\"probability\":0.9946685433387756}," + - "{\"label\":\"Arctic fox\",\"probability\":0.0036631159018725157}," + - "{\"label\":\"ice bear\",\"probability\":3.378273395355791E-4}]"); - }); + public void testImageRecognitionProcessorInception() throws IOException { + List> expected = deserializeAndRoundToNPlaces( + "[{\"label\":\"giant panda\",\"probability\":0.9946685433387756}," + + "{\"label\":\"Arctic fox\",\"probability\":0.003663112409412861}," + + "{\"label\":\"ice bear\",\"probability\":3.378273395355791E-4}]", + 6); + imageRecognitionProcessorInception(verify(expected)); } - @Test - @EnabledOnOs(OS.LINUX) - public void testImageRecognitionProcessorInceptionLinux() throws IOException { - testImageRecognitionProcessorInception(message -> { - String jsonRecognizedObjects = (String) message.getHeaders().get(ImageRecognitionProcessorConfiguration.RECOGNIZED_OBJECTS_HEADER); - assertThat(jsonRecognizedObjects) - .isEqualTo("[{\"label\":\"giant panda\",\"probability\":0.9946685433387756}," + - "{\"label\":\"Arctic fox\",\"probability\":0.003663112409412861}," + - "{\"label\":\"ice bear\",\"probability\":3.378273395355791E-4}]"); - }); - } - - private void testImageRecognitionProcessorInception(Consumer> consumer) throws IOException { + private void imageRecognitionProcessorInception(Consumer> consumer) throws IOException { try (ConfigurableApplicationContext context = new SpringApplicationBuilder( TestChannelBinderConfiguration.getCompleteConfiguration(ImageRecognitionProcessorTestApplication.class)) - .web(WebApplicationType.NONE) - .run("--image.recognition.model=https://storage.googleapis.com/scdf-tensorflow-models/image-recognition/tensorflow_inception_graph.pb", - "--image.recognition.modelType=inception", - "--image.recognition.responseSize=3", - "--image.recognition.debugOutput=true", - "--image.recognition.debugOutputPath=./target/image-recognition-inception.png")) { + .web(WebApplicationType.NONE) + .run("--image.recognition.model=https://storage.googleapis.com/scdf-tensorflow-models/image-recognition/tensorflow_inception_graph.pb", + "--image.recognition.modelType=inception", + "--image.recognition.responseSize=3", + "--image.recognition.debugOutput=true", + "--image.recognition.debugOutputPath=./target/image-recognition-inception.png")) { InputDestination processorInput = context.getBean(InputDestination.class); OutputDestination processorOutput = context.getBean(OutputDestination.class); @@ -175,6 +144,44 @@ public class ImageRecognitionProcessorTests { } } + private Consumer> verify(List> expected) { + return message -> { + List> actual = deserializeAndRoundToNPlaces((String) message.getHeaders() + .get(ImageRecognitionProcessorConfiguration.RECOGNIZED_OBJECTS_HEADER), 6); + assertThat(expected) + .isEqualTo(actual); + }; + } + + private List> deserializeAndRoundToNPlaces(String json, int places) { + List> result = null; + try { + result = objectMapper.readValue(json, ArrayList.class); + } + catch (JsonProcessingException e) { + throw new IllegalStateException(e.getMessage(), e); + } + + result.forEach(map -> { + if (map.containsKey("probability")) { + map.put("probability", round((double) map.get("probability"), places)); + } + }); + + return result; + + } + + private static double round(double value, int places) { + if (places < 0) { + throw new IllegalArgumentException(); + } + + BigDecimal bd = new BigDecimal(Double.toString(value)); + bd = bd.setScale(places, RoundingMode.HALF_UP); + return bd.doubleValue(); + } + @SpringBootApplication @Import({ ImageRecognitionProcessorConfiguration.class }) public static class ImageRecognitionProcessorTestApplication { diff --git a/applications/source/cdc-debezium-source/README.adoc b/applications/source/cdc-debezium-source/README.adoc index ddc9a2f8..758304bb 100644 --- a/applications/source/cdc-debezium-source/README.adoc +++ b/applications/source/cdc-debezium-source/README.adoc @@ -11,7 +11,8 @@ It supports all Debezium configuration properties. Just add the `cdc.config.` pr We provide convenient shortcuts for the most frequently used Debezium properties. For example instead of the long `cdc.config.connector.class=io.debezium.connector.mysql.MySqlConnector` Debezium property you can use our `cdc.connector=mysql` shortcut. The table below lists all available shortcuts along with the Debezium properties they represent. The Debezium properties (e.g. `cdc.config.XXX`) always have precedence over the shortcuts! -The CDC Source introduces a new default `BackingOffsetStore` configuration, based on the [MetadataStore](https://github.com/spring-cloud/stream-applications/tree/master/functions/common/metadata-store-common) service. Later provides various microservices friendly ways for storing the offset metadata. +The CDC Source introduces a new default `BackingOffsetStore` configuration, based on the link:../../../functions/common/metadata-store-common/README.adoc[MetadataStore] service. Later provides various microservices friendly ways for storing the offset metadata. + == Options @@ -31,6 +32,22 @@ $$cdc.offset.storage$$:: $$Kafka connector tracks the number processed records a $$cdc.schema$$:: $$Include the schema's as part of the outbound message.$$ *($$Boolean$$, default: `$$false$$`)* $$cdc.stream.header.convert-connect-headers$$:: $$When true the {@link org.apache.kafka.connect.header.Header} are converted into message headers with the {@link org.apache.kafka.connect.header.Header#key()} as name and {@link org.apache.kafka.connect.header.Header#value()}.$$ *($$Boolean$$, default: `$$true$$`)* $$cdc.stream.header.offset$$:: $$Serializes the source record's offset metadata into the outbound message header under cdc.offset.$$ *($$Boolean$$, default: `$$false$$`)* +$$metadata.store.dynamo-db.create-delay$$:: $$Delay between create table retries.$$ *($$Integer$$, default: `$$1$$`)* +$$metadata.store.dynamo-db.create-retries$$:: $$Retry number for create table request.$$ *($$Integer$$, default: `$$25$$`)* +$$metadata.store.dynamo-db.read-capacity$$:: $$Read capacity on the table.$$ *($$Long$$, default: `$$1$$`)* +$$metadata.store.dynamo-db.table$$:: $$Table name for metadata.$$ *($$String$$, default: `$$$$`)* +$$metadata.store.dynamo-db.time-to-live$$:: $$TTL for table entries.$$ *($$Integer$$, default: `$$$$`)* +$$metadata.store.dynamo-db.write-capacity$$:: $$Write capacity on the table.$$ *($$Long$$, default: `$$1$$`)* +$$metadata.store.gemfire.region$$:: $$Gemfire region name for metadata.$$ *($$String$$, default: `$$$$`)* +$$metadata.store.jdbc.region$$:: $$Unique grouping identifier for messages persisted with this store.$$ *($$String$$, default: `$$DEFAULT$$`)* +$$metadata.store.jdbc.table-prefix$$:: $$Prefix for the custom table name.$$ *($$String$$, default: `$$$$`)* +$$metadata.store.mongo-db.collection$$:: $$MongoDB collection name for metadata.$$ *($$String$$, default: `$$metadataStore$$`)* +$$metadata.store.redis.key$$:: $$Redis key for metadata.$$ *($$String$$, default: `$$$$`)* +$$metadata.store.type$$:: $$Indicates the type of metadata store to configure (default is 'memory'). You must include the corresponding Spring Integration dependency to use a persistent store.$$ *($$StoreType$$, default: `$$$$`, possible values: `mongodb`,`gemfire`,`redis`,`dynamodb`,`jdbc`,`zookeeper`,`hazelcast`,`memory`)* +$$metadata.store.zookeeper.connect-string$$:: $$Zookeeper connect string in form HOST:PORT.$$ *($$String$$, default: `$$127.0.0.1:2181$$`)* +$$metadata.store.zookeeper.encoding$$:: $$Encoding to use when storing data in Zookeeper.$$ *($$Charset$$, default: `$$UTF-8$$`)* +$$metadata.store.zookeeper.retry-interval$$:: $$Retry interval for Zookeeper operations in milliseconds.$$ *($$Integer$$, default: `$$1000$$`)* +$$metadata.store.zookeeper.root$$:: $$Root node - store entries are children of this node.$$ *($$String$$, default: `$$/SpringIntegration-MetadataStore$$`)* //end::configuration-properties[] ==== Debezium property Shortcut mapping diff --git a/applications/source/cdc-debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/source/cdc-debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties index 38208648..30b9254a 100644 --- a/applications/source/cdc-debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties +++ b/applications/source/cdc-debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -2,4 +2,11 @@ configuration-properties.classes=org.springframework.cloud.fn.supplier.cdc.CdcSu org.springframework.cloud.fn.supplier.cdc.CdcSupplierProperties$Header, \ org.springframework.cloud.fn.common.cdc.CdcCommonProperties, \ org.springframework.cloud.fn.common.cdc.CdcCommonProperties$Flattering, \ - org.springframework.cloud.fn.common.cdc.CdcCommonProperties$Offset + org.springframework.cloud.fn.common.cdc.CdcCommonProperties$Offset, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Gemfire, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Redis, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$DynamoDb, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Jdbc, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Zookeeper, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Mongo \ diff --git a/applications/source/cdc-debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties b/applications/source/cdc-debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties new file mode 100644 index 00000000..30b9254a --- /dev/null +++ b/applications/source/cdc-debezium-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties @@ -0,0 +1,12 @@ +configuration-properties.classes=org.springframework.cloud.fn.supplier.cdc.CdcSupplierProperties, \ + org.springframework.cloud.fn.supplier.cdc.CdcSupplierProperties$Header, \ + org.springframework.cloud.fn.common.cdc.CdcCommonProperties, \ + org.springframework.cloud.fn.common.cdc.CdcCommonProperties$Flattering, \ + org.springframework.cloud.fn.common.cdc.CdcCommonProperties$Offset, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Gemfire, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Redis, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$DynamoDb, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Jdbc, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Zookeeper, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Mongo \ diff --git a/applications/source/ftp-source/README.adoc b/applications/source/ftp-source/README.adoc index e98b80ad..c0c4d539 100644 --- a/applications/source/ftp-source/README.adoc +++ b/applications/source/ftp-source/README.adoc @@ -13,7 +13,7 @@ When using `--mode=lines`, you can also provide the additional option `--withMar If set to `true`, the underlying `FileSplitter` will emit additional _start-of-file_ and _end-of-file_ marker messages before and after the actual data. The payload of these 2 additional marker messages is of type `FileSplitter.FileMarker`. The option `withMarkers` defaults to `false` if not explicitly set. -See also https://github.com/spring-cloud/stream-applications/blob/master/functions/common/metadata-store-common/README.adoc[`MetaDataStore`] options for possible shared persistent store configuration for the `FtpPersistentAcceptOnceFileListFilter` used in the FTP Source. +See also link:../../../functions/common/metadata-store-common/README.adoc[MetadataStore] options for possible shared persistent store configuration used to prevent duplicate messages on restart. == Input @@ -87,6 +87,22 @@ $$ftp.supplier.preserve-timestamp$$:: $$Set to true to preserve the original tim $$ftp.supplier.remote-dir$$:: $$The remote FTP directory.$$ *($$String$$, default: `$$/$$`)* $$ftp.supplier.remote-file-separator$$:: $$The remote file separator.$$ *($$String$$, default: `$$/$$`)* $$ftp.supplier.tmp-file-suffix$$:: $$The suffix to use while the transfer is in progress.$$ *($$String$$, default: `$$.tmp$$`)* +$$metadata.store.dynamo-db.create-delay$$:: $$Delay between create table retries.$$ *($$Integer$$, default: `$$1$$`)* +$$metadata.store.dynamo-db.create-retries$$:: $$Retry number for create table request.$$ *($$Integer$$, default: `$$25$$`)* +$$metadata.store.dynamo-db.read-capacity$$:: $$Read capacity on the table.$$ *($$Long$$, default: `$$1$$`)* +$$metadata.store.dynamo-db.table$$:: $$Table name for metadata.$$ *($$String$$, default: `$$$$`)* +$$metadata.store.dynamo-db.time-to-live$$:: $$TTL for table entries.$$ *($$Integer$$, default: `$$$$`)* +$$metadata.store.dynamo-db.write-capacity$$:: $$Write capacity on the table.$$ *($$Long$$, default: `$$1$$`)* +$$metadata.store.gemfire.region$$:: $$Gemfire region name for metadata.$$ *($$String$$, default: `$$$$`)* +$$metadata.store.jdbc.region$$:: $$Unique grouping identifier for messages persisted with this store.$$ *($$String$$, default: `$$DEFAULT$$`)* +$$metadata.store.jdbc.table-prefix$$:: $$Prefix for the custom table name.$$ *($$String$$, default: `$$$$`)* +$$metadata.store.mongo-db.collection$$:: $$MongoDB collection name for metadata.$$ *($$String$$, default: `$$metadataStore$$`)* +$$metadata.store.redis.key$$:: $$Redis key for metadata.$$ *($$String$$, default: `$$$$`)* +$$metadata.store.type$$:: $$Indicates the type of metadata store to configure (default is 'memory'). You must include the corresponding Spring Integration dependency to use a persistent store.$$ *($$StoreType$$, default: `$$$$`, possible values: `mongodb`,`gemfire`,`redis`,`dynamodb`,`jdbc`,`zookeeper`,`hazelcast`,`memory`)* +$$metadata.store.zookeeper.connect-string$$:: $$Zookeeper connect string in form HOST:PORT.$$ *($$String$$, default: `$$127.0.0.1:2181$$`)* +$$metadata.store.zookeeper.encoding$$:: $$Encoding to use when storing data in Zookeeper.$$ *($$Charset$$, default: `$$UTF-8$$`)* +$$metadata.store.zookeeper.retry-interval$$:: $$Retry interval for Zookeeper operations in milliseconds.$$ *($$Integer$$, default: `$$1000$$`)* +$$metadata.store.zookeeper.root$$:: $$Root node - store entries are children of this node.$$ *($$String$$, default: `$$/SpringIntegration-MetadataStore$$`)* //end::configuration-properties[] == Examples diff --git a/applications/source/ftp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/source/ftp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties index b4779474..25ea3cc2 100644 --- a/applications/source/ftp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties +++ b/applications/source/ftp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -1,3 +1,10 @@ configuration-properties.classes=org.springframework.cloud.fn.supplier.ftp.FtpSupplierProperties, \ org.springframework.cloud.fn.common.ftp.FtpSessionFactoryProperties, \ - org.springframework.cloud.fn.common.file.FileConsumerProperties + org.springframework.cloud.fn.common.file.FileConsumerProperties, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Gemfire, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Redis, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$DynamoDb, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Jdbc, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Zookeeper, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Mongo \ diff --git a/applications/source/ftp-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties b/applications/source/ftp-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties index b4779474..25ea3cc2 100644 --- a/applications/source/ftp-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties +++ b/applications/source/ftp-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties @@ -1,3 +1,10 @@ configuration-properties.classes=org.springframework.cloud.fn.supplier.ftp.FtpSupplierProperties, \ org.springframework.cloud.fn.common.ftp.FtpSessionFactoryProperties, \ - org.springframework.cloud.fn.common.file.FileConsumerProperties + org.springframework.cloud.fn.common.file.FileConsumerProperties, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Gemfire, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Redis, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$DynamoDb, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Jdbc, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Zookeeper, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Mongo \ diff --git a/applications/source/s3-source/README.adoc b/applications/source/s3-source/README.adoc index d2415188..e0264ea5 100644 --- a/applications/source/s3-source/README.adoc +++ b/applications/source/s3-source/README.adoc @@ -15,6 +15,9 @@ When using `--mode=lines`, you can also provide the additional option `--withMar If set to `true`, the underlying `FileSplitter` will emit additional _start-of-file_ and _end-of-file_ marker messages before and after the actual data. The payload of these 2 additional marker messages is of type `FileSplitter.FileMarker`. The option `withMarkers` defaults to `false` if not explicitly set. +See also link:../../../functions/common/metadata-store-common/README.adoc[MetadataStore] options for possible shared persistent store configuration used to prevent duplicate messages on restart. + + === mode = lines ==== Headers: @@ -53,6 +56,22 @@ The **$$s3$$** $$source$$ has the following options: $$file.consumer.markers-json$$:: $$When 'fileMarkers == true', specify if they should be produced as FileSplitter.FileMarker objects or JSON.$$ *($$Boolean$$, default: `$$true$$`)* $$file.consumer.mode$$:: $$The FileReadingMode to use for file reading sources. Values are 'ref' - The File object, 'lines' - a message per line, or 'contents' - the contents as bytes.$$ *($$FileReadingMode$$, default: `$$$$`, possible values: `ref`,`lines`,`contents`)* $$file.consumer.with-markers$$:: $$Set to true to emit start of file/end of file marker messages before/after the data. Only valid with FileReadingMode 'lines'.$$ *($$Boolean$$, default: `$$$$`)* +$$metadata.store.dynamo-db.create-delay$$:: $$Delay between create table retries.$$ *($$Integer$$, default: `$$1$$`)* +$$metadata.store.dynamo-db.create-retries$$:: $$Retry number for create table request.$$ *($$Integer$$, default: `$$25$$`)* +$$metadata.store.dynamo-db.read-capacity$$:: $$Read capacity on the table.$$ *($$Long$$, default: `$$1$$`)* +$$metadata.store.dynamo-db.table$$:: $$Table name for metadata.$$ *($$String$$, default: `$$$$`)* +$$metadata.store.dynamo-db.time-to-live$$:: $$TTL for table entries.$$ *($$Integer$$, default: `$$$$`)* +$$metadata.store.dynamo-db.write-capacity$$:: $$Write capacity on the table.$$ *($$Long$$, default: `$$1$$`)* +$$metadata.store.gemfire.region$$:: $$Gemfire region name for metadata.$$ *($$String$$, default: `$$$$`)* +$$metadata.store.jdbc.region$$:: $$Unique grouping identifier for messages persisted with this store.$$ *($$String$$, default: `$$DEFAULT$$`)* +$$metadata.store.jdbc.table-prefix$$:: $$Prefix for the custom table name.$$ *($$String$$, default: `$$$$`)* +$$metadata.store.mongo-db.collection$$:: $$MongoDB collection name for metadata.$$ *($$String$$, default: `$$metadataStore$$`)* +$$metadata.store.redis.key$$:: $$Redis key for metadata.$$ *($$String$$, default: `$$$$`)* +$$metadata.store.type$$:: $$Indicates the type of metadata store to configure (default is 'memory'). You must include the corresponding Spring Integration dependency to use a persistent store.$$ *($$StoreType$$, default: `$$$$`, possible values: `mongodb`,`gemfire`,`redis`,`dynamodb`,`jdbc`,`zookeeper`,`hazelcast`,`memory`)* +$$metadata.store.zookeeper.connect-string$$:: $$Zookeeper connect string in form HOST:PORT.$$ *($$String$$, default: `$$127.0.0.1:2181$$`)* +$$metadata.store.zookeeper.encoding$$:: $$Encoding to use when storing data in Zookeeper.$$ *($$Charset$$, default: `$$UTF-8$$`)* +$$metadata.store.zookeeper.retry-interval$$:: $$Retry interval for Zookeeper operations in milliseconds.$$ *($$Integer$$, default: `$$1000$$`)* +$$metadata.store.zookeeper.root$$:: $$Root node - store entries are children of this node.$$ *($$String$$, default: `$$/SpringIntegration-MetadataStore$$`)* $$s3.common.endpoint-url$$:: $$Optional endpoint url to connect to s3 compatible storage.$$ *($$String$$, default: `$$$$`)* $$s3.common.path-style-access$$:: $$Use path style access.$$ *($$Boolean$$, default: `$$false$$`)* $$s3.supplier.auto-create-local-dir$$:: $$Create or not the local directory.$$ *($$Boolean$$, default: `$$true$$`)* diff --git a/applications/source/s3-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/source/s3-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties index d87af164..d39ded0c 100644 --- a/applications/source/s3-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties +++ b/applications/source/s3-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -1,3 +1,10 @@ configuration-properties.classes=org.springframework.cloud.fn.supplier.s3.AwsS3SupplierProperties,\ org.springframework.cloud.fn.common.file.FileConsumerProperties,\ - org.springframework.cloud.fn.common.aws.s3.AmazonS3Properties + org.springframework.cloud.fn.common.aws.s3.AmazonS3Properties, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Gemfire, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Redis, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$DynamoDb, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Jdbc, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Zookeeper, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Mongo \ diff --git a/applications/source/s3-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties b/applications/source/s3-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties index d87af164..d39ded0c 100644 --- a/applications/source/s3-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties +++ b/applications/source/s3-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties @@ -1,3 +1,10 @@ configuration-properties.classes=org.springframework.cloud.fn.supplier.s3.AwsS3SupplierProperties,\ org.springframework.cloud.fn.common.file.FileConsumerProperties,\ - org.springframework.cloud.fn.common.aws.s3.AmazonS3Properties + org.springframework.cloud.fn.common.aws.s3.AmazonS3Properties, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Gemfire, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Redis, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$DynamoDb, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Jdbc, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Zookeeper, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Mongo \ diff --git a/applications/source/sftp-source/README.adoc b/applications/source/sftp-source/README.adoc index cdf87bae..a1759f51 100644 --- a/applications/source/sftp-source/README.adoc +++ b/applications/source/sftp-source/README.adoc @@ -15,6 +15,9 @@ The payload of these 2 additional marker messages is of type `FileSplitter.FileM See link:../../../functions/supplier/sftp-supplier/README.adoc[`sftp-supplier`] for advanced configuration options. +See also link:../../../functions/common/metadata-store-common/README.adoc[MetadataStore] options for possible shared persistent store configuration used to prevent duplicate messages on restart. + + == Input N/A (Fetches files from an SFTP server). @@ -82,6 +85,22 @@ The **$$ftp$$** $$source$$ has the following options: $$file.consumer.markers-json$$:: $$When 'fileMarkers == true', specify if they should be produced as FileSplitter.FileMarker objects or JSON.$$ *($$Boolean$$, default: `$$true$$`)* $$file.consumer.mode$$:: $$The FileReadingMode to use for file reading sources. Values are 'ref' - The File object, 'lines' - a message per line, or 'contents' - the contents as bytes.$$ *($$FileReadingMode$$, default: `$$$$`, possible values: `ref`,`lines`,`contents`)* $$file.consumer.with-markers$$:: $$Set to true to emit start of file/end of file marker messages before/after the data. Only valid with FileReadingMode 'lines'.$$ *($$Boolean$$, default: `$$$$`)* +$$metadata.store.dynamo-db.create-delay$$:: $$Delay between create table retries.$$ *($$Integer$$, default: `$$1$$`)* +$$metadata.store.dynamo-db.create-retries$$:: $$Retry number for create table request.$$ *($$Integer$$, default: `$$25$$`)* +$$metadata.store.dynamo-db.read-capacity$$:: $$Read capacity on the table.$$ *($$Long$$, default: `$$1$$`)* +$$metadata.store.dynamo-db.table$$:: $$Table name for metadata.$$ *($$String$$, default: `$$$$`)* +$$metadata.store.dynamo-db.time-to-live$$:: $$TTL for table entries.$$ *($$Integer$$, default: `$$$$`)* +$$metadata.store.dynamo-db.write-capacity$$:: $$Write capacity on the table.$$ *($$Long$$, default: `$$1$$`)* +$$metadata.store.gemfire.region$$:: $$Gemfire region name for metadata.$$ *($$String$$, default: `$$$$`)* +$$metadata.store.jdbc.region$$:: $$Unique grouping identifier for messages persisted with this store.$$ *($$String$$, default: `$$DEFAULT$$`)* +$$metadata.store.jdbc.table-prefix$$:: $$Prefix for the custom table name.$$ *($$String$$, default: `$$$$`)* +$$metadata.store.mongo-db.collection$$:: $$MongoDB collection name for metadata.$$ *($$String$$, default: `$$metadataStore$$`)* +$$metadata.store.redis.key$$:: $$Redis key for metadata.$$ *($$String$$, default: `$$$$`)* +$$metadata.store.type$$:: $$Indicates the type of metadata store to configure (default is 'memory'). You must include the corresponding Spring Integration dependency to use a persistent store.$$ *($$StoreType$$, default: `$$$$`, possible values: `mongodb`,`gemfire`,`redis`,`dynamodb`,`jdbc`,`zookeeper`,`hazelcast`,`memory`)* +$$metadata.store.zookeeper.connect-string$$:: $$Zookeeper connect string in form HOST:PORT.$$ *($$String$$, default: `$$127.0.0.1:2181$$`)* +$$metadata.store.zookeeper.encoding$$:: $$Encoding to use when storing data in Zookeeper.$$ *($$Charset$$, default: `$$UTF-8$$`)* +$$metadata.store.zookeeper.retry-interval$$:: $$Retry interval for Zookeeper operations in milliseconds.$$ *($$Integer$$, default: `$$1000$$`)* +$$metadata.store.zookeeper.root$$:: $$Root node - store entries are children of this node.$$ *($$String$$, default: `$$/SpringIntegration-MetadataStore$$`)* $$sftp.supplier.auto-create-local-dir$$:: $$Set to true to create the local directory if it does not exist.$$ *($$Boolean$$, default: `$$true$$`)* $$sftp.supplier.delay-when-empty$$:: $$Duration of delay when no new files are detected.$$ *($$Duration$$, default: `$$1s$$`)* $$sftp.supplier.delete-remote-files$$:: $$Set to true to delete remote files after successful transfer.$$ *($$Boolean$$, default: `$$false$$`)* diff --git a/applications/source/sftp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/source/sftp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties index 2517cd6b..a6fa7926 100644 --- a/applications/source/sftp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties +++ b/applications/source/sftp-source/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties @@ -1,3 +1,10 @@ configuration-properties.classes=org.springframework.cloud.fn.supplier.sftp.SftpSupplierProperties, \ org.springframework.cloud.fn.supplier.sftp.SftpSupplierProperties$Factory,\ - org.springframework.cloud.fn.common.file.FileConsumerProperties + org.springframework.cloud.fn.common.file.FileConsumerProperties, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Gemfire, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Redis, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$DynamoDb, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Jdbc, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Zookeeper, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Mongo \ diff --git a/applications/source/sftp-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties b/applications/source/sftp-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties index 2517cd6b..a6fa7926 100644 --- a/applications/source/sftp-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties +++ b/applications/source/sftp-source/src/main/resources/META-INF/dataflow-configuration-metadata.properties @@ -1,3 +1,10 @@ configuration-properties.classes=org.springframework.cloud.fn.supplier.sftp.SftpSupplierProperties, \ org.springframework.cloud.fn.supplier.sftp.SftpSupplierProperties$Factory,\ - org.springframework.cloud.fn.common.file.FileConsumerProperties + org.springframework.cloud.fn.common.file.FileConsumerProperties, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Gemfire, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Redis, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$DynamoDb, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Jdbc, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Zookeeper, \ + org.springframework.cloud.fn.common.metadata.store.MetadataStoreProperties$Mongo \ diff --git a/applications/source/time-source/src/test/java/org/springframework/cloud/stream/app/source/time/TimeSourceTests.java b/applications/source/time-source/src/test/java/org/springframework/cloud/stream/app/source/time/TimeSourceTests.java index 1ccda3c6..6ccfaf47 100644 --- a/applications/source/time-source/src/test/java/org/springframework/cloud/stream/app/source/time/TimeSourceTests.java +++ b/applications/source/time-source/src/test/java/org/springframework/cloud/stream/app/source/time/TimeSourceTests.java @@ -77,7 +77,6 @@ public class TimeSourceTests { OutputDestination target = context.getBean(OutputDestination.class); Message sourceMessage = target.receive(10000); final String actual = new String(sourceMessage.getPayload()); - System.out.println(actual); assertThat(((int) sourceMessage.getHeaders().get("seconds")) % 2).isZero(); } } diff --git a/applications/stream-applications-core/common/stream-applications-micrometer-common/pom.xml b/applications/stream-applications-core/common/stream-applications-micrometer-common/pom.xml index 55765635..ff15c2d2 100644 --- a/applications/stream-applications-core/common/stream-applications-micrometer-common/pom.xml +++ b/applications/stream-applications-core/common/stream-applications-micrometer-common/pom.xml @@ -8,9 +8,27 @@ 3.0.0-SNAPSHOT ../.. + + 1.49 + + 4.0.0 stream-applications-micrometer-common + + + + maven-surefire-plugin + + + -javaagent:"${settings.localRepository}"/org/jmockit/jmockit/${jmockit.version}/jmockit-${jmockit.version}.jar + + + + + + + io.micrometer @@ -25,7 +43,18 @@ io.pivotal.cfenv java-cfenv-test-support test - 2.1.1.RELEASE + 2.2.2.RELEASE + + + org.jmockit + jmockit + + + + + org.jmockit + jmockit + ${jmockit.version} org.springframework.boot diff --git a/applications/stream-applications-core/common/stream-applications-micrometer-common/src/test/java/org/springframework/cloud/stream/app/micrometer/common/AbstractMicrometerTagTest.java b/applications/stream-applications-core/common/stream-applications-micrometer-common/src/test/java/org/springframework/cloud/stream/app/micrometer/common/AbstractMicrometerTagTest.java index 62f0f3c5..471dbcbd 100644 --- a/applications/stream-applications-core/common/stream-applications-micrometer-common/src/test/java/org/springframework/cloud/stream/app/micrometer/common/AbstractMicrometerTagTest.java +++ b/applications/stream-applications-core/common/stream-applications-micrometer-common/src/test/java/org/springframework/cloud/stream/app/micrometer/common/AbstractMicrometerTagTest.java @@ -23,7 +23,7 @@ import io.micrometer.core.instrument.Clock; import io.micrometer.core.instrument.Meter; import io.micrometer.core.instrument.simple.SimpleConfig; import io.micrometer.core.instrument.simple.SimpleMeterRegistry; -import io.pivotal.cfenv.test.CfEnvTestUtils; +import io.pivotal.cfenv.test.AbstractCfEnvTests; import org.junit.Before; import org.junit.BeforeClass; import org.junit.runner.RunWith; @@ -61,11 +61,11 @@ public class AbstractMicrometerTagTest { protected Meter meter; @BeforeClass - public static void setup() throws IOException { + public static void mockVcapServices() throws IOException { String serviceJson = StreamUtils.copyToString(new DefaultResourceLoader().getResource( "classpath:/org/springframework/cloud/stream/app/micrometer/common/pcf-scs-info.json") .getInputStream(), Charset.forName("UTF-8")); - CfEnvTestUtils.mockVcapServicesFromString(serviceJson); + new Vcap().mockServices(serviceJson); } @Before @@ -95,3 +95,9 @@ public class AbstractMicrometerTagTest { } } } + +class Vcap extends AbstractCfEnvTests { + public void mockServices(String json) { + super.mockVcapServices(json); + } +} diff --git a/functions/common/metadata-store-common/README.adoc b/functions/common/metadata-store-common/README.adoc index d1a32b96..1adaa540 100644 --- a/functions/common/metadata-store-common/README.adoc +++ b/functions/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/functions/common/metadata-store-common/pom.xml b/functions/common/metadata-store-common/pom.xml index 7bd5cd07..dddcd4c5 100644 --- a/functions/common/metadata-store-common/pom.xml +++ b/functions/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/functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java b/functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java index e185d9e3..d3e58ee3 100644 --- a/functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreAutoConfiguration.java +++ b/functions/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/functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java b/functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java index 39b898b1..ac030cd7 100644 --- a/functions/common/metadata-store-common/src/main/java/org/springframework/cloud/fn/common/metadata/store/MetadataStoreProperties.java +++ b/functions/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/functions/supplier/s3-supplier/pom.xml b/functions/supplier/s3-supplier/pom.xml index 045dbd58..54203f31 100644 --- a/functions/supplier/s3-supplier/pom.xml +++ b/functions/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/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 3e9ca2cf..ed01b4c0 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 @@ -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/functions/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierProperties.java b/functions/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierProperties.java index 5bf24501..7b50ede1 100644 --- a/functions/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierProperties.java +++ b/functions/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/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 161c7a2e..ea42103a 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 @@ -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/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 new file mode 100644 index 00000000..207c3fd6 --- /dev/null +++ b/functions/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/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java b/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java index a9242a38..5e37ce0a 100644 --- a/functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java +++ b/functions/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/functions/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java b/functions/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java index 3dc9c71a..6020a261 100644 --- a/functions/supplier/sftp-supplier/src/test/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierApplicationTests.java +++ b/functions/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()); })