This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch 2350-new-processor-staticmetadataenrichment
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/2350-new-processor-staticmetadataenrichment by this push:
new b54d87068 feat(#2350): Add application logic for new processor
StaticMetaDataEnrichment
b54d87068 is described below
commit b54d8706849c868f9fae245e857e5d658f353731
Author: Philipp Zehnder <[email protected]>
AuthorDate: Sat Dec 23 07:35:50 2023 +0100
feat(#2350): Add application logic for new processor
StaticMetaDataEnrichment
---
.../pom.xml | 12 ++
.../StaticMetaDataConfiguration.java | 26 +++
.../StaticMetaDataEnrichmentProcessor.java | 219 ++++++++++++++++-----
.../StaticMetaDataEnrichmentProcessorTest.java | 118 +++++++++++
4 files changed, 325 insertions(+), 50 deletions(-)
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/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
index e503e2b0a..72eefe4ad 100644
---
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
@@ -18,97 +18,216 @@
package
org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata;
-import org.apache.streampipes.commons.exceptions.SpConfigurationException;
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+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.DataProcessorDescription;
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.wrapper.params.compat.ProcessorParams;
-import org.apache.streampipes.wrapper.standalone.StreamPipesDataProcessor;
+import org.apache.streampipes.sdk.utils.Datatypes;
-public class StaticMetaDataEnrichmentProcessor extends StreamPipesDataProcessor
- implements
ResolvesContainerProvidedOutputStrategy<DataProcessorInvocation,
ProcessingElementParameterExtractor> {
+import java.util.ArrayList;
+import java.util.List;
- private static final String STATIC_METADATA_INPUT = "static-metadata-input";
- private static final String STATIC_METADATA_INPUT_RUNTIME_NAME =
"static-metadata-input-runtime-name";
- private static final String STATIC_METADATA_INPUT_VALUE =
"static-metadata-input-value";
- private static final String STATIC_METADATA_INPUT_DATATYPE =
"static-metadata-input-datatype";
+public class StaticMetaDataEnrichmentProcessor
+ implements IStreamPipesDataProcessor,
+
ResolvesContainerProvidedOutputStrategy<DataProcessorInvocation,
ProcessingElementParameterExtractor> {
- private static final String OPTION_BOOL = "Bool";
- private static final String OPTION_STRING = "String";
- private static final String OPTION_FLOAT = "Float";
- private static final String OPTION_INTEGER = "Integer";
+ 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 DataProcessorDescription declareModel() {
- return 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();
+ 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
- ) throws SpConfigurationException {
- return null;
+ ) {
+
+ var metaDataConfigurations = getMetaDataConfigurations(parameterExtractor);
+
+ var eventSchema = processingElement.getInputStreams()
+ .get(0)
+ .getEventSchema();
+
+ addMetaDataConfigurationPropertiesToEventSchema(metaDataConfigurations,
eventSchema);
+
+ return eventSchema;
}
+
@Override
- public void onInvocation(
- ProcessorParams parameters,
- SpOutputCollector spOutputCollector,
+ public void onPipelineStarted(
+ IDataProcessorParameters params,
+ SpOutputCollector collector,
EventProcessorRuntimeContext runtimeContext
- ) throws SpRuntimeException {
-
+ ) {
+ 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 onDetach() {
+ 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/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);
+ }
+
+}
