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 8b50d99a75..61d775d217 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 @@ -23,6 +23,7 @@ import org.apache.streampipes.extensions.api.migration.IModelMigrator; import org.apache.streampipes.extensions.api.pe.IStreamPipesPipelineElement; import org.apache.streampipes.processors.transformation.jvm.migrations.StaticMetadataEnrichmentProcessorMigrationV1; +import org.apache.streampipes.processors.transformation.jvm.migrations.StaticMetadataEnrichmentProcessorMigrationV2; import org.apache.streampipes.processors.transformation.jvm.processor.array.count.CountArrayProcessor; import org.apache.streampipes.processors.transformation.jvm.processor.array.split.SplitArrayProcessor; import org.apache.streampipes.processors.transformation.jvm.processor.booloperator.counter.BooleanCounterProcessor; @@ -94,6 +95,8 @@ public List> pipelineElements() { @Override public List> migrators() { - return List.of(new StaticMetadataEnrichmentProcessorMigrationV1()); + return List.of( + new StaticMetadataEnrichmentProcessorMigrationV1(), + new StaticMetadataEnrichmentProcessorMigrationV2()); } } diff --git a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/migrations/StaticMetadataEnrichmentProcessorMigrationV2.java b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/migrations/StaticMetadataEnrichmentProcessorMigrationV2.java new file mode 100644 index 0000000000..93599063ef --- /dev/null +++ b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/migrations/StaticMetadataEnrichmentProcessorMigrationV2.java @@ -0,0 +1,85 @@ +/* + * 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.migrations; + +import org.apache.streampipes.extensions.api.extractor.IDataProcessorParameterExtractor; +import org.apache.streampipes.extensions.api.migration.IDataProcessorMigrator; +import org.apache.streampipes.model.extensions.svcdiscovery.SpServiceTagPrefix; +import org.apache.streampipes.model.graph.DataProcessorInvocation; +import org.apache.streampipes.model.migration.MigrationResult; +import org.apache.streampipes.model.migration.ModelMigratorConfig; +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.processors.transformation.jvm.processor.staticmetadata.StaticMetaDataEnrichmentProcessor; +import org.apache.streampipes.sdk.StaticProperties; +import org.apache.streampipes.sdk.helpers.Labels; + +import java.util.List; + +import static org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata.StaticMetaDataEnrichmentProcessor.STATIC_METADATA_INPUT_DESCRIPTION; +import static org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata.StaticMetaDataEnrichmentProcessor.STATIC_METADATA_INPUT_LABEL; + +public class StaticMetadataEnrichmentProcessorMigrationV2 implements IDataProcessorMigrator { + @Override + public ModelMigratorConfig config() { + return new ModelMigratorConfig( + StaticMetaDataEnrichmentProcessor.ID, + SpServiceTagPrefix.DATA_PROCESSOR, + 1, + 2 + ); + } + + @Override + public MigrationResult migrate(DataProcessorInvocation element, + IDataProcessorParameterExtractor extractor) + throws RuntimeException { + element.getStaticProperties() + .stream() + .filter(sp -> sp.getInternalName().equals(StaticMetaDataEnrichmentProcessor.STATIC_METADATA_INPUT)) + .map(sp -> (CollectionStaticProperty) sp) + .forEach(sp -> { + addLabelAndDescriptionConfig((StaticPropertyGroup) sp.getStaticPropertyTemplate()); + sp.getMembers().forEach(member -> { + addLabelAndDescriptionConfig((StaticPropertyGroup) member); + }); + }); + return MigrationResult.success(element); + } + + private void addLabelAndDescriptionConfig(StaticPropertyGroup group) { + group.setHorizontalRendering(false); + group.getStaticProperties().addAll( + List.of( + makeFreeTextProperty(STATIC_METADATA_INPUT_LABEL), + makeFreeTextProperty(STATIC_METADATA_INPUT_DESCRIPTION) + ) + ); + } + + private StaticProperty makeFreeTextProperty(String id) { + var sp = StaticProperties.stringFreeTextProperty( + Labels.withId( + id)); + sp.setOptional(true); + sp.setValue(""); + return sp; + } +} 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 index d3ffdee874..ce4df542e6 100644 --- 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 @@ -21,6 +21,8 @@ public record StaticMetaDataConfiguration( String runtimeName, String value, - String dataType + String dataType, + String label, + String description ) { } 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 e60947db72..649896117e 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 @@ -53,14 +53,16 @@ public class StaticMetaDataEnrichmentProcessor implements IStreamPipesDataProcessor, - ResolvesContainerProvidedOutputStrategy { + ResolvesContainerProvidedOutputStrategy { public static final String ID = "org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata"; - protected static final String STATIC_METADATA_INPUT = "static-metadata-input"; + public 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"; + public static final String STATIC_METADATA_INPUT_LABEL = "static-metadata-input-label"; + public static final String STATIC_METADATA_INPUT_DESCRIPTION = "static-metadata-input-description"; protected static final String OPTION_BOOL = "Bool"; protected static final String OPTION_STRING = "String"; @@ -75,42 +77,51 @@ public IDataProcessorConfiguration declareConfig() { return DataProcessorConfiguration.create( StaticMetaDataEnrichmentProcessor::new, ProcessingElementBuilder.create( - ID, - 1 - ) - .category( - DataProcessorType.ENRICH) - .withLocales( - Locales.EN) - .withAssets( - ExtensionAssetType.DOCUMENTATION, - ExtensionAssetType.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() + ID, + 2 + ) + .category( + DataProcessorType.ENRICH) + .withLocales( + Locales.EN) + .withAssets( + ExtensionAssetType.DOCUMENTATION, + ExtensionAssetType.ICON + ) + .requiredStaticProperty( + StaticProperties.collection( + Labels.withId( + STATIC_METADATA_INPUT), + false, + 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 + ) + ), + StaticProperties.stringFreeTextProperty( + Labels.withId( + STATIC_METADATA_INPUT_LABEL)), + StaticProperties.stringFreeTextProperty( + Labels.withId( + STATIC_METADATA_INPUT_DESCRIPTION)) + ) + ) + .requiredStream( + StreamRequirementsBuilder.any()) + .outputStrategy( + OutputStrategies.customTransformation()) + .build() ); } @@ -123,8 +134,8 @@ public EventSchema resolveOutputStrategy( var metaDataConfigurations = getMetaDataConfigurations(parameterExtractor); var eventSchema = processingElement.getInputStreams() - .get(0) - .getEventSchema(); + .get(0) + .getEventSchema(); addMetaDataConfigurationPropertiesToEventSchema(metaDataConfigurations, eventSchema); @@ -185,7 +196,9 @@ private StaticMetaDataConfiguration getMetaDataConfiguration(StaticPropertyExtra 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); + var label = memberExtractor.textParameter(STATIC_METADATA_INPUT_LABEL); + var description = memberExtractor.textParameter(STATIC_METADATA_INPUT_DESCRIPTION); + return new StaticMetaDataConfiguration(runtimeName, value, dataType, label, description); } protected Object castValueOfMetaDataConfiguration(StaticMetaDataConfiguration staticMetaDataConfiguration) { @@ -214,7 +227,7 @@ private void addMetaDataConfigurationPropertiesToEventSchema( for (StaticMetaDataConfiguration metaDataConfiguration : metaDataConfigurations) { var metaDataEventProperty = getMetaDataEventProperty(metaDataConfiguration); eventSchema.getEventProperties() - .add(metaDataEventProperty); + .add(metaDataEventProperty); } } @@ -226,6 +239,8 @@ private EventPropertyPrimitive getMetaDataEventProperty( transformToStreamPipesDataType(metaDataConfiguration.dataType()), metaDataConfiguration.runtimeName() ) + .label(metaDataConfiguration.label()) + .description(metaDataConfiguration.description()) .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/strings.en b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/resources/org.apache.streampipes.processors.transformation.jvm.processor.staticmetadata/strings.en index 2c2826d82e..b4b5b747fa 100644 --- 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 @@ -30,3 +30,8 @@ static-metadata-input-value.description=This is used for the runtime value in th static-metadata-input-datatype.title=Data Type static-metadata-input-datatype.description=Select the data type of the value +static-metadata-input-label.title=Label +static-metadata-input-label.description=An optional short label which describes the field + +static-metadata-input-description.title=Description +static-metadata-input-description.description=An optional description of the field diff --git a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGenerator.java b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGenerator.java index 77551acb4a..bf2e38aae6 100644 --- a/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGenerator.java +++ b/streampipes-pipeline-management/src/main/java/org/apache/streampipes/manager/matching/output/CustomTransformOutputSchemaGenerator.java @@ -33,6 +33,7 @@ import org.apache.http.entity.ContentType; import java.io.IOException; +import java.nio.charset.StandardCharsets; public class CustomTransformOutputSchemaGenerator extends OutputSchemaGenerator { @@ -79,7 +80,7 @@ private EventSchema makeRequest() { } private EventSchema handleResponse(Response httpResp) throws JsonSyntaxException, IOException { - String resp = httpResp.returnContent().asString(); + String resp = httpResp.returnContent().asString(StandardCharsets.UTF_8); return JacksonSerializer .getObjectMapper()