This is an automated email from the ASF dual-hosted git repository. bossenti pushed a commit to branch dev in repository https://gitbox.apache.org/repos/asf/incubator-streampipes-extensions.git
commit 639f844c0ff31a666e5868823a1ea570d92ae04f Author: bossenti <[email protected]> AuthorDate: Mon Mar 29 19:32:38 2021 +0200 [STREAMPIPES-334] implement changes for Buffer Labeler as well --- .../state/labeler/buffer/StateBufferLabeler.java | 4 +++- .../labeler/buffer/StateBufferLabelerController.java | 19 +++++++++++++++++-- .../labeler/buffer/StateBufferLabelerParameters.java | 7 +++++++ .../strings.en | 4 ++++ 4 files changed, 31 insertions(+), 3 deletions(-) diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabeler.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabeler.java index 6a73ef8..529e728 100644 --- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabeler.java +++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabeler.java @@ -39,6 +39,7 @@ public class StateBufferLabeler implements EventProcessor<StateBufferLabelerPara private String stateProperty; private String stateFilter; private String selectedOperation; + private String labelName; private List<Statement> statements; @Override @@ -51,6 +52,7 @@ public class StateBufferLabeler implements EventProcessor<StateBufferLabelerPara this.stateProperty = stateBufferLabelerParameters.getStateProperty(); this.stateFilter = stateBufferLabelerParameters.getStateFilter(); this.selectedOperation = stateBufferLabelerParameters.getSelectedOperation(); + this.labelName = stateBufferLabelerParameters.getLabelName(); this.statements = StatementUtils.getStatements( stateBufferLabelerParameters.getNumberValues(), @@ -75,7 +77,7 @@ public class StateBufferLabeler implements EventProcessor<StateBufferLabelerPara calculatedValue = Stats.of(values).mean(); } - Event resultEvent = StatementUtils.addLabel(inputEvent, calculatedValue, this.statements, LOG); + Event resultEvent = StatementUtils.addLabel(inputEvent, labelName, calculatedValue, this.statements, LOG); out.collect(resultEvent); } } diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabelerController.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabelerController.java index 6d490bc..3ca5268 100644 --- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabelerController.java +++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabelerController.java @@ -18,13 +18,17 @@ package org.apache.streampipes.processors.transformation.jvm.processor.state.labeler.buffer; +import org.apache.streampipes.commons.exceptions.SpRuntimeException; +import org.apache.streampipes.container.api.ResolvesContainerProvidedOutputStrategy; import org.apache.streampipes.model.graph.DataProcessorDescription; import org.apache.streampipes.model.graph.DataProcessorInvocation; +import org.apache.streampipes.model.schema.EventSchema; import org.apache.streampipes.model.schema.PropertyScope; import org.apache.streampipes.model.staticproperty.FreeTextStaticProperty; import org.apache.streampipes.model.staticproperty.OneOfStaticProperty; import org.apache.streampipes.model.staticproperty.Option; import org.apache.streampipes.model.staticproperty.StaticPropertyGroup; +import org.apache.streampipes.processors.transformation.jvm.processor.state.labeler.LabelerUtils; import org.apache.streampipes.sdk.StaticProperties; import org.apache.streampipes.sdk.builder.ProcessingElementBuilder; import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder; @@ -38,7 +42,7 @@ import org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcess import java.util.List; import java.util.stream.Collectors; -public class StateBufferLabelerController extends StandaloneEventProcessingDeclarer<StateBufferLabelerParameters> { +public class StateBufferLabelerController extends StandaloneEventProcessingDeclarer<StateBufferLabelerParameters> implements ResolvesContainerProvidedOutputStrategy<DataProcessorInvocation, ProcessingElementParameterExtractor> { public static final String STATE_FILTER_ID = "stateFilterId"; public static final String STATE_FIELD_ID = "stateFieldId"; @@ -48,6 +52,7 @@ public class StateBufferLabelerController extends StandaloneEventProcessingDecla public static final String COMPARATOR_ID = "comparatorId"; public static final String NUMBER_VALUE_ID = "numberValueId"; public static final String LABEL_STRING_ID = "labelStringId"; + public static final String LABEL_NAME = "labelName"; public static final String LABEL = "label"; @@ -70,6 +75,7 @@ public class StateBufferLabelerController extends StandaloneEventProcessingDecla Labels.withId(STATE_FIELD_ID), PropertyScope.NONE) .build()) + .requiredTextParameter(Labels.withId(LABEL_NAME)) .requiredTextParameter(Labels.withId(STATE_FILTER_ID)) .requiredSingleValueSelection(Labels.withId(OPERATIONS_ID), Options.from(MINIMUM, MAXIMUM, AVERAGE)) @@ -95,6 +101,7 @@ public class StateBufferLabelerController extends StandaloneEventProcessingDecla String stateProperty = extractor.mappingPropertyValue(STATE_FIELD_ID); String stateFilter = extractor.singleValueParameter(STATE_FILTER_ID, String.class); String selectedOperation = extractor.selectedSingleValue(OPERATIONS_ID, String.class); + String labelName = extractor.textParameter(LABEL_NAME); List<StaticPropertyGroup> groupItems = extractor.collectionMembersAsGroup(LABEL_COLLECTION_ID); List<Integer> numberValues = groupItems @@ -124,8 +131,16 @@ public class StateBufferLabelerController extends StandaloneEventProcessingDecla .stream() .filter(Option::isSelected).findFirst().get().getName()) .collect(Collectors.toList()); - StateBufferLabelerParameters params = new StateBufferLabelerParameters(graph, sensorListValueProperty, stateProperty, stateFilter, selectedOperation, numberValues, labelStrings, comparators); + StateBufferLabelerParameters params = new StateBufferLabelerParameters(graph, sensorListValueProperty, stateProperty, stateFilter, selectedOperation, labelName, numberValues, labelStrings, comparators); return new ConfiguredEventProcessor<>(params, StateBufferLabeler::new); } + + @Override + public EventSchema resolveOutputStrategy(DataProcessorInvocation processingElement, ProcessingElementParameterExtractor parameterExtractor) throws SpRuntimeException { + + String labelName = parameterExtractor.textParameter(LABEL_NAME); + + return LabelerUtils.resolveOutputStrategy(processingElement, labelName); + } } diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabelerParameters.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabelerParameters.java index b289b4c..002836c 100644 --- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabelerParameters.java +++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/state/labeler/buffer/StateBufferLabelerParameters.java @@ -29,6 +29,7 @@ public class StateBufferLabelerParameters extends EventProcessorBindingParams { private String stateProperty; private String stateFilter; private String selectedOperation; + private String labelName; private List<Integer> numberValues; private List<String> labelStrings; private List<String> comparators; @@ -38,6 +39,7 @@ public class StateBufferLabelerParameters extends EventProcessorBindingParams { String stateProperty, String stateFilter, String selectedOperation, + String labelName, List<Integer> numberValues, List<String> labelStrings, List<String> comparators) { @@ -46,6 +48,7 @@ public class StateBufferLabelerParameters extends EventProcessorBindingParams { this.stateProperty = stateProperty; this.stateFilter = stateFilter; this.selectedOperation = selectedOperation; + this.labelName = labelName; this.numberValues = numberValues; this.labelStrings = labelStrings; this.comparators = comparators; @@ -106,4 +109,8 @@ public class StateBufferLabelerParameters extends EventProcessorBindingParams { public void setComparators(List<String> comparators) { this.comparators = comparators; } + + public String getLabelName() { + return labelName; + } } diff --git a/streampipes-processors-transformation-jvm/src/main/resources/org.apache.streampipes.processors.transformation.jvm.processor.state.labeler.buffer/strings.en b/streampipes-processors-transformation-jvm/src/main/resources/org.apache.streampipes.processors.transformation.jvm.processor.state.labeler.buffer/strings.en index 1b3d794..43e7bd7 100644 --- a/streampipes-processors-transformation-jvm/src/main/resources/org.apache.streampipes.processors.transformation.jvm.processor.state.labeler.buffer/strings.en +++ b/streampipes-processors-transformation-jvm/src/main/resources/org.apache.streampipes.processors.transformation.jvm.processor.state.labeler.buffer/strings.en @@ -27,3 +27,7 @@ comparatorId.description=Select a comparator numberValueId.title=Numerical Value numberValueId.description=Specify the numerical value for the condition + +labelName.title=Label Name +labelName.description=Specify the name of the label (how it can be accessed in following components) +
