Repository: apache/streampipes
Pull request #1625 (2023-05-27 11:42:04)

Migrate processors in siddhi module (#1624)
<!--
  ~ 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.
  ~
  -->
  
  <!--
Thanks for contributing! Here are some tips you can follow to help us incorporate your contribution quickly and easily:
1. If this is your first time, please read our contributor guidelines:
    - https://streampipes.apache.org/getinvolved.html
    - https://cwiki.apache.org/confluence/display/STREAMPIPES/Getting+Started
2. Make sure the PR title is formatted like: `[#<GitHub issue id>] PR title ...`
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., `[WIP][#<GitHub issue id>] PR title ...`.
4. Please write your PR title to summarize what this PR proposes/fixes.
5. Link the PR to the corresponding GitHub issue (if present) in the `Development` section in the right menu bar. 
6. Be sure to keep the PR description updated to reflect all changes.
7. If possible, provide a concise example to reproduce the issue for a faster review.
8. Make sure tests pass via `mvn clean install`.
9. (Optional) If the contribution is large, please file an Apache ICLA
    - http://apache.org/licenses/icla.pdf
-->

### Purpose
<!--
Please clarify what changes you are proposing and describe how those changes will address the issue.
Furthermore, describe potential consequences the changes might have.
-->

Migrate all data processors in streampipes-processors-siddhi.

### Remarks
<!--
Is there anything left we need to pay attention on?
Are there some references that might be important? E.g. links to Confluence, or discussions
on the mailing list or GitHub.
-->
PR introduces (a) breaking change(s): no

PR introduces (a) deprecation(s): no



Patch:
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 c181382b2e..06e7272151 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 048b003859..0000000000
--- 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 a928171199..0000000000
--- 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 6bc5488c3d..8d44056736 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 15e10f4f87..0000000000
--- 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 53b65480a9..0000000000
--- 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 4895ccacd4..3438aebcd0 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 86ce203486..0000000000
--- 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 b9c129ef1c..0000000000
--- 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 7a9a386dd5..b8b7718b59 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 f5f02785b1..0000000000
--- 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 9d085c9763..0000000000
--- 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 15cb0be43f..2c1cd55785 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 93363dae30..0000000000
--- 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 311ae2eaa2..0000000000
--- 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 b8f7670750..6b16c2dd12 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 ccc9f5ba6d..0000000000
--- 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 6dad7c618d..c5a915b168 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 ae3df9dd02..0000000000
--- 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 30a7073165..d628119137 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/ui/package-lock.json b/ui/package-lock.json
index 3f14ca39d5..c13ec4ee2e 100644
--- a/ui/package-lock.json
+++ b/ui/package-lock.json
@@ -24212,9 +24212,9 @@
       }
     },
     "node_modules/socket.io-parser": {
-      "version": "4.2.1",
-      "resolved": "https://registry.npmjs.org/socket.io-parser/-/socket.io-parser-4.2.1.tgz",
-      "integrity": "sha512-V4GrkLy+HeF1F/en3SpUaM+7XxYXpuMUWLGde1kSSh5nQMN4hLrbPIkD+otwh6q9R6NOQBN4AMaOZ2zVjui82g==",
+      "version": "4.2.3",
+      "resolved": "https://registry.npmjs.org/socket.io-parser/-/socket.io-parser-4.2.3.tgz",
+      "integrity": "sha512-JMafRntWVO2DCJimKsRTh/wnqVvO4hrfwOqtO7f+uzwsQMuxO6VwImtYxaQ+ieoyshWOTJyV0fA21lccEXRPpQ==",
       "dev": true,
       "dependencies": {
         "@socket.io/component-emitter": "~3.1.0",
@@ -46307,9 +46307,9 @@
       }
     },
     "socket.io-parser": {
-      "version": "4.2.1",
-      "resolved": "https://registry.npmjs.org/socket.io-parser/-/socket.io-parser-4.2.1.tgz",
-      "integrity": "sha512-V4GrkLy+HeF1F/en3SpUaM+7XxYXpuMUWLGde1kSSh5nQMN4hLrbPIkD+otwh6q9R6NOQBN4AMaOZ2zVjui82g==",
+      "version": "4.2.3",
+      "resolved": "https://registry.npmjs.org/socket.io-parser/-/socket.io-parser-4.2.3.tgz",
+      "integrity": "sha512-JMafRntWVO2DCJimKsRTh/wnqVvO4hrfwOqtO7f+uzwsQMuxO6VwImtYxaQ+ieoyshWOTJyV0fA21lccEXRPpQ==",
       "dev": true,
       "requires": {
         "@socket.io/component-emitter": "~3.1.0",


Test patch:
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 de9c2a1d27..1f028f0e96 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 0000000000..673f3185fb
--- /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) {
+
+  }
+}