This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch
1624-migrate-data-processors-in-streampipes-processors-siddhi-module
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/1624-migrate-data-processors-in-streampipes-processors-siddhi-module
by this push:
new 15c72c8ef Migrate processors in siddhi module (#1624)
15c72c8ef is described below
commit 15c72c8efb3b3059d2aa47ee59c79ad86037405b
Author: Dominik Riemer <[email protected]>
AuthorDate: Sat May 27 13:41:14 2023 +0200
Migrate processors in siddhi module (#1624)
---
.../processors/siddhi/FiltersSiddhiInit.java | 10 +-
.../processors/siddhi/filter/NumericalFilter.java | 85 -----------
.../siddhi/filter/NumericalFilterParameters.java | 50 -------
...er.java => NumericalFilterSiddhiProcessor.java} | 61 ++++++--
.../processors/siddhi/frequency/Frequency.java | 54 -------
.../siddhi/frequency/FrequencyParameters.java | 50 -------
...ntroller.java => FrequencySiddhiProcessor.java} | 41 ++++--
.../siddhi/frequencychange/FrequencyChange.java | 54 -------
.../frequencychange/FrequencyChangeParameters.java | 61 --------
...er.java => FrequencyChangeSiddhiProcessor.java} | 40 ++++--
.../processors/siddhi/sequence/Sequence.java | 57 --------
.../siddhi/sequence/SequenceParameters.java | 40 ------
...ontroller.java => SequenceSiddhiProcessor.java} | 42 ++++--
.../processors/siddhi/stop/StreamStop.java | 60 --------
.../siddhi/stop/StreamStopParameters.java | 40 ------
...troller.java => StreamStopSiddhiProcessor.java} | 41 ++++--
.../processors/siddhi/trend/TrendController.java | 91 ------------
.../processors/siddhi/trend/TrendOperator.java | 13 +-
.../processors/siddhi/trend/TrendParameters.java | 66 ---------
.../trend/{Trend.java => TrendProcessor.java} | 105 ++++++++++----
.../siddhi/trend/TestTrendProcessor.java | 31 +++-
.../siddhi/trend/TrendConfigurationVisitor.java | 156 +++++++++++++++++++++
22 files changed, 445 insertions(+), 803 deletions(-)
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/FiltersSiddhiInit.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/FiltersSiddhiInit.java
index c181382b2..06e727215 100644
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/FiltersSiddhiInit.java
+++
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/FiltersSiddhiInit.java
@@ -28,11 +28,11 @@ import
org.apache.streampipes.messaging.jms.SpJmsProtocolFactory;
import org.apache.streampipes.messaging.kafka.SpKafkaProtocolFactory;
import org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory;
import org.apache.streampipes.processors.siddhi.count.CountAggregation;
-import
org.apache.streampipes.processors.siddhi.filter.NumericalFilterController;
+import
org.apache.streampipes.processors.siddhi.filter.NumericalFilterSiddhiProcessor;
import org.apache.streampipes.processors.siddhi.listcollector.ListCollector;
import org.apache.streampipes.processors.siddhi.listfilter.ListFilter;
import org.apache.streampipes.processors.siddhi.topk.TopK;
-import org.apache.streampipes.processors.siddhi.trend.TrendController;
+import org.apache.streampipes.processors.siddhi.trend.TrendProcessor;
import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter;
public class FiltersSiddhiInit extends ExtensionsModelSubmitter {
@@ -48,13 +48,13 @@ public class FiltersSiddhiInit extends
ExtensionsModelSubmitter {
"",
8090)
.registerPipelineElements(
- new TrendController(),
- new NumericalFilterController(),
+ new TrendProcessor(),
+ new NumericalFilterSiddhiProcessor(),
new ListFilter(),
new ListCollector(),
new CountAggregation(),
new TopK())
- // Currently not working: StreamStopController,
FrequencyChangeController, FrequencyController
+ // Currently not working: StreamStopSiddhiProcessor,
FrequencyChangeSiddhiProcessor, FrequencySiddhiProcessor
.registerMessagingFormats(
new JsonDataFormatFactory(),
new CborDataFormatFactory(),
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilter.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilter.java
deleted file mode 100644
index 048b00385..000000000
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilter.java
+++ /dev/null
@@ -1,85 +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;
-
-import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfig;
-import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfigBuilder;
-import org.apache.streampipes.wrapper.siddhi.SiddhiQueryBuilder;
-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.InsertIntoClause;
-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> {
-
- public NumericalFilter() {
- super();
- }
-
- public NumericalFilter(SiddhiDebugCallback callback) {
- super(callback);
- }
-
- private Expression makeProperty(String fieldName) {
- return Expressions.property(SiddhiStreamSelector.FIRST_INPUT_STREAM,
fieldName);
- }
-
- @Override
- public SiddhiAppConfig
makeStatements(SiddhiProcessorParams<NumericalFilterParameters> siddhiParams,
- String finalInsertIntoStreamName) {
- NumericalFilterParameters filterParameters = siddhiParams.getParams();
- String filterProperty = filterParameters.getFilterProperty();
- RelationalOperator operator = filterParameters.getFilterOperator();
- Double threshold = filterParameters.getThreshold();
-
- // 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()
+"]";
-
- 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);
-
- SelectClause selectClause = SelectClause.create();
- siddhiParams
- .getOutputEventKeys()
- .forEach(fieldName ->
selectClause.addProperty(makeProperty(fieldName)));
-
- InsertIntoClause insertIntoClause =
InsertIntoClause.create(finalInsertIntoStreamName);
-
- return SiddhiAppConfigBuilder
- .create()
- .addQuery(SiddhiQueryBuilder
- .create(fromClause, insertIntoClause)
- .withSelectClause(selectClause)
- .build())
- .build();
- }
-}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilterParameters.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilterParameters.java
deleted file mode 100644
index a92817119..000000000
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilterParameters.java
+++ /dev/null
@@ -1,50 +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;
-
-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 RelationalOperator filterOperator;
- private String filterProperty;
-
- public NumericalFilterParameters(DataProcessorInvocation graph, Double
threshold, RelationalOperator
- filterOperator, String filterProperty) {
- super(graph);
- this.threshold = threshold;
- this.filterOperator = filterOperator;
- this.filterProperty = filterProperty;
- }
-
- public double getThreshold() {
- return threshold;
- }
-
- public RelationalOperator getFilterOperator() {
- return filterOperator;
- }
-
- public String getFilterProperty() {
- return filterProperty;
- }
-}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilterController.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilterSiddhiProcessor.java
similarity index 57%
rename from
streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilterController.java
rename to
streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilterSiddhiProcessor.java
index 6bc5488c3..8d4405673 100644
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilterController.java
+++
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/filter/NumericalFilterSiddhiProcessor.java
@@ -19,22 +19,31 @@ package org.apache.streampipes.processors.siddhi.filter;
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.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.utils.Assets;
+import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfig;
+import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfigBuilder;
+import org.apache.streampipes.wrapper.siddhi.SiddhiQueryBuilder;
+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.InsertIntoClause;
+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.standalone.ConfiguredEventProcessor;
-import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
+import
org.apache.streampipes.wrapper.siddhi.query.expression.RelationalOperatorExpression;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
-public class NumericalFilterController extends
StandaloneEventProcessingDeclarer<NumericalFilterParameters> {
+public class NumericalFilterSiddhiProcessor extends StreamPipesSiddhiProcessor
{
private static final String NUMBER_MAPPING = "number-mapping";
private static final String VALUE = "value";
@@ -58,11 +67,15 @@ public class NumericalFilterController extends
StandaloneEventProcessingDeclarer
.build();
}
+ private Expression makeProperty(String fieldName) {
+ return Expressions.property(SiddhiStreamSelector.FIRST_INPUT_STREAM,
fieldName);
+ }
+
@Override
- public ConfiguredEventProcessor<NumericalFilterParameters> onInvocation(
- DataProcessorInvocation graph,
- ProcessingElementParameterExtractor extractor) {
+ public SiddhiAppConfig makeStatements(SiddhiProcessorParams<ProcessorParams>
siddhiParams,
+ String finalInsertIntoStreamName) {
+ var extractor = siddhiParams.getParams().extractor();
Double threshold = extractor.singleValueParameter(VALUE, Double.class);
String stringOperation = extractor.selectedSingleValue(OPERATION,
String.class);
@@ -82,12 +95,32 @@ public class NumericalFilterController extends
StandaloneEventProcessingDeclarer
String filterProperty = extractor.mappingPropertyValue(NUMBER_MAPPING);
- NumericalFilterParameters staticParam = new
NumericalFilterParameters(graph,
- threshold,
- operator,
- filterProperty);
+ // 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 new ConfiguredEventProcessor<>(staticParam, NumericalFilter::new);
- }
+ 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);
+ SelectClause selectClause = SelectClause.create();
+ siddhiParams
+ .getOutputEventKeys()
+ .forEach(fieldName ->
selectClause.addProperty(makeProperty(fieldName)));
+
+ InsertIntoClause insertIntoClause =
InsertIntoClause.create(finalInsertIntoStreamName);
+
+ return SiddhiAppConfigBuilder
+ .create()
+ .addQuery(SiddhiQueryBuilder
+ .create(fromClause, insertIntoClause)
+ .withSelectClause(selectClause)
+ .build())
+ .build();
+ }
}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequency/Frequency.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequency/Frequency.java
deleted file mode 100644
index 15e10f4f8..000000000
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequency/Frequency.java
+++ /dev/null
@@ -1,54 +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.frequency;
-
-import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfig;
-import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfigBuilder;
-import org.apache.streampipes.wrapper.siddhi.SiddhiQueryBuilder;
-import org.apache.streampipes.wrapper.siddhi.engine.SiddhiEventEngine;
-import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
-import org.apache.streampipes.wrapper.siddhi.query.InsertIntoClause;
-import org.apache.streampipes.wrapper.siddhi.query.SelectClause;
-
-public class Frequency extends SiddhiEventEngine<FrequencyParameters> {
-
- private String fromStatement(SiddhiProcessorParams<FrequencyParameters>
siddhiParams) {
- return "from every not "
- + siddhiParams.getInputStreamNames().get(0)
- + " for "
- + siddhiParams.getParams().getDuration() + " sec";
- }
-
- private SelectClause
selectStatement(SiddhiProcessorParams<FrequencyParameters> siddhiParams) {
- return SelectClause.createWildcard();
- }
-
- @Override
- public SiddhiAppConfig
makeStatements(SiddhiProcessorParams<FrequencyParameters> siddhiParams,
- String finalInsertIntoStreamName) {
-
- InsertIntoClause insertIntoClause =
InsertIntoClause.create(finalInsertIntoStreamName);
- return SiddhiAppConfigBuilder
- .create()
- .addQuery(SiddhiQueryBuilder
- .create(fromStatement(siddhiParams), insertIntoClause)
- .withSelectClause(selectStatement(siddhiParams))
- .build())
- .build();
- }
-}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequency/FrequencyParameters.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequency/FrequencyParameters.java
deleted file mode 100644
index 53b65480a..000000000
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequency/FrequencyParameters.java
+++ /dev/null
@@ -1,50 +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.frequency;
-
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import
org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams;
-
-public class FrequencyParameters extends EventProcessorBindingParams {
-
- private int duration;
- private String timeUnit;
-
- public FrequencyParameters(DataProcessorInvocation graph, int duration,
String timeUnit) {
- super(graph);
- this.duration = duration;
- this.timeUnit = timeUnit;
- }
-
- public int getDuration() {
- return duration;
- }
-
- public void setDuration(int duration) {
- this.duration = duration;
- }
-
- public String getTimeUnit() {
- return timeUnit;
- }
-
- public void setTimeUnit(String timeUnit) {
- this.timeUnit = timeUnit;
- }
-}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequency/FrequencyController.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequency/FrequencySiddhiProcessor.java
similarity index 59%
rename from
streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequency/FrequencyController.java
rename to
streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequency/FrequencySiddhiProcessor.java
index 4895ccacd..3438aebcd 100644
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequency/FrequencyController.java
+++
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequency/FrequencySiddhiProcessor.java
@@ -19,20 +19,24 @@ package org.apache.streampipes.processors.siddhi.frequency;
import org.apache.streampipes.model.DataProcessorType;
import org.apache.streampipes.model.graph.DataProcessorDescription;
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
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.utils.Assets;
-import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
-import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
+import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfig;
+import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfigBuilder;
+import org.apache.streampipes.wrapper.siddhi.SiddhiQueryBuilder;
+import org.apache.streampipes.wrapper.siddhi.engine.StreamPipesSiddhiProcessor;
+import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
+import org.apache.streampipes.wrapper.siddhi.query.InsertIntoClause;
+import org.apache.streampipes.wrapper.siddhi.query.SelectClause;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
-public class FrequencyController extends
StandaloneEventProcessingDeclarer<FrequencyParameters> {
+public class FrequencySiddhiProcessor extends StreamPipesSiddhiProcessor {
private static final String DURATION = "duration";
private static final String TIME_UNIT = "timeUnit";
@@ -55,15 +59,32 @@ public class FrequencyController extends
StandaloneEventProcessingDeclarer<Frequ
}
@Override
- public ConfiguredEventProcessor<FrequencyParameters>
onInvocation(DataProcessorInvocation graph,
-
ProcessingElementParameterExtractor extractor) {
-
+ public SiddhiAppConfig makeStatements(SiddhiProcessorParams<ProcessorParams>
siddhiParams,
+ String finalInsertIntoStreamName) {
+ var extractor = siddhiParams.getParams().extractor();
int duration = extractor.singleValueParameter(DURATION, Integer.class);
String timeUnit = extractor.selectedSingleValue(TIME_UNIT, String.class);
- FrequencyParameters staticParam = new FrequencyParameters(graph, duration,
timeUnit);
+ InsertIntoClause insertIntoClause =
InsertIntoClause.create(finalInsertIntoStreamName);
+ return SiddhiAppConfigBuilder
+ .create()
+ .addQuery(SiddhiQueryBuilder
+ .create(fromStatement(siddhiParams, duration, timeUnit),
insertIntoClause)
+ .withSelectClause(selectStatement())
+ .build())
+ .build();
+ }
- return new ConfiguredEventProcessor<>(staticParam, Frequency::new);
+ private String fromStatement(SiddhiProcessorParams<ProcessorParams> params,
+ int duration,
+ String timeUnit) {
+ return "from every not "
+ + params.getInputStreamNames().get(0)
+ + " for "
+ + duration + " " + timeUnit;
}
+ private SelectClause selectStatement() {
+ return SelectClause.createWildcard();
+ }
}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequencychange/FrequencyChange.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequencychange/FrequencyChange.java
deleted file mode 100644
index 86ce20348..000000000
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequencychange/FrequencyChange.java
+++ /dev/null
@@ -1,54 +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.frequencychange;
-
-import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfig;
-import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfigBuilder;
-import org.apache.streampipes.wrapper.siddhi.SiddhiQueryBuilder;
-import org.apache.streampipes.wrapper.siddhi.engine.SiddhiEventEngine;
-import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
-import org.apache.streampipes.wrapper.siddhi.query.InsertIntoClause;
-import org.apache.streampipes.wrapper.siddhi.query.SelectClause;
-
-public class FrequencyChange extends
SiddhiEventEngine<FrequencyChangeParameters> {
-
- private String
fromStatement(SiddhiProcessorParams<FrequencyChangeParameters> siddhiParams) {
- return "from every not "
- + siddhiParams.getInputStreamNames().get(0)
- + " for "
- + siddhiParams.getParams().getDuration()
- + " sec";
- }
-
- private String
selectStatement(SiddhiProcessorParams<FrequencyChangeParameters> siddhiParams) {
- return SelectClause.createWildcard().toSiddhiEpl();
- }
-
- @Override
- public SiddhiAppConfig
makeStatements(SiddhiProcessorParams<FrequencyChangeParameters> siddhiParams,
- String finalInsertIntoStreamName) {
-
- InsertIntoClause insertIntoClause =
InsertIntoClause.create(finalInsertIntoStreamName);
- return SiddhiAppConfigBuilder
- .create()
- .addQuery(SiddhiQueryBuilder.create(fromStatement(siddhiParams),
insertIntoClause)
- .withSelectClause(selectStatement(siddhiParams))
- .build())
- .build();
- }
-}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequencychange/FrequencyChangeParameters.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequencychange/FrequencyChangeParameters.java
deleted file mode 100644
index b9c129ef1..000000000
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequencychange/FrequencyChangeParameters.java
+++ /dev/null
@@ -1,61 +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.frequencychange;
-
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import
org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams;
-
-public class FrequencyChangeParameters extends EventProcessorBindingParams {
-
- private int duration;
- private String timeUnit;
- private int increase;
-
-
- public FrequencyChangeParameters(DataProcessorInvocation graph, int
duration, String timeUnit, int increase) {
- super(graph);
- this.duration = duration;
- this.timeUnit = timeUnit;
- this.increase = increase;
- }
-
- public int getDuration() {
- return duration;
- }
-
- public void setDuration(int duration) {
- this.duration = duration;
- }
-
- public String getTimeUnit() {
- return timeUnit;
- }
-
- public void setTimeUnit(String timeUnit) {
- this.timeUnit = timeUnit;
- }
-
- public int getIncrease() {
- return increase;
- }
-
- public void setIncrease(int increase) {
- this.increase = increase;
- }
-}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequencychange/FrequencyChangeController.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequencychange/FrequencyChangeSiddhiProcessor.java
similarity index 61%
rename from
streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequencychange/FrequencyChangeController.java
rename to
streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequencychange/FrequencyChangeSiddhiProcessor.java
index 7a9a386dd..b8b7718b5 100644
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequencychange/FrequencyChangeController.java
+++
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/frequencychange/FrequencyChangeSiddhiProcessor.java
@@ -19,20 +19,24 @@ package
org.apache.streampipes.processors.siddhi.frequencychange;
import org.apache.streampipes.model.DataProcessorType;
import org.apache.streampipes.model.graph.DataProcessorDescription;
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
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.utils.Assets;
-import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
-import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
+import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfig;
+import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfigBuilder;
+import org.apache.streampipes.wrapper.siddhi.SiddhiQueryBuilder;
+import org.apache.streampipes.wrapper.siddhi.engine.StreamPipesSiddhiProcessor;
+import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
+import org.apache.streampipes.wrapper.siddhi.query.InsertIntoClause;
+import org.apache.streampipes.wrapper.siddhi.query.SelectClause;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
-public class FrequencyChangeController extends
StandaloneEventProcessingDeclarer<FrequencyChangeParameters> {
+public class FrequencyChangeSiddhiProcessor extends StreamPipesSiddhiProcessor
{
private static final String DURATION = "duration";
private static final String TIME_UNIT = "timeUnit";
@@ -58,17 +62,33 @@ public class FrequencyChangeController extends
StandaloneEventProcessingDeclarer
}
@Override
- public ConfiguredEventProcessor<FrequencyChangeParameters> onInvocation(
- DataProcessorInvocation graph,
- ProcessingElementParameterExtractor extractor) {
+ public SiddhiAppConfig makeStatements(SiddhiProcessorParams<ProcessorParams>
siddhiParams,
+ String finalInsertIntoStreamName) {
+ var extractor = siddhiParams.getParams().extractor();
int duration = extractor.singleValueParameter(DURATION, Integer.class);
String timeUnit = extractor.selectedSingleValue(TIME_UNIT, String.class);
int increase = extractor.selectedSingleValue(INCREASE, Integer.class);
- FrequencyChangeParameters staticParam = new
FrequencyChangeParameters(graph, duration, timeUnit, increase);
+ InsertIntoClause insertIntoClause =
InsertIntoClause.create(finalInsertIntoStreamName);
+ return SiddhiAppConfigBuilder
+ .create()
+ .addQuery(SiddhiQueryBuilder.create(fromStatement(siddhiParams,
duration), insertIntoClause)
+ .withSelectClause(selectStatement(siddhiParams))
+ .build())
+ .build();
+ }
- return new ConfiguredEventProcessor<>(staticParam, FrequencyChange::new);
+ private String fromStatement(SiddhiProcessorParams<ProcessorParams>
siddhiParams,
+ int duration) {
+ return "from every not "
+ + siddhiParams.getInputStreamNames().get(0)
+ + " for "
+ + duration
+ + " sec";
}
+ private String selectStatement(SiddhiProcessorParams<ProcessorParams>
siddhiParams) {
+ return SelectClause.createWildcard().toSiddhiEpl();
+ }
}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/sequence/Sequence.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/sequence/Sequence.java
deleted file mode 100644
index f5f02785b..000000000
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/sequence/Sequence.java
+++ /dev/null
@@ -1,57 +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.sequence;
-
-import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfig;
-import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfigBuilder;
-import org.apache.streampipes.wrapper.siddhi.SiddhiQueryBuilder;
-import org.apache.streampipes.wrapper.siddhi.engine.SiddhiEventEngine;
-import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
-import org.apache.streampipes.wrapper.siddhi.query.InsertIntoClause;
-
-public class Sequence extends SiddhiEventEngine<SequenceParameters> {
-
- private String fromStatement(SiddhiProcessorParams<SequenceParameters>
siddhiParams) {
-
- return "from every not "
- + siddhiParams.getInputStreamNames().get(0)
- + " for "
- + siddhiParams.getParams().getDuration()
- + " sec";
- }
-
- private String selectStatement(SiddhiProcessorParams<SequenceParameters>
siddhiParams) {
-// return getCustomOutputSelectStatement(params.getGraph());
- return "select *";
-// return "select currentTimeMillis() as s0timestamp, 'Customer has not
arrived' as message";
- }
-
- @Override
- public SiddhiAppConfig
makeStatements(SiddhiProcessorParams<SequenceParameters> siddhiParams,
- String finalInsertIntoStreamName) {
-
- InsertIntoClause insertIntoClause =
InsertIntoClause.create(finalInsertIntoStreamName);
- return SiddhiAppConfigBuilder
- .create()
- .addQuery(SiddhiQueryBuilder
- .create(fromStatement(siddhiParams), insertIntoClause)
- .withSelectClause(selectStatement(siddhiParams))
- .build())
- .build();
- }
-}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/sequence/SequenceParameters.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/sequence/SequenceParameters.java
deleted file mode 100644
index 9d085c976..000000000
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/sequence/SequenceParameters.java
+++ /dev/null
@@ -1,40 +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.sequence;
-
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import
org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams;
-
-public class SequenceParameters extends EventProcessorBindingParams {
-
- private int duration;
-
- public SequenceParameters(DataProcessorInvocation graph, int duration) {
- super(graph);
- this.duration = duration;
- }
-
- public int getDuration() {
- return duration;
- }
-
- public void setDuration(int duration) {
- this.duration = duration;
- }
-}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/sequence/SequenceController.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/sequence/SequenceSiddhiProcessor.java
similarity index 58%
rename from
streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/sequence/SequenceController.java
rename to
streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/sequence/SequenceSiddhiProcessor.java
index 15cb0be43..2c1cd5578 100644
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/sequence/SequenceController.java
+++
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/sequence/SequenceSiddhiProcessor.java
@@ -19,19 +19,23 @@ package org.apache.streampipes.processors.siddhi.sequence;
import org.apache.streampipes.model.DataProcessorType;
import org.apache.streampipes.model.graph.DataProcessorDescription;
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
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.standalone.ConfiguredEventProcessor;
-import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
+import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfig;
+import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfigBuilder;
+import org.apache.streampipes.wrapper.siddhi.SiddhiQueryBuilder;
+import org.apache.streampipes.wrapper.siddhi.engine.StreamPipesSiddhiProcessor;
+import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
+import org.apache.streampipes.wrapper.siddhi.query.InsertIntoClause;
+import org.apache.streampipes.wrapper.siddhi.query.SelectClause;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
-public class SequenceController extends
StandaloneEventProcessingDeclarer<SequenceParameters> {
+public class SequenceSiddhiProcessor extends StreamPipesSiddhiProcessor {
private static final String Duration = "duration";
@@ -56,14 +60,32 @@ public class SequenceController extends
StandaloneEventProcessingDeclarer<Sequen
}
@Override
- public ConfiguredEventProcessor<SequenceParameters>
onInvocation(DataProcessorInvocation graph,
-
ProcessingElementParameterExtractor extractor) {
-
+ public SiddhiAppConfig makeStatements(SiddhiProcessorParams<ProcessorParams>
siddhiParams,
+ String finalInsertIntoStreamName) {
+ var extractor = siddhiParams.getParams().extractor();
int duration = extractor.singleValueParameter(Duration, Integer.class);
- SequenceParameters staticParam = new SequenceParameters(graph, duration);
+ InsertIntoClause insertIntoClause =
InsertIntoClause.create(finalInsertIntoStreamName);
+ return SiddhiAppConfigBuilder
+ .create()
+ .addQuery(SiddhiQueryBuilder
+ .create(fromStatement(siddhiParams, duration), insertIntoClause)
+ .withSelectClause(selectStatement(siddhiParams))
+ .build())
+ .build();
+ }
+
+ private String fromStatement(SiddhiProcessorParams<ProcessorParams>
siddhiParams,
+ int duration) {
- return new ConfiguredEventProcessor<>(staticParam, Sequence::new);
+ return "from every not "
+ + siddhiParams.getInputStreamNames().get(0)
+ + " for "
+ + duration
+ + " sec";
}
+ private SelectClause selectStatement(SiddhiProcessorParams<ProcessorParams>
siddhiParams) {
+ return SelectClause.createWildcard();
+ }
}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/stop/StreamStop.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/stop/StreamStop.java
deleted file mode 100644
index 93363dae3..000000000
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/stop/StreamStop.java
+++ /dev/null
@@ -1,60 +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.stop;
-
-import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfig;
-import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfigBuilder;
-import org.apache.streampipes.wrapper.siddhi.SiddhiQueryBuilder;
-import org.apache.streampipes.wrapper.siddhi.engine.SiddhiEventEngine;
-import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
-import org.apache.streampipes.wrapper.siddhi.query.InsertIntoClause;
-
-public class StreamStop extends SiddhiEventEngine<StreamStopParameters> {
-
- public StreamStop() {
- super();
- }
-
- private String fromStatement(SiddhiProcessorParams<StreamStopParameters>
siddhiParams) {
- return "define stream Test(timestamp LONG,message STRING);\n"
- + "from every not "
- + siddhiParams.getInputStreamNames().get(0)
- + " for " + siddhiParams.getParams().getDuration()
- + " sec";
- }
-
- private String selectStatement(SiddhiProcessorParams<StreamStopParameters>
siddhiParams) {
- //setSortedEventKeys(Arrays.asList("timestamp", "message"));
- return "select currentTimeMillis() as timestamp, 'Event stream has
stopped' as message";
- }
-
- @Override
- public SiddhiAppConfig
makeStatements(SiddhiProcessorParams<StreamStopParameters> siddhiParams,
- String finalInsertIntoStreamName) {
-
- InsertIntoClause insertIntoClause =
InsertIntoClause.create(finalInsertIntoStreamName);
-
- return SiddhiAppConfigBuilder
- .create()
- .addQuery(SiddhiQueryBuilder
- .create(fromStatement(siddhiParams), insertIntoClause)
- .withSelectClause(selectStatement(siddhiParams))
- .build())
- .build();
- }
-}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/stop/StreamStopParameters.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/stop/StreamStopParameters.java
deleted file mode 100644
index 311ae2eaa..000000000
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/stop/StreamStopParameters.java
+++ /dev/null
@@ -1,40 +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.stop;
-
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
-import
org.apache.streampipes.wrapper.params.binding.EventProcessorBindingParams;
-
-public class StreamStopParameters extends EventProcessorBindingParams {
-
- private int duration;
-
- public StreamStopParameters(DataProcessorInvocation graph, int duration) {
- super(graph);
- this.duration = duration;
- }
-
- public int getDuration() {
- return duration;
- }
-
- public void setDuration(int duration) {
- this.duration = duration;
- }
-}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/stop/StreamStopController.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/stop/StreamStopSiddhiProcessor.java
similarity index 59%
rename from
streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/stop/StreamStopController.java
rename to
streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/stop/StreamStopSiddhiProcessor.java
index b8f767075..6b16c2dd1 100644
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/stop/StreamStopController.java
+++
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/stop/StreamStopSiddhiProcessor.java
@@ -19,22 +19,25 @@ package org.apache.streampipes.processors.siddhi.stop;
import org.apache.streampipes.model.DataProcessorType;
import org.apache.streampipes.model.graph.DataProcessorDescription;
-import org.apache.streampipes.model.graph.DataProcessorInvocation;
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.EpProperties;
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.standalone.ConfiguredEventProcessor;
-import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
+import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfig;
+import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfigBuilder;
+import org.apache.streampipes.wrapper.siddhi.SiddhiQueryBuilder;
+import org.apache.streampipes.wrapper.siddhi.engine.StreamPipesSiddhiProcessor;
+import org.apache.streampipes.wrapper.siddhi.model.SiddhiProcessorParams;
+import org.apache.streampipes.wrapper.siddhi.query.InsertIntoClause;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
import java.util.Arrays;
-public class StreamStopController extends
StandaloneEventProcessingDeclarer<StreamStopParameters> {
+public class StreamStopSiddhiProcessor extends StreamPipesSiddhiProcessor {
private static final String Duration = "duration";
private static final String Message = "message";
@@ -61,14 +64,32 @@ public class StreamStopController extends
StandaloneEventProcessingDeclarer<Stre
}
@Override
- public ConfiguredEventProcessor<StreamStopParameters>
onInvocation(DataProcessorInvocation graph,
-
ProcessingElementParameterExtractor extractor) {
-
+ public SiddhiAppConfig makeStatements(SiddhiProcessorParams<ProcessorParams>
siddhiParams,
+ String finalInsertIntoStreamName) {
+ var extractor = siddhiParams.getParams().extractor();
int duration = extractor.singleValueParameter(Duration, Integer.class);
- StreamStopParameters staticParam = new StreamStopParameters(graph,
duration);
+ InsertIntoClause insertIntoClause =
InsertIntoClause.create(finalInsertIntoStreamName);
+
+ return SiddhiAppConfigBuilder
+ .create()
+ .addQuery(SiddhiQueryBuilder
+ .create(fromStatement(siddhiParams, duration), insertIntoClause)
+ .withSelectClause(selectStatement(siddhiParams))
+ .build())
+ .build();
+ }
- return new ConfiguredEventProcessor<>(staticParam, StreamStop::new);
+ private String fromStatement(SiddhiProcessorParams<ProcessorParams>
siddhiParams,
+ int duration) {
+ return "define stream Test(timestamp LONG,message STRING);\n"
+ + "from every not "
+ + siddhiParams.getInputStreamNames().get(0)
+ + " for " + duration
+ + " sec";
}
+ private String selectStatement(SiddhiProcessorParams<ProcessorParams>
siddhiParams) {
+ return "select currentTimeMillis() as timestamp, 'Event stream has
stopped' as message";
+ }
}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendController.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendController.java
deleted file mode 100644
index ccc9f5ba6..000000000
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendController.java
+++ /dev/null
@@ -1,91 +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.trend;
-
-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.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.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";
- private static final String Increase = "increase";
- private static final String Operation = "operation";
- private static final String Duration = "duration";
-
- @Override
- public DataProcessorDescription declareModel() {
- return
ProcessingElementBuilder.create("org.apache.streampipes.processors.siddhi.increase")
- .withLocales(Locales.EN)
- .category(DataProcessorType.PATTERN_DETECT)
- .withAssets(Assets.DOCUMENTATION, Assets.ICON)
- .requiredStream(StreamRequirementsBuilder.create()
- .requiredPropertyWithUnaryMapping(EpRequirements.numberReq(),
Labels.withId
- (Mapping), PropertyScope.MEASUREMENT_PROPERTY)
- .build())
- .requiredSingleValueSelection(Labels.withId(Operation), Options
- .from("Increase", "Decrease"))
- .requiredIntegerParameter(Labels.withId(Increase), 0, 500, 1)
- .requiredIntegerParameter(Labels.withId(Duration))
- .outputStrategy(OutputStrategies.custom())
- .build();
- }
-
- @Override
- public ConfiguredEventProcessor<TrendParameters>
onInvocation(DataProcessorInvocation
-
invocationGraph,
-
ProcessingElementParameterExtractor extractor) {
- String operation = extractor.selectedSingleValue(Operation, String.class);
- int increase = extractor.singleValueParameter(Increase, Integer.class);
- int duration = extractor.singleValueParameter(Duration, Integer.class);
- String mapping = extractor.mappingPropertyValue(Mapping);
- List<String> outputFieldSelectors = extractor.outputKeySelectors();
- TrendParameters params = new TrendParameters(invocationGraph,
- getOperation(operation),
- increase,
- duration,
- mapping,
- outputFieldSelectors);
-
- return new ConfiguredEventProcessor<>(params, Trend::new);
- }
-
- private TrendOperator getOperation(String operation) {
- if (operation.equals("Increase")) {
- return TrendOperator.INCREASE;
- } else {
- return TrendOperator.DECREASE;
- }
- }
-
-
-}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendOperator.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendOperator.java
index 6dad7c618..c5a915b16 100644
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendOperator.java
+++
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendOperator.java
@@ -19,5 +19,16 @@
package org.apache.streampipes.processors.siddhi.trend;
public enum TrendOperator {
- INCREASE, DECREASE
+ INCREASE("Increase"),
+ DECREASE("Decrease");
+
+ private String label;
+
+ TrendOperator(String label) {
+ this.label = label;
+ }
+
+ public String getLabel() {
+ return label;
+ }
}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendParameters.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendParameters.java
deleted file mode 100644
index ae3df9dd0..000000000
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendParameters.java
+++ /dev/null
@@ -1,66 +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.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;
- private int increase;
- private int duration;
-
- private String mapping;
- private List<String> outputFieldSelectors;
-
-
- public TrendParameters(DataProcessorInvocation invocationGraph,
- TrendOperator operation, int increase, int duration,
- 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-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/Trend.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendProcessor.java
similarity index 55%
rename from
streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/Trend.java
rename to
streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendProcessor.java
index 30a707316..d62811913 100644
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/Trend.java
+++
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/trend/TrendProcessor.java
@@ -17,10 +17,21 @@
*/
package org.apache.streampipes.processors.siddhi.trend;
+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.Options;
+import org.apache.streampipes.sdk.helpers.OutputStrategies;
+import org.apache.streampipes.sdk.utils.Assets;
import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfig;
import org.apache.streampipes.wrapper.siddhi.SiddhiAppConfigBuilder;
import org.apache.streampipes.wrapper.siddhi.SiddhiQueryBuilder;
-import org.apache.streampipes.wrapper.siddhi.engine.SiddhiEventEngine;
+import org.apache.streampipes.wrapper.siddhi.engine.StreamPipesSiddhiProcessor;
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;
@@ -34,25 +45,78 @@ import
org.apache.streampipes.wrapper.siddhi.query.expression.SiddhiTimeUnit;
import org.apache.streampipes.wrapper.siddhi.query.expression.StreamExpression;
import
org.apache.streampipes.wrapper.siddhi.query.expression.StreamFilterExpression;
import
org.apache.streampipes.wrapper.siddhi.query.expression.pattern.PatternCountOperator;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
import java.util.List;
-public class Trend extends SiddhiEventEngine<TrendParameters> {
+import static
org.apache.streampipes.wrapper.siddhi.utils.SiddhiUtils.prepareName;
- public Trend() {
+public class TrendProcessor extends StreamPipesSiddhiProcessor {
+
+ private static final String Mapping = "mapping";
+
+ public static final String INCREASE = "increase";
+ public static final String OPERATION = "operation";
+ public static final String DURATION = "duration";
+
+ public TrendProcessor() {
super();
}
- public Trend(SiddhiDebugCallback callback) {
+ public TrendProcessor(SiddhiDebugCallback callback) {
super(callback);
}
- public FromClause fromStatement(SiddhiProcessorParams<TrendParameters>
siddhiParams) {
- TrendParameters trendParameters = siddhiParams.getParams();
+ @Override
+ public DataProcessorDescription declareModel() {
+ return
ProcessingElementBuilder.create("org.apache.streampipes.processors.siddhi.increase")
+ .withLocales(Locales.EN)
+ .category(DataProcessorType.PATTERN_DETECT)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .requiredStream(StreamRequirementsBuilder.create()
+ .requiredPropertyWithUnaryMapping(EpRequirements.numberReq(),
Labels.withId
+ (Mapping), PropertyScope.MEASUREMENT_PROPERTY)
+ .build())
+ .requiredSingleValueSelection(Labels.withId(OPERATION), Options
+ .from(TrendOperator.INCREASE.getLabel(),
TrendOperator.DECREASE.getLabel()))
+ .requiredIntegerParameter(Labels.withId(INCREASE), 0, 500, 1)
+ .requiredIntegerParameter(Labels.withId(DURATION))
+ .outputStrategy(OutputStrategies.custom())
+ .build();
+ }
+
+ private TrendOperator getOperation(String operation) {
+ if (operation.equals("Increase")) {
+ return TrendOperator.INCREASE;
+ } else {
+ return TrendOperator.DECREASE;
+ }
+ }
+
+ @Override
+ public SiddhiAppConfig makeStatements(SiddhiProcessorParams<ProcessorParams>
siddhiParams,
+ String finalInsertIntoStreamName) {
- String mappingProperty = prepareName(trendParameters.getMapping());
- int duration = trendParameters.getDuration();
- double increase = trendParameters.getIncrease();
+ InsertIntoClause insertIntoClause =
InsertIntoClause.create(finalInsertIntoStreamName);
+
+ return SiddhiAppConfigBuilder
+ .create()
+ .addQuery(SiddhiQueryBuilder
+ .create(fromStatement(siddhiParams), insertIntoClause)
+ .withSelectClause(selectStatement(siddhiParams))
+ .build())
+ .build();
+ }
+
+ public FromClause fromStatement(SiddhiProcessorParams<ProcessorParams>
siddhiParams) {
+ var extractor = siddhiParams.getParams().extractor();
+ var operation = getOperation(extractor.selectedSingleValue(OPERATION,
String.class));
+ double increase = extractor.singleValueParameter(INCREASE, Double.class);
+ int duration = extractor.singleValueParameter(DURATION, Integer.class);
+ String mapping = extractor.mappingPropertyValue(Mapping);
+
+
+ String mappingProperty = prepareName(mapping);
increase = (increase / 100) + 1;
FromClause fromClause = FromClause.create();
@@ -60,11 +124,11 @@ public class Trend extends
SiddhiEventEngine<TrendParameters> {
Expressions.stream("e1", siddhiParams.getInputStreamNames().get(0)));
StreamExpression exp2 = Expressions.stream("e2",
siddhiParams.getInputStreamNames().get(0));
- PropertyExpressionBase mathExp = trendParameters.getOperation() ==
TrendOperator.INCREASE
+ PropertyExpressionBase mathExp = operation == TrendOperator.INCREASE
? Expressions.divide(Expressions.property(mappingProperty),
Expressions.staticValue(increase)) :
Expressions.multiply(Expressions.property(mappingProperty),
Expressions.staticValue(increase));
- RelationalOperatorExpression opExp = trendParameters.getOperation() ==
TrendOperator.INCREASE
+ RelationalOperatorExpression opExp = operation == TrendOperator.INCREASE
? Expressions.le(Expressions.property("e1", mappingProperty), mathExp)
:
Expressions.ge(Expressions.property("e1", mappingProperty), mathExp);
@@ -80,28 +144,13 @@ public class Trend extends
SiddhiEventEngine<TrendParameters> {
return fromClause;
}
- private SelectClause selectStatement(SiddhiProcessorParams<TrendParameters>
siddhiParams) {
+ private SelectClause selectStatement(SiddhiProcessorParams<ProcessorParams>
siddhiParams) {
+ List<String> outputFieldSelectors =
siddhiParams.getParams().extractor().outputKeySelectors();
SelectClause selectClause = SelectClause.create();
- List<String> outputFieldSelectors =
siddhiParams.getParams().getOutputFieldSelectors();
outputFieldSelectors
.forEach(outputFieldSelector -> selectClause
.addProperty(Expressions.property("e2", outputFieldSelector,
"last")));
return selectClause;
}
-
- @Override
- public SiddhiAppConfig makeStatements(SiddhiProcessorParams<TrendParameters>
siddhiParams,
- String finalInsertIntoStreamName) {
-
- InsertIntoClause insertIntoClause =
InsertIntoClause.create(finalInsertIntoStreamName);
-
- return SiddhiAppConfigBuilder
- .create()
- .addQuery(SiddhiQueryBuilder
- .create(fromStatement(siddhiParams), insertIntoClause)
- .withSelectClause(selectStatement(siddhiParams))
- .build())
- .build();
- }
}
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java
index de9c2a1d2..1f028f0e9 100644
---
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java
+++
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java
@@ -19,6 +19,7 @@ package org.apache.streampipes.processors.siddhi.trend;
import org.apache.streampipes.model.graph.DataProcessorDescription;
import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.output.CustomOutputStrategy;
import org.apache.streampipes.model.runtime.Event;
import org.apache.streampipes.model.runtime.EventFactory;
import org.apache.streampipes.model.runtime.SchemaInfo;
@@ -28,6 +29,7 @@ import
org.apache.streampipes.test.generator.EventStreamGenerator;
import org.apache.streampipes.test.generator.InvocationGraphGenerator;
import org.apache.streampipes.test.generator.grounding.EventGroundingGenerator;
import
org.apache.streampipes.wrapper.siddhi.engine.callback.SiddhiDebugCallback;
+import org.apache.streampipes.wrapper.standalone.ProcessorParams;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -41,13 +43,14 @@ import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
import static org.junit.Assert.assertEquals;
@RunWith(Parameterized.class)
public class TestTrendProcessor {
- private static final Logger LOG =
LoggerFactory.getLogger(TestTrendProcessor.class);
+ private static final Logger LOG =
LoggerFactory.getLogger(TrendProcessor.class);
@org.junit.runners.Parameterized.Parameters
public static Iterable<Object[]> data() {
@@ -100,7 +103,7 @@ public class TestTrendProcessor {
@Test
public void testTrend() {
final Integer[] actualMatchCount = {0};
- DataProcessorDescription originalGraph = new
TrendController().declareModel();
+ DataProcessorDescription originalGraph = new
TrendProcessor().declareModel();
originalGraph.setSupportedGrounding(EventGroundingGenerator.makeDummyGrounding());
DataProcessorInvocation graph =
@@ -110,12 +113,26 @@ public class TestTrendProcessor {
.singletonList(EventStreamGenerator
.makeStreamWithProperties(Collections.singletonList("randomValue"))));
+ graph.setOutputStrategies(
+ graph.getOutputStrategies()
+ .stream()
+ .filter(o -> o instanceof CustomOutputStrategy)
+ .peek(o -> ((CustomOutputStrategy)
o).setSelectedPropertyKeys(Arrays.asList("s0::randomValue")))
+ .collect(Collectors.toList())
+ );
+
graph.setOutputStream(EventStreamGenerator.makeStreamWithProperties(Collections.singletonList("randomValue")));
graph.getOutputStream().getEventGrounding().getTransportProtocol().getTopicDefinition()
.setActualTopicName("output-topic");
- TrendParameters params = new TrendParameters(graph, trendOperator,
increase, timeWindow, "s0"
- + "::randomValue", Arrays.asList("s0::randomValue"));
+
+ var visitor = new TrendConfigurationVisitor("s0::randomValue",
+ trendOperator,
+ increase,
+ timeWindow);
+ graph.getStaticProperties().forEach(sp -> sp.accept(visitor));
+
+ ProcessorParams processorParams = new ProcessorParams(graph);
SiddhiDebugCallback callback = new SiddhiDebugCallback() {
@Override
@@ -129,8 +146,8 @@ public class TestTrendProcessor {
}
};
- Trend trend = new Trend(callback);
- trend.onInvocation(params, null, null);
+ TrendProcessor trend = new TrendProcessor(callback);
+ trend.onInvocation(processorParams, null, null);
sendEvents(trend);
LOG.info("Expected match count is {}", expectedMatchCount);
@@ -138,7 +155,7 @@ public class TestTrendProcessor {
assertEquals(expectedMatchCount, actualMatchCount[0]);
}
- private void sendEvents(Trend trend) {
+ private void sendEvents(TrendProcessor trend) {
List<Tuple2<Integer, Event>> events = makeEvents();
for (Tuple2<Integer, Event> event : events) {
LOG.info("Sending event with value " +
event.v.getFieldBySelector("s0::randomValue"));
diff --git
a/streampipes-extensions/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TrendConfigurationVisitor.java
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TrendConfigurationVisitor.java
new file mode 100644
index 000000000..673f3185f
--- /dev/null
+++
b/streampipes-extensions/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TrendConfigurationVisitor.java
@@ -0,0 +1,156 @@
+/*
+ * 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.trend;
+
+import org.apache.streampipes.model.staticproperty.AnyStaticProperty;
+import org.apache.streampipes.model.staticproperty.CodeInputStaticProperty;
+import org.apache.streampipes.model.staticproperty.CollectionStaticProperty;
+import org.apache.streampipes.model.staticproperty.ColorPickerStaticProperty;
+import org.apache.streampipes.model.staticproperty.DomainStaticProperty;
+import org.apache.streampipes.model.staticproperty.FileStaticProperty;
+import org.apache.streampipes.model.staticproperty.FreeTextStaticProperty;
+import org.apache.streampipes.model.staticproperty.MappingPropertyNary;
+import org.apache.streampipes.model.staticproperty.MappingPropertyUnary;
+import org.apache.streampipes.model.staticproperty.MatchingStaticProperty;
+import org.apache.streampipes.model.staticproperty.OneOfStaticProperty;
+import org.apache.streampipes.model.staticproperty.RemoteOneOfStaticProperty;
+import
org.apache.streampipes.model.staticproperty.RuntimeResolvableTreeInputStaticProperty;
+import org.apache.streampipes.model.staticproperty.SecretStaticProperty;
+import org.apache.streampipes.model.staticproperty.SlideToggleStaticProperty;
+import org.apache.streampipes.model.staticproperty.StaticPropertyAlternative;
+import org.apache.streampipes.model.staticproperty.StaticPropertyAlternatives;
+import org.apache.streampipes.model.staticproperty.StaticPropertyGroup;
+import org.apache.streampipes.model.staticproperty.StaticPropertyVisitor;
+
+public class TrendConfigurationVisitor implements StaticPropertyVisitor {
+
+ private final String mappingPropertyValue;
+ private final TrendOperator trendOperator;
+ private final int increase;
+ private final int timeWindow;
+
+ public TrendConfigurationVisitor(String mappingPropertyValue,
+ TrendOperator trendOperator,
+ Integer increase,
+ Integer timeWindow) {
+ this.mappingPropertyValue = mappingPropertyValue;
+ this.trendOperator = trendOperator;
+ this.increase = increase;
+ this.timeWindow = timeWindow;
+ }
+
+ @Override
+ public void visit(AnyStaticProperty property) {
+
+ }
+
+ @Override
+ public void visit(CodeInputStaticProperty codeInputStaticProperty) {
+
+ }
+
+ @Override
+ public void visit(CollectionStaticProperty collectionStaticProperty) {
+
+ }
+
+ @Override
+ public void visit(ColorPickerStaticProperty colorPickerStaticProperty) {
+
+ }
+
+ @Override
+ public void visit(DomainStaticProperty domainStaticProperty) {
+
+ }
+
+ @Override
+ public void visit(FileStaticProperty fileStaticProperty) {
+
+ }
+
+ @Override
+ public void visit(FreeTextStaticProperty freeTextStaticProperty) {
+ if
(freeTextStaticProperty.getInternalName().equals(TrendProcessor.DURATION)) {
+ freeTextStaticProperty.setValue(String.valueOf(this.timeWindow));
+ } else if
(freeTextStaticProperty.getInternalName().equals(TrendProcessor.INCREASE)) {
+ freeTextStaticProperty.setValue(String.valueOf(this.increase));
+ }
+ }
+
+ @Override
+ public void visit(MappingPropertyNary mappingPropertyNary) {
+
+ }
+
+ @Override
+ public void visit(MappingPropertyUnary mappingPropertyUnary) {
+ mappingPropertyUnary.setSelectedProperty(mappingPropertyValue);
+ }
+
+ @Override
+ public void visit(MatchingStaticProperty matchingStaticProperty) {
+
+ }
+
+ @Override
+ public void visit(OneOfStaticProperty oneOfStaticProperty) {
+ var label = this.trendOperator.getLabel();
+ oneOfStaticProperty.getOptions().forEach(option -> {
+ if (option.getName().equals(label)) {
+ option.setSelected(true);
+ }
+ });
+ }
+
+ @Override
+ public void visit(SecretStaticProperty secretStaticProperty) {
+
+ }
+
+ @Override
+ public void visit(StaticPropertyAlternative staticPropertyAlternative) {
+
+ }
+
+ @Override
+ public void visit(StaticPropertyAlternatives staticPropertyAlternatives) {
+
+ }
+
+ @Override
+ public void visit(StaticPropertyGroup staticPropertyGroup) {
+
+ }
+
+ @Override
+ public void visit(RemoteOneOfStaticProperty remoteOneOfStaticProperty) {
+
+ }
+
+ @Override
+ public void visit(SlideToggleStaticProperty slideToggleStaticProperty) {
+
+ }
+
+ @Override
+ public void visit(RuntimeResolvableTreeInputStaticProperty
treeInputStaticProperty) {
+
+ }
+}
