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 4d4f896 [STREAMPIPES-250] Refactor Siddhi processors
new ae46263 Merge branch 'dev' of
github.com:apache/incubator-streampipes-extensions into dev
4d4f896 is described below
commit 4d4f89649f8fa37fba35af4147bfaaa9eaf1a711
Author: Dominik Riemer <[email protected]>
AuthorDate: Tue Oct 20 22:48:14 2020 +0200
[STREAMPIPES-250] Refactor Siddhi processors
---
.../processors/siddhi/FiltersSiddhiInit.java | 11 ++-
.../processors/siddhi/filter/NumericalFilter.java | 29 +++---
.../processors/siddhi/frequency/Frequency.java | 12 ++-
.../siddhi/frequencychange/FrequencyChange.java | 13 ++-
.../siddhi/listcollector/ListCollector.java | 102 +++++++++++++++++++++
.../processors/siddhi/listfilter/ListFilter.java | 64 +++++++++++++
.../processors/siddhi/sequence/Sequence.java | 13 ++-
.../processors/siddhi/stop/StreamStop.java | 13 +--
.../streampipes/processors/siddhi/trend/Trend.java | 64 ++++++-------
.../documentation.md | 48 ++++++++++
.../strings.en | 8 ++
.../documentation.md | 48 ++++++++++
.../strings.en | 8 ++
.../development/env | 2 +-
14 files changed, 362 insertions(+), 73 deletions(-)
diff --git
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/FiltersSiddhiInit.java
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/FiltersSiddhiInit.java
index 58a1b9b..a6423cf 100644
---
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/FiltersSiddhiInit.java
+++
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/FiltersSiddhiInit.java
@@ -28,9 +28,8 @@ import
org.apache.streampipes.messaging.jms.SpJmsProtocolFactory;
import org.apache.streampipes.messaging.kafka.SpKafkaProtocolFactory;
import org.apache.streampipes.processors.siddhi.config.FilterSiddhiConfig;
import
org.apache.streampipes.processors.siddhi.filter.NumericalFilterController;
-import org.apache.streampipes.processors.siddhi.frequency.FrequencyController;
-import
org.apache.streampipes.processors.siddhi.frequencychange.FrequencyChangeController;
-import org.apache.streampipes.processors.siddhi.stop.StreamStopController;
+import org.apache.streampipes.processors.siddhi.listcollector.ListCollector;
+import org.apache.streampipes.processors.siddhi.listfilter.ListFilter;
import org.apache.streampipes.processors.siddhi.trend.TrendController;
public class FiltersSiddhiInit extends StandaloneModelSubmitter {
@@ -39,8 +38,10 @@ public class FiltersSiddhiInit extends
StandaloneModelSubmitter {
DeclarersSingleton
.getInstance()
.add(new TrendController())
- .add(new NumericalFilterController());
- // TODO: currently not working
+ .add(new NumericalFilterController())
+ .add(new ListFilter())
+ .add(new ListCollector());
+ // TODO: currently not working
// .add(new StreamStopController())
// .add(new FrequencyController())
// .add(new FrequencyChangeController())
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 9368996..dc1f6bc 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
@@ -17,10 +17,9 @@
*/
package org.apache.streampipes.processors.siddhi.filter;
-import org.apache.streampipes.wrapper.siddhi.engine.SiddhiDebugCallback;
import org.apache.streampipes.wrapper.siddhi.engine.SiddhiEventEngine;
-
-import java.util.List;
+import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
+import
org.apache.streampipes.wrapper.siddhi.engine.callback.SiddhiDebugCallback;
public class NumericalFilter extends
SiddhiEventEngine<NumericalFilterParameters> {
@@ -33,32 +32,34 @@ public class NumericalFilter extends
SiddhiEventEngine<NumericalFilterParameters
}
@Override
- protected String fromStatement(List<String> inputStreamNames,
NumericalFilterParameters params) {
- String filterProperty = prepareName(params.getFilterProperty());
+ public String fromStatement(SiddhiProcessorParams<NumericalFilterParameters>
siddhiParams) {
+ NumericalFilterParameters filterParameters = siddhiParams.getParams();
+ String filterProperty = prepareName(filterParameters.getFilterProperty());
String filterOperator = "";
- if (params.getFilterOperator() == FilterOperator.EQ) {
+ if (filterParameters.getFilterOperator() == FilterOperator.EQ) {
filterOperator = "==";
- } else if (params.getFilterOperator() == FilterOperator.GE) {
+ } else if (filterParameters.getFilterOperator() == FilterOperator.GE) {
filterOperator = ">=";
- } else if (params.getFilterOperator() == FilterOperator.GT) {
+ } else if (filterParameters.getFilterOperator() == FilterOperator.GT) {
filterOperator = ">";
- } else if (params.getFilterOperator() == FilterOperator.LE) {
+ } else if (filterParameters.getFilterOperator() == FilterOperator.LE) {
filterOperator = "<=";
- } else if (params.getFilterOperator() == FilterOperator.LT) {
+ } else if (filterParameters.getFilterOperator() == FilterOperator.LT) {
filterOperator = "<";
- } else if (params.getFilterOperator() == FilterOperator.IE) {
+ } else if (filterParameters.getFilterOperator() == FilterOperator.IE) {
filterOperator = "!=";
}
// e.g. Filter for numberField value less than 10 and output all fields
//
// Siddhi query: from inputstreamname[numberField<10]
- return "from " + inputStreamNames.get(0) +"[" + filterProperty +
filterOperator + params.getThreshold() +"]";
+ return "from " + siddhiParams.getInputStreamNames().get(0) +"[" +
filterProperty + filterOperator + filterParameters.getThreshold() +"]";
}
@Override
- protected String selectStatement(NumericalFilterParameters params) {
- return getCustomOutputSelectStatement(params.getGraph());
+ public String
selectStatement(SiddhiProcessorParams<NumericalFilterParameters> siddhiParams) {
+ return
siddhiParams.getCustomOutputSelectStatement(siddhiParams.getParams().getGraph());
}
+
}
diff --git
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequency/Frequency.java
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequency/Frequency.java
index 1f317ab..19fd905 100644
---
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequency/Frequency.java
+++
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequency/Frequency.java
@@ -18,18 +18,20 @@
package org.apache.streampipes.processors.siddhi.frequency;
import org.apache.streampipes.wrapper.siddhi.engine.SiddhiEventEngine;
-
-import java.util.List;
+import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
public class Frequency extends SiddhiEventEngine<FrequencyParameters> {
@Override
- protected String fromStatement(List<String> inputStreamNames,
FrequencyParameters params) {
- return "from every not " + inputStreamNames.get(0) + " for " +
params.getDuration() + " sec";
+ public String fromStatement(SiddhiProcessorParams<FrequencyParameters>
siddhiParams) {
+ return "from every not "
+ + siddhiParams.getInputStreamNames().get(0)
+ + " for "
+ + siddhiParams.getParams().getDuration() + " sec";
}
@Override
- protected String selectStatement(FrequencyParameters params) {
+ public String selectStatement(SiddhiProcessorParams<FrequencyParameters>
siddhiParams) {
return "select *";
}
diff --git
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequencychange/FrequencyChange.java
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequencychange/FrequencyChange.java
index ffffc7a..e5003c4 100644
---
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequencychange/FrequencyChange.java
+++
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequencychange/FrequencyChange.java
@@ -18,18 +18,21 @@
package org.apache.streampipes.processors.siddhi.frequencychange;
import org.apache.streampipes.wrapper.siddhi.engine.SiddhiEventEngine;
-
-import java.util.List;
+import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
public class FrequencyChange extends
SiddhiEventEngine<FrequencyChangeParameters> {
@Override
- protected String fromStatement(List<String> inputStreamNames,
FrequencyChangeParameters params) {
- return "from every not " + inputStreamNames.get(0) + " for " +
params.getDuration() + " sec";
+ public String fromStatement(SiddhiProcessorParams<FrequencyChangeParameters>
siddhiParams) {
+ return "from every not "
+ + siddhiParams.getInputStreamNames().get(0)
+ + " for "
+ + siddhiParams.getParams().getDuration()
+ + " sec";
}
@Override
- protected String selectStatement(FrequencyChangeParameters params) {
+ public String
selectStatement(SiddhiProcessorParams<FrequencyChangeParameters> siddhiParams) {
return "select *";
}
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
new file mode 100644
index 0000000..f44e509
--- /dev/null
+++
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/listcollector/ListCollector.java
@@ -0,0 +1,102 @@
+/*
+ * 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.listcollector;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import
org.apache.streampipes.container.api.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.schema.*;
+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.*;
+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.engine.StreamPipesSiddhiProcessor;
+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";
+
+ @Override
+ public DataProcessorDescription declareModel() {
+ return
ProcessingElementBuilder.create("org.apache.streampipes.processors.siddhi.listcollector")
+ .withLocales(Locales.EN)
+ .category(DataProcessorType.PATTERN_DETECT)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .requiredStream(StreamRequirementsBuilder.create()
+
.requiredPropertyWithUnaryMapping(EpRequirements.anyProperty(), Labels.withId
+ (LIST_KEY), PropertyScope.MEASUREMENT_PROPERTY)
+ .build())
+ .requiredIntegerParameter(Labels.withId(TIME_WINDOW_SIZE))
+ .outputStrategy(OutputStrategies.customTransformation())
+ .build();
+
+ }
+
+ @Override
+ public String fromStatement(SiddhiProcessorParams<ProcessorParams>
siddhiParams) {
+ return "from " +siddhiParams.getInputStreamNames().get(0) +
"#window.lengthBatch(10)";
+ }
+
+ @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";
+
+ }
+
+ @Override
+ public EventSchema resolveOutputStrategy(DataProcessorInvocation
processingElement, ProcessingElementParameterExtractor extractor) throws
SpRuntimeException {
+ String propertySelector = extractor.mappingPropertyValue(LIST_KEY);
+ EventSchema inputSchema =
processingElement.getInputStreams().get(0).getEventSchema();
+
+ List<EventProperty> selectedProperty = extractor
+
.getInputStreamEventPropertySubset(Collections.singletonList(propertySelector));
+
+ if (selectedProperty.size() > 0) {
+ EventPropertyPrimitive prop = (EventPropertyPrimitive)
selectedProperty.get(0);
+ String newPropertyName = prop.getRuntimeName() + "_list";
+ String newDatatype = prop.getRuntimeType();
+ String newDomainProperty = prop.getDomainProperties().get(0).toString();
+
+ EventPropertyList ep = EpProperties.listEp(Labels.from("list", "", ""),
+ newPropertyName,
+ Datatypes.fromDatatypeString(newDatatype),
+ newDomainProperty);
+
+ inputSchema.addEventProperty(ep);
+ }
+
+ return inputSchema;
+ }
+}
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
new file mode 100644
index 0000000..54d52bc
--- /dev/null
+++
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/listfilter/ListFilter.java
@@ -0,0 +1,64 @@
+/*
+ * 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.listfilter;
+
+import org.apache.streampipes.model.DataProcessorType;
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+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.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.standalone.ProcessorParams;
+
+public class ListFilter extends StreamPipesSiddhiProcessor {
+
+ private static final String LIST_KEY = "list-key";
+ private static final String REQUIRED_VALUE_KEY = "required-value";
+
+ @Override
+ public DataProcessorDescription declareModel() {
+ return
ProcessingElementBuilder.create("org.apache.streampipes.processors.siddhi.listfilter")
+ .withLocales(Locales.EN)
+ .category(DataProcessorType.PATTERN_DETECT)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .requiredStream(StreamRequirementsBuilder.create()
+
.requiredPropertyWithUnaryMapping(EpRequirements.listRequirement(),
Labels.withId
+ (LIST_KEY), PropertyScope.MEASUREMENT_PROPERTY)
+ .build())
+ .requiredTextParameter(Labels.withId(REQUIRED_VALUE_KEY))
+ .outputStrategy(OutputStrategies.keep())
+ .build();
+ }
+
+
+ @Override
+ public String fromStatement(SiddhiProcessorParams<ProcessorParams>
siddhiParams) {
+ return null;
+ }
+
+ @Override
+ public String selectStatement(SiddhiProcessorParams<ProcessorParams>
siddhiParams) {
+ return null;
+ }
+}
diff --git
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/sequence/Sequence.java
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/sequence/Sequence.java
index 3aa1324..088fff3 100644
---
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/sequence/Sequence.java
+++
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/sequence/Sequence.java
@@ -18,23 +18,26 @@
package org.apache.streampipes.processors.siddhi.sequence;
import org.apache.streampipes.wrapper.siddhi.engine.SiddhiEventEngine;
-
-import java.util.List;
+import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
public class Sequence extends SiddhiEventEngine<SequenceParameters> {
@Override
- protected String fromStatement(List<String> inputStreamNames,
SequenceParameters params) {
+ public String fromStatement(SiddhiProcessorParams<SequenceParameters>
siddhiParams) {
// from every (e1=MaterialSupplyStream) -> e2=MaterialConsumptionStream
within 10 min
// return "from every(e1=" + inputStreamNames.get(0) + ") -> not e2=" +
inputStreamNames.get(0) + " for " + params.getDuration() + " sec";
// return "define stream Test(timestamp LONG,message STRING);\n" +
- return "from every not " + inputStreamNames.get(0) + " for " +
params.getDuration() + " sec";
+ return "from every not "
+ + siddhiParams.getInputStreamNames().get(0)
+ + " for "
+ + siddhiParams.getParams().getDuration()
+ + " sec";
}
@Override
- protected String selectStatement(SequenceParameters params) {
+ public String selectStatement(SiddhiProcessorParams<SequenceParameters>
siddhiParams) {
// return getCustomOutputSelectStatement(params.getGraph());
return "select *";
// return "select currentTimeMillis() as s0timestamp, 'Customer has not
arrived' as message";
diff --git
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/stop/StreamStop.java
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/stop/StreamStop.java
index 26a380d..030dfd2 100644
---
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/stop/StreamStop.java
+++
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/stop/StreamStop.java
@@ -18,9 +18,7 @@
package org.apache.streampipes.processors.siddhi.stop;
import org.apache.streampipes.wrapper.siddhi.engine.SiddhiEventEngine;
-
-import java.util.Arrays;
-import java.util.List;
+import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
public class StreamStop extends SiddhiEventEngine<StreamStopParameters> {
@@ -29,13 +27,16 @@ public class StreamStop extends
SiddhiEventEngine<StreamStopParameters> {
}
@Override
- protected String fromStatement(List<String> inputStreamNames,
StreamStopParameters params) {
+ public String fromStatement(SiddhiProcessorParams<StreamStopParameters>
siddhiParams) {
return "define stream Test(timestamp LONG,message STRING);\n" +
- "from every not " + inputStreamNames.get(0) + " for " +
params.getDuration() + " sec";
+ "from every not "
+ + siddhiParams.getInputStreamNames().get(0)
+ + " for " + siddhiParams.getParams().getDuration()
+ + " sec";
}
@Override
- protected String selectStatement(StreamStopParameters params) {
+ public String selectStatement(SiddhiProcessorParams<StreamStopParameters>
siddhiParams) {
//setSortedEventKeys(Arrays.asList("timestamp", "message"));
return "select currentTimeMillis() as timestamp, 'Event stream has
stopped' as message";
}
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 84bd24d..444c1c7 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
@@ -17,10 +17,9 @@
*/
package org.apache.streampipes.processors.siddhi.trend;
-import org.apache.streampipes.wrapper.siddhi.engine.SiddhiDebugCallback;
import org.apache.streampipes.wrapper.siddhi.engine.SiddhiEventEngine;
-
-import java.util.List;
+import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
+import
org.apache.streampipes.wrapper.siddhi.engine.callback.SiddhiDebugCallback;
public class Trend extends SiddhiEventEngine<TrendParameters> {
@@ -33,37 +32,38 @@ public class Trend extends
SiddhiEventEngine<TrendParameters> {
}
@Override
- protected String fromStatement(List<String> inputStreamNames,
TrendParameters params) {
- String mappingProperty = prepareName(params.getMapping());
- int duration = params.getDuration();
- String inequaloperator;
- String operator;
+ 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 = Double.valueOf(params.getIncrease());
- increase = (increase / 100) + 1;
+ double increase = trendParameters.getIncrease();
+ increase = (increase / 100) + 1;
- if (params.getOperation() == TrendOperator.INCREASE) {
- inequaloperator = "<=";
- operator = "/";
+ if (trendParameters.getOperation() == TrendOperator.INCREASE) {
+ inequaloperator = "<=";
+ operator = "/";
- } else {
- inequaloperator = ">=";
- operator = "*";
- }
+ } else {
+ inequaloperator = ">=";
+ operator = "*";
+ }
- String s = "from every(e1="
- + inputStreamNames.get(0)
- +") -> e2="
- +inputStreamNames.get(0)
- + "[e1." + mappingProperty
- + inequaloperator
- + "("
- + mappingProperty
- + operator
- + increase
- +")"
- + "]<1>"
- + " within " + duration + " sec";
+ 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]";
@@ -71,8 +71,8 @@ public class Trend extends SiddhiEventEngine<TrendParameters>
{
}
@Override
- protected String selectStatement(TrendParameters params) {
- return getCustomOutputSelectStatement(params.getGraph(), "e2");
+ public String selectStatement(SiddhiProcessorParams<TrendParameters>
siddhiParams) {
+ return
siddhiParams.getCustomOutputSelectStatement(siddhiParams.getParams().getGraph(),
"e2");
}
}
diff --git
a/streampipes-processors-filters-siddhi/src/main/resources/org.apache.streampipes.processors.siddhi.listcollector/documentation.md
b/streampipes-processors-filters-siddhi/src/main/resources/org.apache.streampipes.processors.siddhi.listcollector/documentation.md
new file mode 100644
index 0000000..c4d6b3e
--- /dev/null
+++
b/streampipes-processors-filters-siddhi/src/main/resources/org.apache.streampipes.processors.siddhi.listcollector/documentation.md
@@ -0,0 +1,48 @@
+<!--
+ ~ 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.
+ ~
+ -->
+
+## List Collector
+
+<p align="center">
+ <img src="icon.png" width="150px;" class="pe-image-documentation"/>
+</p>
+
+***
+
+## Description
+
+Detects the increase of a numerical field over a customizable time window.
Example: A temperature value increases by 10 percent within 5 minutes.
+Add a detailed description here
+
+***
+
+## Required input
+
+
+***
+
+## Configuration
+
+Describe the configuration parameters here
+
+### 1st parameter
+
+
+### 2nd parameter
+
+## Output
\ No newline at end of file
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
new file mode 100644
index 0000000..c017263
--- /dev/null
+++
b/streampipes-processors-filters-siddhi/src/main/resources/org.apache.streampipes.processors.siddhi.listcollector/strings.en
@@ -0,0 +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.
+
+list-field.title=Value to Observe
+list-field.description=Specifies the value that should be monitored.
+
+time-window-size.title=Time Window Size
+time-window-size.description=The time window size in seconds
diff --git
a/streampipes-processors-filters-siddhi/src/main/resources/org.apache.streampipes.processors.siddhi.listfilter/documentation.md
b/streampipes-processors-filters-siddhi/src/main/resources/org.apache.streampipes.processors.siddhi.listfilter/documentation.md
new file mode 100644
index 0000000..7ebdc2d
--- /dev/null
+++
b/streampipes-processors-filters-siddhi/src/main/resources/org.apache.streampipes.processors.siddhi.listfilter/documentation.md
@@ -0,0 +1,48 @@
+<!--
+ ~ 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.
+ ~
+ -->
+
+## Trend
+
+<p align="center">
+ <img src="icon.png" width="150px;" class="pe-image-documentation"/>
+</p>
+
+***
+
+## Description
+
+Detects the increase of a numerical field over a customizable time window.
Example: A temperature value increases by 10 percent within 5 minutes.
+Add a detailed description here
+
+***
+
+## Required input
+
+
+***
+
+## Configuration
+
+Describe the configuration parameters here
+
+### 1st parameter
+
+
+### 2nd parameter
+
+## Output
\ No newline at end of file
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
new file mode 100644
index 0000000..6e640c5
--- /dev/null
+++
b/streampipes-processors-filters-siddhi/src/main/resources/org.apache.streampipes.processors.siddhi.listfilter/strings.en
@@ -0,0 +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.
+
+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-transformation-jvm/development/env
b/streampipes-processors-transformation-jvm/development/env
index 6a20bde..10821f4 100644
--- a/streampipes-processors-transformation-jvm/development/env
+++ b/streampipes-processors-transformation-jvm/development/env
@@ -14,5 +14,5 @@
# limitations under the License.
SP_PORT=6060
-SP_HOST=host.docker.internal
+SP_HOST=localhost
SP_DEBUG=true
