This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch dev
in repository
https://gitbox.apache.org/repos/asf/incubator-streampipes-extensions.git
The following commit(s) were added to refs/heads/dev by this push:
new 45eeb0f [STREAMPIPES-250] Refactor Siddhi processors, add list filter
element
45eeb0f is described below
commit 45eeb0f051653dd099048a26374a1c52c0e6ed71
Author: Dominik Riemer <[email protected]>
AuthorDate: Mon Oct 26 21:58:00 2020 +0100
[STREAMPIPES-250] Refactor Siddhi processors, add list filter element
---
.../processors/siddhi/filter/FilterOperator.java | 23 ---------
.../processors/siddhi/filter/NumericalFilter.java | 29 +++++------
.../siddhi/filter/NumericalFilterController.java | 20 ++++----
.../siddhi/filter/NumericalFilterParameters.java | 7 +--
.../siddhi/listcollector/ListCollector.java | 34 ++++++++----
.../processors/siddhi/listfilter/ListFilter.java | 26 ++++++++--
.../streampipes/processors/siddhi/trend/Trend.java | 60 ++++++++++------------
.../strings.en | 10 ++--
.../strings.en | 4 +-
.../siddhi/trend/TestTrendProcessor.java | 2 +-
10 files changed, 110 insertions(+), 105 deletions(-)
diff --git
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/FilterOperator.java
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/FilterOperator.java
deleted file mode 100644
index c3a59e1..0000000
---
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/FilterOperator.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * 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.siddhi.filter;
-
-public enum FilterOperator {
-GE, GT, LE, LT, EQ, IE
-}
diff --git
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilter.java
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilter.java
index 9b206c1..4145b56 100644
---
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilter.java
+++
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilter.java
@@ -21,9 +21,12 @@ import
org.apache.streampipes.wrapper.siddhi.constants.SiddhiStreamSelector;
import org.apache.streampipes.wrapper.siddhi.engine.SiddhiEventEngine;
import
org.apache.streampipes.wrapper.siddhi.engine.callback.SiddhiDebugCallback;
import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
+import org.apache.streampipes.wrapper.siddhi.query.FromClause;
import org.apache.streampipes.wrapper.siddhi.query.SelectClause;
import org.apache.streampipes.wrapper.siddhi.query.expression.Expression;
import org.apache.streampipes.wrapper.siddhi.query.expression.Expressions;
+import
org.apache.streampipes.wrapper.siddhi.query.expression.RelationalOperator;
+import
org.apache.streampipes.wrapper.siddhi.query.expression.RelationalOperatorExpression;
public class NumericalFilter extends
SiddhiEventEngine<NumericalFilterParameters> {
@@ -38,27 +41,21 @@ public class NumericalFilter extends
SiddhiEventEngine<NumericalFilterParameters
@Override
public String fromStatement(SiddhiProcessorParams<NumericalFilterParameters>
siddhiParams) {
NumericalFilterParameters filterParameters = siddhiParams.getParams();
- String filterProperty = prepareName(filterParameters.getFilterProperty());
- String filterOperator = "";
+ String filterProperty = filterParameters.getFilterProperty();
+ RelationalOperator operator = filterParameters.getFilterOperator();
+ Double threshold = filterParameters.getThreshold();
- if (filterParameters.getFilterOperator() == FilterOperator.EQ) {
- filterOperator = "==";
- } else if (filterParameters.getFilterOperator() == FilterOperator.GE) {
- filterOperator = ">=";
- } else if (filterParameters.getFilterOperator() == FilterOperator.GT) {
- filterOperator = ">";
- } else if (filterParameters.getFilterOperator() == FilterOperator.LE) {
- filterOperator = "<=";
- } else if (filterParameters.getFilterOperator() == FilterOperator.LT) {
- filterOperator = "<";
- } else if (filterParameters.getFilterOperator() == FilterOperator.IE) {
- filterOperator = "!=";
- }
+ FromClause fromClause = FromClause.create();
+ Expression filter = new RelationalOperatorExpression(operator,
Expressions.property(filterProperty), Expressions.staticValue(threshold));
+ Expression stream =
Expressions.filter(Expressions.stream(siddhiParams.getInputStreamNames().get(0)),
filter);
+ fromClause.add(stream);
+
+ return fromClause.toSiddhiEpl();
// e.g. Filter for numberField value less than 10 and output all fields
//
// Siddhi query: from inputstreamname[numberField<10]
- return "from " + siddhiParams.getInputStreamNames().get(0) +"[" +
filterProperty + filterOperator + filterParameters.getThreshold() +"]";
+ //return "from " + siddhiParams.getInputStreamNames().get(0) +"[" +
filterProperty + filterOperator + filterParameters.getThreshold() +"]";
}
@Override
diff --git
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilterController.java
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilterController.java
index 549c7ae..9f13160 100644
---
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilterController.java
+++
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilterController.java
@@ -26,6 +26,7 @@ import
org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
import
org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
import org.apache.streampipes.sdk.helpers.*;
import org.apache.streampipes.sdk.utils.Assets;
+import
org.apache.streampipes.wrapper.siddhi.query.expression.RelationalOperator;
import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
@@ -59,25 +60,26 @@ public class NumericalFilterController extends
StandaloneEventProcessingDeclarer
Double threshold = extractor.singleValueParameter(VALUE, Double.class);
String stringOperation = extractor.selectedSingleValue(OPERATION,
String.class);
- String operation = "GT";
+ RelationalOperator operator = RelationalOperator.GREATER_THAN;
if (stringOperation.equals("<=")) {
- operation = "LT";
+ operator = RelationalOperator.LESSER_EQUALS;
} else if (stringOperation.equals("<")) {
- operation = "LE";
+ operator = RelationalOperator.LESSER_THAN;
} else if (stringOperation.equals(">=")) {
- operation = "GE";
+ operator = RelationalOperator.GREATER_EQUALS;
} else if (stringOperation.equals("==")) {
- operation = "EQ";
+ operator = RelationalOperator.EQUALS;
} else if (stringOperation.equals("!=")) {
- operation = "IE";
+ operator = RelationalOperator.NOT_EQUALS;
}
String filterProperty = extractor.mappingPropertyValue(NUMBER_MAPPING);
- NumericalFilterParameters staticParam = new
NumericalFilterParameters(graph, threshold, FilterOperator.valueOf
- (operation)
- , filterProperty);
+ NumericalFilterParameters staticParam = new
NumericalFilterParameters(graph,
+ threshold,
+ operator,
+ filterProperty);
return new ConfiguredEventProcessor<>(staticParam, NumericalFilter::new);
}
diff --git
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilterParameters.java
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilterParameters.java
index b918928..4702977 100644
---
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilterParameters.java
+++
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilterParameters.java
@@ -20,14 +20,15 @@ package org.apache.streampipes.processors.siddhi.filter;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
import
org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams;
+import
org.apache.streampipes.wrapper.siddhi.query.expression.RelationalOperator;
public class NumericalFilterParameters extends EventProcessorBindingParams {
private double threshold;
- private FilterOperator filterOperator;
+ private RelationalOperator filterOperator;
private String filterProperty;
- public NumericalFilterParameters(DataProcessorInvocation graph, Double
threshold, FilterOperator
+ public NumericalFilterParameters(DataProcessorInvocation graph, Double
threshold, RelationalOperator
filterOperator, String filterProperty) {
super(graph);
this.threshold = threshold;
@@ -39,7 +40,7 @@ public class NumericalFilterParameters extends
EventProcessorBindingParams {
return threshold;
}
- public FilterOperator getFilterOperator() {
+ public RelationalOperator getFilterOperator() {
return filterOperator;
}
diff --git
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/listcollector/ListCollector.java
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/listcollector/ListCollector.java
index f44e509..036f8d2 100644
---
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/listcollector/ListCollector.java
+++
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/listcollector/ListCollector.java
@@ -29,21 +29,22 @@ import
org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
import org.apache.streampipes.sdk.helpers.*;
import org.apache.streampipes.sdk.utils.Assets;
import org.apache.streampipes.sdk.utils.Datatypes;
-import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
+import org.apache.streampipes.wrapper.siddhi.constants.SiddhiStreamSelector;
import org.apache.streampipes.wrapper.siddhi.engine.StreamPipesSiddhiProcessor;
+import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
+import org.apache.streampipes.wrapper.siddhi.query.FromClause;
+import org.apache.streampipes.wrapper.siddhi.query.SelectClause;
+import org.apache.streampipes.wrapper.siddhi.query.expression.Expressions;
import org.apache.streampipes.wrapper.standalone.ProcessorParams;
import java.util.Collections;
import java.util.List;
-import java.util.StringJoiner;
-
-import static
org.apache.streampipes.wrapper.siddhi.utils.SiddhiUtils.prepareName;
public class ListCollector extends StreamPipesSiddhiProcessor
implements
ResolvesContainerProvidedOutputStrategy<DataProcessorInvocation,
ProcessingElementParameterExtractor> {
private static final String LIST_KEY = "list-key";
- private static final String TIME_WINDOW_SIZE = "time-window-size";
+ private static final String WINDOW_SIZE = "window-size";
@Override
public DataProcessorDescription declareModel() {
@@ -55,7 +56,7 @@ public class ListCollector extends StreamPipesSiddhiProcessor
.requiredPropertyWithUnaryMapping(EpRequirements.anyProperty(), Labels.withId
(LIST_KEY), PropertyScope.MEASUREMENT_PROPERTY)
.build())
- .requiredIntegerParameter(Labels.withId(TIME_WINDOW_SIZE))
+ .requiredIntegerParameter(Labels.withId(WINDOW_SIZE))
.outputStrategy(OutputStrategies.customTransformation())
.build();
@@ -63,16 +64,29 @@ public class ListCollector extends
StreamPipesSiddhiProcessor
@Override
public String fromStatement(SiddhiProcessorParams<ProcessorParams>
siddhiParams) {
- return "from " +siddhiParams.getInputStreamNames().get(0) +
"#window.lengthBatch(10)";
+
+ Integer batchWindowSize =
siddhiParams.getParams().extractor().singleValueParameter(WINDOW_SIZE,
Integer.class);
+
+ FromClause fromClause = FromClause.create();
+
fromClause.add(Expressions.stream(siddhiParams.getInputStreamNames().get(0),
Expressions.batchWindow(batchWindowSize)));
+
+ return fromClause.toSiddhiEpl();
}
@Override
public String selectStatement(SiddhiProcessorParams<ProcessorParams>
siddhiParams) {
+
String propertySelector =
siddhiParams.getParams().extractor().mappingPropertyValue(LIST_KEY);
- StringJoiner joiner = new StringJoiner(",");
- siddhiParams.getEventTypeInfo().forEach((key, value) ->
value.forEach(field -> joiner.add(field.getSelectorPrefix() +
field.getFieldName())));
- return "select " +joiner.toString() +", list:collect(" +
prepareName(propertySelector) +") as " + prepareName(propertySelector) +
"_list";
+ SelectClause selectClause = SelectClause.create();
+ siddhiParams
+ .getEventTypeInfo()
+ .forEach((key, value) -> value
+ .forEach(field ->
selectClause.addProperty(Expressions.property(SiddhiStreamSelector.FIRST_INPUT_STREAM,
field.getFieldName()))));
+
+
selectClause.addProperty(Expressions.as(Expressions.collectList(Expressions.property(propertySelector)),
propertySelector + "_list"));
+
+ return selectClause.toSiddhiEpl();
}
@Override
diff --git
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/listfilter/ListFilter.java
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/listfilter/ListFilter.java
index 54d52bc..59b7ff7 100644
---
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/listfilter/ListFilter.java
+++
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/listfilter/ListFilter.java
@@ -19,16 +19,23 @@ package org.apache.streampipes.processors.siddhi.listfilter;
import org.apache.streampipes.model.DataProcessorType;
import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.schema.EventPropertyList;
+import org.apache.streampipes.model.schema.EventPropertyPrimitive;
import org.apache.streampipes.model.schema.PropertyScope;
import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+import
org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor;
import org.apache.streampipes.sdk.helpers.EpRequirements;
import org.apache.streampipes.sdk.helpers.Labels;
import org.apache.streampipes.sdk.helpers.Locales;
import org.apache.streampipes.sdk.helpers.OutputStrategies;
import org.apache.streampipes.sdk.utils.Assets;
-import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
import org.apache.streampipes.wrapper.siddhi.engine.StreamPipesSiddhiProcessor;
+import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
+import org.apache.streampipes.wrapper.siddhi.query.FromClause;
+import org.apache.streampipes.wrapper.siddhi.query.SelectClause;
+import org.apache.streampipes.wrapper.siddhi.query.expression.Expression;
+import org.apache.streampipes.wrapper.siddhi.query.expression.Expressions;
import org.apache.streampipes.wrapper.standalone.ProcessorParams;
public class ListFilter extends StreamPipesSiddhiProcessor {
@@ -51,14 +58,25 @@ public class ListFilter extends StreamPipesSiddhiProcessor {
.build();
}
-
@Override
public String fromStatement(SiddhiProcessorParams<ProcessorParams>
siddhiParams) {
- return null;
+ String filteredFieldSelector =
siddhiParams.getParams().extractor().mappingPropertyValue(LIST_KEY);
+ Object filterValue = extractFilterValue(filteredFieldSelector,
siddhiParams.getParams().extractor());
+ FromClause fromClause = FromClause.create();
+
+ Expression containsExp =
Expressions.containsListItem(Expressions.property(filteredFieldSelector),
filterValue);
+
fromClause.add(Expressions.filter(Expressions.stream(siddhiParams.getInputStreamNames().get(0)),
containsExp));
+
+ return fromClause.toSiddhiEpl();
}
@Override
public String selectStatement(SiddhiProcessorParams<ProcessorParams>
siddhiParams) {
- return null;
+ return SelectClause.createWildcard().toSiddhiEpl();
+ }
+
+ private Object extractFilterValue(String selector,
ProcessingElementParameterExtractor extractor) {
+ EventPropertyList prop = (EventPropertyList)
extractor.getEventPropertyBySelector(selector);
+ return extractor.singleValueParameter((EventPropertyPrimitive)
prop.getEventProperty(), REQUIRED_VALUE_KEY);
}
}
diff --git
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/Trend.java
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/Trend.java
index fb4651a..e5d533f 100644
---
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/Trend.java
+++
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/Trend.java
@@ -20,8 +20,10 @@ package org.apache.streampipes.processors.siddhi.trend;
import org.apache.streampipes.wrapper.siddhi.engine.SiddhiEventEngine;
import
org.apache.streampipes.wrapper.siddhi.engine.callback.SiddhiDebugCallback;
import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
+import org.apache.streampipes.wrapper.siddhi.query.FromClause;
import org.apache.streampipes.wrapper.siddhi.query.SelectClause;
-import org.apache.streampipes.wrapper.siddhi.query.expression.Expressions;
+import org.apache.streampipes.wrapper.siddhi.query.expression.*;
+import
org.apache.streampipes.wrapper.siddhi.query.expression.pattern.PatternCountOperator;
import java.util.List;
@@ -38,40 +40,35 @@ public class Trend extends
SiddhiEventEngine<TrendParameters> {
@Override
public String fromStatement(SiddhiProcessorParams<TrendParameters>
siddhiParams) {
TrendParameters trendParameters = siddhiParams.getParams();
+
String mappingProperty = prepareName(trendParameters.getMapping());
int duration = trendParameters.getDuration();
- String inequaloperator;
- String operator;
-
double increase = trendParameters.getIncrease();
increase = (increase / 100) + 1;
- if (trendParameters.getOperation() == TrendOperator.INCREASE) {
- inequaloperator = "<=";
- operator = "/";
-
- } else {
- inequaloperator = ">=";
- operator = "*";
- }
-
- String s = "from every(e1="
- + siddhiParams.getInputStreamNames().get(0)
- + ") -> e2="
- + siddhiParams.getInputStreamNames().get(0)
- + "[e1." + mappingProperty
- + inequaloperator
- + "("
- + mappingProperty
- + operator
- + increase
- + ")"
- + "]<1>"
- + " within " + duration + " sec";
-
- //String s = "from e1="+inputStreamNames.get(0) + "[e1.s0randomValue > 5]";
-
- return s;
+ FromClause fromClause = FromClause.create();
+ StreamExpression exp1 = Expressions.every(
+ Expressions.stream("e1",
siddhiParams.getInputStreamNames().get(0)));
+ StreamExpression exp2 = Expressions.stream("e2",
siddhiParams.getInputStreamNames().get(0));
+
+ PropertyExpressionBase mathExp = trendParameters.getOperation() ==
TrendOperator.INCREASE ?
+ Expressions.divide(Expressions.property(mappingProperty),
Expressions.staticValue(increase)) :
+ Expressions.multiply(Expressions.property(mappingProperty),
Expressions.staticValue(increase));
+
+ RelationalOperatorExpression opExp = trendParameters.getOperation() ==
TrendOperator.INCREASE ?
+ Expressions.le(Expressions.property("e1", mappingProperty),
mathExp) :
+ Expressions.ge(Expressions.property("e1", mappingProperty),
mathExp);
+
+ StreamFilterExpression filterExp = Expressions
+ .filter(exp2, Expressions.patternCount(1,
PatternCountOperator.EXACTLY_N), opExp);
+
+ Expression sequence = (Expressions.sequence(exp1,
+ filterExp,
+ Expressions.within(duration, SiddhiTimeUnit.SECONDS)));
+
+ fromClause.add(sequence);
+
+ return fromClause.toSiddhiEpl();
}
@Override
@@ -80,10 +77,9 @@ public class Trend extends
SiddhiEventEngine<TrendParameters> {
List<String> outputFieldSelectors =
siddhiParams.getParams().getOutputFieldSelectors();
outputFieldSelectors
.forEach(outputFieldSelector -> selectClause
- .addProperty(Expressions.property("e2",
outputFieldSelector)));
+ .addProperty(Expressions.property("e2",
outputFieldSelector)));
return selectClause.toSiddhiEpl();
- //return
siddhiParams.getCustomOutputSelectStatement(siddhiParams.getParams().getGraph(),
"e2");
}
}
diff --git
a/streampipes-processors-filters-siddhi/src/main/resources/org.apache.streampipes.processors.siddhi.listcollector/strings.en
b/streampipes-processors-filters-siddhi/src/main/resources/org.apache.streampipes.processors.siddhi.listcollector/strings.en
index c017263..47aa939 100644
---
a/streampipes-processors-filters-siddhi/src/main/resources/org.apache.streampipes.processors.siddhi.listcollector/strings.en
+++
b/streampipes-processors-filters-siddhi/src/main/resources/org.apache.streampipes.processors.siddhi.listcollector/strings.en
@@ -1,8 +1,8 @@
org.apache.streampipes.processors.siddhi.listcollector.title=List Collector
-org.apache.streampipes.processors.siddhi.listcollector.description=Filters a
field from an input event that represents a list for the presence of a given
value.
+org.apache.streampipes.processors.siddhi.listcollector.description=Collects
all values from a field within a specified batch window into a list.
-list-field.title=Value to Observe
-list-field.description=Specifies the value that should be monitored.
+list-key.title=Field
+list-key.description=The field where values should be collected into a list.
-time-window-size.title=Time Window Size
-time-window-size.description=The time window size in seconds
+window-size.title=Batch Window Size
+window-size.description=The batch window size.
diff --git
a/streampipes-processors-filters-siddhi/src/main/resources/org.apache.streampipes.processors.siddhi.listfilter/strings.en
b/streampipes-processors-filters-siddhi/src/main/resources/org.apache.streampipes.processors.siddhi.listfilter/strings.en
index 6e640c5..1956314 100644
---
a/streampipes-processors-filters-siddhi/src/main/resources/org.apache.streampipes.processors.siddhi.listfilter/strings.en
+++
b/streampipes-processors-filters-siddhi/src/main/resources/org.apache.streampipes.processors.siddhi.listfilter/strings.en
@@ -1,8 +1,8 @@
org.apache.streampipes.processors.siddhi.listfilter.title=List Filter
org.apache.streampipes.processors.siddhi.listfilter.description=Filters a
field from an input event that represents a list for the presence of a given
value.
-list-field.title=Value to Observe
-list-field.description=Specifies the value that should be monitored.
+list-key.title=Value to Observe
+list-key.description=Specifies the value that should be monitored.
required-value.title=Required value
required-value.description=The value (a primitive data type such as a number
or text) that should be part of the list.
diff --git
a/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java
b/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java
index 9a5f033..0214494 100644
---
a/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java
+++
b/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java
@@ -107,7 +107,7 @@ public class TestTrendProcessor {
graph.getOutputStream().getEventGrounding().getTransportProtocol().getTopicDefinition().setActualTopicName("output-topic");
TrendParameters params = new TrendParameters(graph, trendOperator,
increase, timeWindow, "s0" +
- "::randomValue", Collections.emptyList());
+ "::randomValue", Arrays.asList("s0::randomValue"));
Trend trend = new Trend(event -> actualMatchCount[0]++);
trend.onInvocation(params, null, null);
