This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new ac83d2b6a feat(#2350): New processor staticmetadataenrichment (#2366)
ac83d2b6a is described below
commit ac83d2b6a66870e87ab7483857b701d20dfdeb8d
Author: Philipp Zehnder <[email protected]>
AuthorDate: Sat Dec 23 09:10:19 2023 +0100
feat(#2350): New processor staticmetadataenrichment (#2366)
* fix(#2350): Add resources for new processor
* feat(#2350): Add documentation for new processing element
* feat(#2350): Add application logic for new processor
StaticMetaDataEnrichment
---
.../pom.xml | 12 ++
.../jvm/TransformationExtensionModuleExport.java | 2 +
.../StaticMetaDataConfiguration.java | 26 +++
.../StaticMetaDataEnrichmentProcessor.java | 233 +++++++++++++++++++++
.../documentation.md | 68 ++++++
.../icon.png | Bin 0 -> 30181 bytes
.../strings.en | 32 +++
.../StaticMetaDataEnrichmentProcessorTest.java | 118 +++++++++++
8 files changed, 491 insertions(+)
diff --git
a/streampipes-extensions/streampipes-processors-transformation-jvm/pom.xml
b/streampipes-extensions/streampipes-processors-transformation-jvm/pom.xml
index 8e261686f..9ea4d6356 100644
--- a/streampipes-extensions/streampipes-processors-transformation-jvm/pom.xml
+++ b/streampipes-extensions/streampipes-processors-transformation-jvm/pom.xml
@@ -53,6 +53,18 @@
<version>0.95.0-SNAPSHOT</version>
<scope>test</scope>
</dependency>
+
+ <!-- Test dependencies -->
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<build>
diff --git
a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/TransformationExtensionModuleExport.java
b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/TransformationExtensionModuleExport.java
index 3afa98d2f..fbd6f3180 100644
---
a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/TransformationExtensionModuleExport.java
+++
b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/TransformationExtensionModuleExport.java
@@ -39,6 +39,7 @@ import
org.apache.streampipes.processors.transformation.jvm.processor.mapper.Fie
import
org.apache.streampipes.processors.transformation.jvm.processor.measurementconverter.MeasurementUnitConverterProcessor;
import
org.apache.streampipes.processors.transformation.jvm.processor.round.RoundProcessor;
import
org.apache.streampipes.processors.transformation.jvm.processor.state.labeler.number.NumberLabelerProcessor;
+import
org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata.StaticMetaDataEnrichmentProcessor;
import
org.apache.streampipes.processors.transformation.jvm.processor.stringoperator.counter.StringCounterProcessor;
import
org.apache.streampipes.processors.transformation.jvm.processor.stringoperator.state.StringToStateProcessor;
import
org.apache.streampipes.processors.transformation.jvm.processor.stringoperator.timer.StringTimerProcessor;
@@ -77,6 +78,7 @@ public class TransformationExtensionModuleExport implements
IExtensionModuleExpo
new MeasurementUnitConverterProcessor(),
new TaskDurationProcessor(),
new TransformToBooleanProcessor(),
+ new StaticMetaDataEnrichmentProcessor(),
new StringTimerProcessor(),
new SignalEdgeFilterProcessor(),
new BooleanToStateProcessor(),
diff --git
a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/staticmetadata/StaticMetaDataConfiguration.java
b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/staticmetadata/StaticMetaDataConfiguration.java
new file mode 100644
index 000000000..d3ffdee87
--- /dev/null
+++
b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/staticmetadata/StaticMetaDataConfiguration.java
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.streampipes.processors.transformation.jvm.processor.staticmetadata;
+
+public record StaticMetaDataConfiguration(
+ String runtimeName,
+ String value,
+ String dataType
+) {
+}
diff --git
a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/staticmetadata/StaticMetaDataEnrichmentProcessor.java
b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/staticmetadata/StaticMetaDataEnrichmentProcessor.java
new file mode 100644
index 000000000..72eefe4ad
--- /dev/null
+++
b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/staticmetadata/StaticMetaDataEnrichmentProcessor.java
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.streampipes.processors.transformation.jvm.processor.staticmetadata;
+
+import
org.apache.streampipes.extensions.api.extractor.IDataProcessorParameterExtractor;
+import org.apache.streampipes.extensions.api.pe.IStreamPipesDataProcessor;
+import
org.apache.streampipes.extensions.api.pe.config.IDataProcessorConfiguration;
+import
org.apache.streampipes.extensions.api.pe.context.EventProcessorRuntimeContext;
+import org.apache.streampipes.extensions.api.pe.param.IDataProcessorParameters;
+import org.apache.streampipes.extensions.api.pe.routing.SpOutputCollector;
+import
org.apache.streampipes.extensions.api.runtime.ResolvesContainerProvidedOutputStrategy;
+import org.apache.streampipes.model.DataProcessorType;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
+import org.apache.streampipes.model.schema.EventSchema;
+import org.apache.streampipes.model.schema.PropertyScope;
+import org.apache.streampipes.model.staticproperty.CollectionStaticProperty;
+import org.apache.streampipes.model.staticproperty.StaticProperty;
+import org.apache.streampipes.model.staticproperty.StaticPropertyGroup;
+import org.apache.streampipes.sdk.StaticProperties;
+import org.apache.streampipes.sdk.builder.PrimitivePropertyBuilder;
+import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
+import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+import org.apache.streampipes.sdk.builder.processor.DataProcessorConfiguration;
+import
org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
+import org.apache.streampipes.sdk.extractor.StaticPropertyExtractor;
+import org.apache.streampipes.sdk.helpers.Labels;
+import org.apache.streampipes.sdk.helpers.Locales;
+import org.apache.streampipes.sdk.helpers.Options;
+import org.apache.streampipes.sdk.helpers.OutputStrategies;
+import org.apache.streampipes.sdk.utils.Assets;
+import org.apache.streampipes.sdk.utils.Datatypes;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class StaticMetaDataEnrichmentProcessor
+ implements IStreamPipesDataProcessor,
+
ResolvesContainerProvidedOutputStrategy<DataProcessorInvocation,
ProcessingElementParameterExtractor> {
+
+ protected static final String STATIC_METADATA_INPUT =
"static-metadata-input";
+ protected static final String STATIC_METADATA_INPUT_RUNTIME_NAME =
"static-metadata-input-runtime-name";
+ protected static final String STATIC_METADATA_INPUT_VALUE =
"static-metadata-input-value";
+ protected static final String STATIC_METADATA_INPUT_DATATYPE =
"static-metadata-input-datatype";
+
+ protected static final String OPTION_BOOL = "Bool";
+ protected static final String OPTION_STRING = "String";
+ protected static final String OPTION_FLOAT = "Float";
+ protected static final String OPTION_INTEGER = "Integer";
+
+
+ private List<StaticMetaDataConfiguration> staticMetaDataConfigurations = new
ArrayList<>();
+
+ @Override
+ public IDataProcessorConfiguration declareConfig() {
+ return DataProcessorConfiguration.create(
+ StaticMetaDataEnrichmentProcessor::new,
+ ProcessingElementBuilder.create(
+
"org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata",
+ 1
+ )
+ .category(
+ DataProcessorType.ENRICH)
+ .withLocales(
+ Locales.EN)
+ .withAssets(
+ Assets.DOCUMENTATION,
+ Assets.ICON
+ )
+ .requiredCollection(
+ Labels.withId(
+ STATIC_METADATA_INPUT),
+ StaticProperties.stringFreeTextProperty(
+ Labels.withId(
+
STATIC_METADATA_INPUT_RUNTIME_NAME)),
+ StaticProperties.stringFreeTextProperty(
+ Labels.withId(
+ STATIC_METADATA_INPUT_VALUE)),
+ StaticProperties.singleValueSelection(
+ Labels.withId(
+ STATIC_METADATA_INPUT_DATATYPE),
+ Options.from(
+ OPTION_BOOL,
+ OPTION_STRING,
+ OPTION_FLOAT,
+ OPTION_INTEGER
+ )
+ )
+ )
+ .requiredStream(
+ StreamRequirementsBuilder.any())
+ .outputStrategy(
+ OutputStrategies.customTransformation())
+ .build()
+ );
+ }
+
+ @Override
+ public EventSchema resolveOutputStrategy(
+ DataProcessorInvocation processingElement,
+ ProcessingElementParameterExtractor parameterExtractor
+ ) {
+
+ var metaDataConfigurations = getMetaDataConfigurations(parameterExtractor);
+
+ var eventSchema = processingElement.getInputStreams()
+ .get(0)
+ .getEventSchema();
+
+ addMetaDataConfigurationPropertiesToEventSchema(metaDataConfigurations,
eventSchema);
+
+ return eventSchema;
+ }
+
+
+ @Override
+ public void onPipelineStarted(
+ IDataProcessorParameters params,
+ SpOutputCollector collector,
+ EventProcessorRuntimeContext runtimeContext
+ ) {
+ this.staticMetaDataConfigurations =
getMetaDataConfigurations(params.extractor());
+ }
+
+
+ @Override
+ public void onEvent(Event event, SpOutputCollector collector) {
+ appendMetadataToEvent(event);
+
+ collector.collect(event);
+ }
+
+ private void appendMetadataToEvent(Event event) {
+ for (StaticMetaDataConfiguration staticMetaDataConfiguration :
this.staticMetaDataConfigurations) {
+ var value =
castValueOfMetaDataConfiguration(staticMetaDataConfiguration);
+ event.addField(staticMetaDataConfiguration.runtimeName(), value);
+ }
+ }
+
+ @Override
+ public void onPipelineStopped() {
+
+ }
+
+ private List<StaticMetaDataConfiguration>
getMetaDataConfigurations(IDataProcessorParameterExtractor extractor) {
+ List<StaticMetaDataConfiguration> configurations = new ArrayList<>();
+ var csp = (CollectionStaticProperty)
extractor.getStaticPropertyByName(STATIC_METADATA_INPUT);
+
+ for (StaticProperty member : csp.getMembers()) {
+ var memberExtractor = getMemberExtractor(member);
+
+ var metadataConfiguration = getMetaDataConfiguration(memberExtractor);
+
+ configurations.add(metadataConfiguration);
+ }
+ return configurations;
+ }
+
+ private StaticPropertyExtractor getMemberExtractor(StaticProperty member) {
+ return StaticPropertyExtractor.from(
+ ((StaticPropertyGroup) member).getStaticProperties(), new ArrayList<>()
+ );
+ }
+
+ private StaticMetaDataConfiguration
getMetaDataConfiguration(StaticPropertyExtractor memberExtractor) {
+ var runtimeName =
memberExtractor.textParameter(STATIC_METADATA_INPUT_RUNTIME_NAME);
+ var value = memberExtractor.textParameter(STATIC_METADATA_INPUT_VALUE);
+ var dataType =
memberExtractor.selectedSingleValue(STATIC_METADATA_INPUT_DATATYPE,
String.class);
+ return new StaticMetaDataConfiguration(runtimeName, value, dataType);
+ }
+
+ protected Object
castValueOfMetaDataConfiguration(StaticMetaDataConfiguration
staticMetaDataConfiguration) {
+ return switch (staticMetaDataConfiguration.dataType()) {
+ case OPTION_BOOL ->
Boolean.parseBoolean(staticMetaDataConfiguration.value());
+ case OPTION_FLOAT ->
Float.parseFloat(staticMetaDataConfiguration.value());
+ case OPTION_INTEGER ->
Integer.parseInt(staticMetaDataConfiguration.value());
+ default -> staticMetaDataConfiguration.value();
+ };
+ }
+
+ protected Datatypes transformToStreamPipesDataType(String option) {
+ return switch (option) {
+ case StaticMetaDataEnrichmentProcessor.OPTION_BOOL -> Datatypes.Boolean;
+ case StaticMetaDataEnrichmentProcessor.OPTION_STRING -> Datatypes.String;
+ case StaticMetaDataEnrichmentProcessor.OPTION_FLOAT -> Datatypes.Float;
+ case StaticMetaDataEnrichmentProcessor.OPTION_INTEGER ->
Datatypes.Integer;
+ default -> throw new IllegalArgumentException("Invalid option: " +
option);
+ };
+ }
+
+ private void addMetaDataConfigurationPropertiesToEventSchema(
+ List<StaticMetaDataConfiguration> metaDataConfigurations,
+ EventSchema eventSchema
+ ) {
+ for (StaticMetaDataConfiguration metaDataConfiguration :
metaDataConfigurations) {
+ var metaDataEventProperty =
getMetaDataEventProperty(metaDataConfiguration);
+ eventSchema.getEventProperties()
+ .add(metaDataEventProperty);
+ }
+ }
+
+ private EventPropertyPrimitive getMetaDataEventProperty(
+ StaticMetaDataConfiguration metaDataConfiguration
+ ) {
+ return PrimitivePropertyBuilder
+ .create(
+ transformToStreamPipesDataType(metaDataConfiguration.dataType()),
+ metaDataConfiguration.runtimeName()
+ )
+ .scope(PropertyScope.MEASUREMENT_PROPERTY)
+ .build();
+ }
+
+
+}
+
diff --git
a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/resources/org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata/documentation.md
b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/resources/org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata/documentation.md
new file mode 100644
index 000000000..ebfc4a709
--- /dev/null
+++
b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/resources/org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata/documentation.md
@@ -0,0 +1,68 @@
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You 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
+ ~
+ ~ http://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.
+ ~
+ -->
+## Static Metadata Enricher
+
+Enrich a data stream by dynamically adding fields based on user-provided
static metadata configuration.
+
+---
+
+## Description
+
+The Static Metadata Enricher is designed to enrich a data stream by
dynamically adding fields based on user-provided
+metadata configuration. Users can specify static properties, and the processor
will process each event, adding fields
+according to the provided key-value pairs. The output strategy is determined
dynamically based on the provided metadata.
+For added convenience, users also have the option of uploading a CSV file with
metadata information.
+
+### Configuration
+
+For each metadata entry, configure the following three options:
+
+- **Runtime Name:** A unique identifier for the property during runtime.
+- **Value:** The value associated with the property.
+- **Data Type:** The data type of the property value.
+
+#### Using CSV Option
+
+Alternatively, you can utilize the CSV upload feature by creating a CSV file
with the following format:
+
+```
+Runtime Name,Runtime Value,Data Type
+sensorType,Temperature,String
+maxSensorValue,100.0,Float
+minSensorValue,0,Float
+```
+
+## Example
+### Input Event
+
+```json
+{
+ "reading": 25.5
+}
+```
+
+### Output Event
+
+```json
+{
+ "reading": 25.5,
+ "sensorType": "Temperature",
+ "maxSensorValue": 100.0,
+ "minSensorValue": 0.0
+}
+```
diff --git
a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/resources/org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata/icon.png
b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/resources/org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata/icon.png
new file mode 100644
index 000000000..61caf18b7
Binary files /dev/null and
b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/resources/org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata/icon.png
differ
diff --git
a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/resources/org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata/strings.en
b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/resources/org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata/strings.en
new file mode 100644
index 000000000..2c2826d82
--- /dev/null
+++
b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/resources/org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata/strings.en
@@ -0,0 +1,32 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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
+#
+# http://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.
+#
+
+org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata.title=Static
Metadata Enricher
+org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata.description=Enriches
event with metadata
+
+static-metadata-input.title=Metadata Input
+static-metadata-input.description=Input collection of metadata that will be
enriched to the event
+
+static-metadata-input-runtime-name.title=Runtime Name
+static-metadata-input-runtime-name.description=This is used for the runtime
name in the event
+
+static-metadata-input-value.title=Runtime Value
+static-metadata-input-value.description=This is used for the runtime value in
the event
+
+static-metadata-input-datatype.title=Data Type
+static-metadata-input-datatype.description=Select the data type of the value
+
diff --git
a/streampipes-extensions/streampipes-processors-transformation-jvm/src/test/java/org/apache/streampipes/processors/transformation/jvm/processor/staticmetadata/StaticMetaDataEnrichmentProcessorTest.java
b/streampipes-extensions/streampipes-processors-transformation-jvm/src/test/java/org/apache/streampipes/processors/transformation/jvm/processor/staticmetadata/StaticMetaDataEnrichmentProcessorTest.java
new file mode 100644
index 000000000..b03c2c2d6
--- /dev/null
+++
b/streampipes-extensions/streampipes-processors-transformation-jvm/src/test/java/org/apache/streampipes/processors/transformation/jvm/processor/staticmetadata/StaticMetaDataEnrichmentProcessorTest.java
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.streampipes.processors.transformation.jvm.processor.staticmetadata;
+
+import org.apache.streampipes.sdk.utils.Datatypes;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+
+
+public class StaticMetaDataEnrichmentProcessorTest {
+
+ private StaticMetaDataEnrichmentProcessor processor;
+
+ @Before
+ public void setup() {
+ processor = new StaticMetaDataEnrichmentProcessor();
+ }
+
+ @Test
+ public void transformToStreamPipesDataTypeForValidOption() {
+ assertEquals(
+ Datatypes.Boolean,
+
processor.transformToStreamPipesDataType(StaticMetaDataEnrichmentProcessor.OPTION_BOOL)
+ );
+ assertEquals(
+ Datatypes.String,
+
processor.transformToStreamPipesDataType(StaticMetaDataEnrichmentProcessor.OPTION_STRING)
+ );
+ assertEquals(
+ Datatypes.Float,
+
processor.transformToStreamPipesDataType(StaticMetaDataEnrichmentProcessor.OPTION_FLOAT)
+ );
+ assertEquals(
+ Datatypes.Integer,
+
processor.transformToStreamPipesDataType(StaticMetaDataEnrichmentProcessor.OPTION_INTEGER)
+ );
+ }
+
+ @Test
+ public void transformToStreamPipesDataTypeForInvalidOption() {
+ assertThrows(IllegalArgumentException.class, () ->
+ processor.transformToStreamPipesDataType("InvalidOption"));
+ }
+
+ @Test
+ public void castValueOfMetaDataConfigurationBoolean() {
+ var config = getSampleMetaDataConfiguration(
+ "true",
+ StaticMetaDataEnrichmentProcessor.OPTION_BOOL
+ );
+
+ assertEquals(true, processor.castValueOfMetaDataConfiguration(config));
+ }
+
+ @Test
+ public void castValueOfMetaDataConfigurationFloat() {
+ var config = getSampleMetaDataConfiguration(
+ "1.23",
+ StaticMetaDataEnrichmentProcessor.OPTION_FLOAT
+ );
+
+ assertEquals(1.23f, processor.castValueOfMetaDataConfiguration(config));
+ }
+
+ @Test
+ public void castValueOfMetaDataConfigurationInteger() {
+ var config = getSampleMetaDataConfiguration(
+ "123",
+ StaticMetaDataEnrichmentProcessor.OPTION_INTEGER
+ );
+
+ assertEquals(123, processor.castValueOfMetaDataConfiguration(config));
+ }
+
+ @Test
+ public void castValueOfMetaDataConfigurationString() {
+ var config = getSampleMetaDataConfiguration(
+ "test",
+ StaticMetaDataEnrichmentProcessor.OPTION_STRING
+ );
+
+ assertEquals("test", processor.castValueOfMetaDataConfiguration(config));
+ }
+
+ @Test
+ public void castValueOfMetaDataConfigurationDefault() {
+ StaticMetaDataConfiguration config = new
StaticMetaDataConfiguration("runtimeName", "default", "default");
+
+ assertEquals("default",
processor.castValueOfMetaDataConfiguration(config));
+ }
+
+ private static StaticMetaDataConfiguration getSampleMetaDataConfiguration(
+ String value, String dataType
+ ) {
+ return new StaticMetaDataConfiguration("runtimeName", value, dataType);
+ }
+
+}
