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 eca1daf [STREAMPIPES-250] Refactor select statements
new b12c1ad Merge branch 'dev' of
github.com:apache/incubator-streampipes-extensions into dev
eca1daf is described below
commit eca1daf7536494232bc899b1fbe86b20f9678cf6
Author: Dominik Riemer <[email protected]>
AuthorDate: Thu Oct 22 20:46:51 2020 +0200
[STREAMPIPES-250] Refactor select statements
---
.../processors/siddhi/filter/NumericalFilter.java | 11 +++++++----
.../processors/siddhi/frequency/Frequency.java | 3 ++-
.../siddhi/frequencychange/FrequencyChange.java | 13 ++++++------
.../streampipes/processors/siddhi/trend/Trend.java | 15 ++++++++++++--
.../processors/siddhi/trend/TrendController.java | 16 +++++++++------
.../processors/siddhi/trend/TrendParameters.java | 13 +++++++-----
.../siddhi/trend/TestTrendProcessor.java | 23 +++++++++-------------
7 files changed, 56 insertions(+), 38 deletions(-)
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 42152ee..9b206c1 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
@@ -19,9 +19,10 @@ package org.apache.streampipes.processors.siddhi.filter;
import org.apache.streampipes.wrapper.siddhi.constants.SiddhiStreamSelector;
import org.apache.streampipes.wrapper.siddhi.engine.SiddhiEventEngine;
-import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
import
org.apache.streampipes.wrapper.siddhi.engine.callback.SiddhiDebugCallback;
+import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
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;
public class NumericalFilter extends
SiddhiEventEngine<NumericalFilterParameters> {
@@ -65,11 +66,13 @@ public class NumericalFilter extends
SiddhiEventEngine<NumericalFilterParameters
SelectClause selectClause = SelectClause.create();
siddhiParams
.getOutputEventKeys()
- .forEach(fieldName -> selectClause.addProperty(Expressions
- .property(SiddhiStreamSelector.FIRST_INPUT_STREAM,
fieldName)));
+ .forEach(fieldName ->
selectClause.addProperty(makeProperty(fieldName)));
return selectClause.toSiddhiEpl();
- //return
siddhiParams.getCustomOutputSelectStatement(siddhiParams.getParams().getGraph());
+ }
+
+ private Expression makeProperty(String fieldName) {
+ return Expressions.property(SiddhiStreamSelector.FIRST_INPUT_STREAM,
fieldName);
}
}
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 19fd905..2a72d26 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
@@ -19,6 +19,7 @@ package org.apache.streampipes.processors.siddhi.frequency;
import org.apache.streampipes.wrapper.siddhi.engine.SiddhiEventEngine;
import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
+import org.apache.streampipes.wrapper.siddhi.query.SelectClause;
public class Frequency extends SiddhiEventEngine<FrequencyParameters> {
@@ -32,7 +33,7 @@ public class Frequency extends
SiddhiEventEngine<FrequencyParameters> {
@Override
public String selectStatement(SiddhiProcessorParams<FrequencyParameters>
siddhiParams) {
- return "select *";
+ return SelectClause.createWildcard().toSiddhiEpl();
}
}
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 e5003c4..388ee4f 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
@@ -19,21 +19,22 @@ package
org.apache.streampipes.processors.siddhi.frequencychange;
import org.apache.streampipes.wrapper.siddhi.engine.SiddhiEventEngine;
import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
+import org.apache.streampipes.wrapper.siddhi.query.SelectClause;
public class FrequencyChange extends
SiddhiEventEngine<FrequencyChangeParameters> {
@Override
public String fromStatement(SiddhiProcessorParams<FrequencyChangeParameters>
siddhiParams) {
- return "from every not "
- + siddhiParams.getInputStreamNames().get(0)
- + " for "
- + siddhiParams.getParams().getDuration()
- + " sec";
+ return "from every not "
+ + siddhiParams.getInputStreamNames().get(0)
+ + " for "
+ + siddhiParams.getParams().getDuration()
+ + " sec";
}
@Override
public String
selectStatement(SiddhiProcessorParams<FrequencyChangeParameters> siddhiParams) {
- return "select *";
+ return SelectClause.createWildcard().toSiddhiEpl();
}
}
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 444c1c7..fb4651a 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
@@ -18,8 +18,12 @@
package org.apache.streampipes.processors.siddhi.trend;
import org.apache.streampipes.wrapper.siddhi.engine.SiddhiEventEngine;
-import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
import
org.apache.streampipes.wrapper.siddhi.engine.callback.SiddhiDebugCallback;
+import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
+import org.apache.streampipes.wrapper.siddhi.query.SelectClause;
+import org.apache.streampipes.wrapper.siddhi.query.expression.Expressions;
+
+import java.util.List;
public class Trend extends SiddhiEventEngine<TrendParameters> {
@@ -72,7 +76,14 @@ public class Trend extends
SiddhiEventEngine<TrendParameters> {
@Override
public String selectStatement(SiddhiProcessorParams<TrendParameters>
siddhiParams) {
- return
siddhiParams.getCustomOutputSelectStatement(siddhiParams.getParams().getGraph(),
"e2");
+ SelectClause selectClause = SelectClause.create();
+ List<String> outputFieldSelectors =
siddhiParams.getParams().getOutputFieldSelectors();
+ outputFieldSelectors
+ .forEach(outputFieldSelector -> selectClause
+ .addProperty(Expressions.property("e2",
outputFieldSelector)));
+
+ return selectClause.toSiddhiEpl();
+ //return
siddhiParams.getCustomOutputSelectStatement(siddhiParams.getParams().getGraph(),
"e2");
}
}
diff --git
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendController.java
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendController.java
index f748f58..510410c 100644
---
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendController.java
+++
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendController.java
@@ -24,15 +24,13 @@ 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.Options;
-import org.apache.streampipes.sdk.helpers.OutputStrategies;
+import org.apache.streampipes.sdk.helpers.*;
import org.apache.streampipes.sdk.utils.Assets;
import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
+import java.util.List;
+
public class TrendController extends
StandaloneEventProcessingDeclarer<TrendParameters> {
private static final String Mapping = "mapping";
@@ -65,7 +63,13 @@ public class TrendController extends
StandaloneEventProcessingDeclarer<TrendPara
int increase = extractor.singleValueParameter(Increase, Integer.class);
int duration = extractor.singleValueParameter(Duration, Integer.class);
String mapping = extractor.mappingPropertyValue(Mapping);
- TrendParameters params = new TrendParameters(invocationGraph,
getOperation(operation), increase, duration, mapping);
+ List<String> outputFieldSelectors = extractor.outputKeySelectors();
+ TrendParameters params = new TrendParameters(invocationGraph,
+ getOperation(operation),
+ increase,
+ duration,
+ mapping,
+ outputFieldSelectors);
return new ConfiguredEventProcessor<>(params, Trend::new);
}
diff --git
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendParameters.java
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendParameters.java
index c1548af..e98c8dd 100644
---
a/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendParameters.java
+++
b/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendParameters.java
@@ -21,6 +21,8 @@ package org.apache.streampipes.processors.siddhi.trend;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
import
org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams;
+import java.util.List;
+
public class TrendParameters extends EventProcessorBindingParams {
private TrendOperator operation;
@@ -28,36 +30,37 @@ public class TrendParameters extends
EventProcessorBindingParams {
private int duration;
private String mapping;
+ private List<String> outputFieldSelectors;
public TrendParameters(DataProcessorInvocation invocationGraph,
TrendOperator operation, int increase, int
duration,
- String mapping) {
+ String mapping, List<String>
outputFieldSelectors) {
super(invocationGraph);
this.operation = operation;
this.increase = increase;
this.duration = duration;
this.mapping = mapping;
+ this.outputFieldSelectors = outputFieldSelectors;
}
-
public TrendOperator getOperation() {
return operation;
}
-
public int getIncrease() {
return increase;
}
-
public int getDuration() {
return duration;
}
-
public String getMapping() {
return mapping;
}
+ public List<String> getOutputFieldSelectors() {
+ return outputFieldSelectors;
+ }
}
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 1a7199d..9a5f033 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
@@ -17,13 +17,6 @@
*/
package org.apache.streampipes.processors.siddhi.trend;
-import static org.junit.Assert.assertEquals;
-
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.apache.streampipes.model.graph.DataProcessorDescription;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
import org.apache.streampipes.model.runtime.Event;
@@ -34,13 +27,15 @@ import org.apache.streampipes.sdk.helpers.Tuple2;
import org.apache.streampipes.test.generator.EventStreamGenerator;
import org.apache.streampipes.test.generator.InvocationGraphGenerator;
import org.apache.streampipes.test.generator.grounding.EventGroundingGenerator;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.*;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import static org.junit.Assert.assertEquals;
@RunWith(Parameterized.class)
public class TestTrendProcessor {
@@ -112,7 +107,7 @@ public class TestTrendProcessor {
graph.getOutputStream().getEventGrounding().getTransportProtocol().getTopicDefinition().setActualTopicName("output-topic");
TrendParameters params = new TrendParameters(graph, trendOperator,
increase, timeWindow, "s0" +
- "::randomValue");
+ "::randomValue", Collections.emptyList());
Trend trend = new Trend(event -> actualMatchCount[0]++);
trend.onInvocation(params, null, null);
