[STREAMPIPES-423] Add python code input example
diff --git a/development/.env b/development/.env
index 285230a..a99fe35 100644
--- a/development/.env
+++ b/development/.env
@@ -15,5 +15,4 @@
# Those parameters are used by IntelliJ to set the default consul parameters for development
SP_PORT=6009
-SP_HOST=host.docker.internal
SP_DEBUG=true
diff --git a/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/client/example/Example.java b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/client/example/Example.java
new file mode 100644
index 0000000..dccd087
--- /dev/null
+++ b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/client/example/Example.java
@@ -0,0 +1,51 @@
+/*
+ * 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.client.example;
+
+import org.apache.commons.collections.MapUtils;
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.client.StreamPipesCredentials;
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+
+import java.util.List;
+
+public class Example {
+
+ public static void main(String[] args) {
+ StreamPipesCredentials credentials = StreamPipesCredentials
+ .from(System.getenv("user"), System.getenv("apiKey"));
+
+ // Create an instance of the StreamPipes client
+ StreamPipesClient client = StreamPipesClient
+ .create("localhost", 80, credentials, true);
+
+ // Get all pipelines
+ List<Pipeline> pipelines = client.pipelines().all();
+
+ // Start a pipeline
+ PipelineOperationStatus message = client.pipelines().start(pipelines.get(0));
+
+ // Get all data streams
+ List<SpDataStream> dataStreams = client.streams().all();
+
+ // Subscribe to a data stream
+ client.streams().subscribe(dataStreams.get(0), event -> MapUtils.debugPrint(System.out, "event", event.getRaw()));
+ }
+}
diff --git a/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/ExamplesInit.java b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/ExamplesInit.java
index e5ac627..634de79 100644
--- a/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/ExamplesInit.java
+++ b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/ExamplesInit.java
@@ -17,59 +17,60 @@
*/
package org.apache.streampipes.pe.examples.jvm;
-import org.apache.streampipes.container.init.DeclarersSingleton;
+import org.apache.streampipes.container.model.SpServiceDefinition;
+import org.apache.streampipes.container.model.SpServiceDefinitionBuilder;
import org.apache.streampipes.container.standalone.init.StandaloneModelSubmitter;
import org.apache.streampipes.dataformat.json.JsonDataFormatFactory;
import org.apache.streampipes.messaging.jms.SpJmsProtocolFactory;
import org.apache.streampipes.messaging.kafka.SpKafkaProtocolFactory;
-import org.apache.streampipes.pe.examples.jvm.config.ExamplesJvmConfig;
import org.apache.streampipes.pe.examples.jvm.engine.ExampleExternalEngineController;
-import org.apache.streampipes.pe.examples.jvm.outputstrategy.AppendOutputController;
-import org.apache.streampipes.pe.examples.jvm.outputstrategy.CustomOutputController;
-import org.apache.streampipes.pe.examples.jvm.outputstrategy.CustomTransformOutputController;
-import org.apache.streampipes.pe.examples.jvm.outputstrategy.FixedOutputController;
-import org.apache.streampipes.pe.examples.jvm.outputstrategy.KeepOutputController;
-import org.apache.streampipes.pe.examples.jvm.outputstrategy.TransformOutputController;
+import org.apache.streampipes.pe.examples.jvm.outputstrategy.*;
import org.apache.streampipes.pe.examples.jvm.requirements.NestedListRequirementsController;
import org.apache.streampipes.pe.examples.jvm.staticproperty.*;
public class ExamplesInit extends StandaloneModelSubmitter {
public static void main(String[] args) {
- DeclarersSingleton
- .getInstance()
- .add(new TextParameterExampleController())
- .add(new NumberParameterExampleController())
- .add(new NumberParameterWithRangeExampleController())
- .add(new UnaryMappingPropertyExampleController())
- .add(new NaryMappingPropertyExampleController())
- .add(new SingleValueSelectionExampleController())
- .add(new MultiValueSelectionExampleController())
- .add(new CollectionExampleController())
- .add(new RuntimeResolvableSingleValue())
- .add(new RuntimeResolvableAnyStaticPropertyController())
- .add(new StaticPropertyAlternativesController())
- .add(new SecretStaticPropertyExampleController())
- .add(new CodeInputExampleController())
- .add(new ColorPickerExampleController())
- .add(new CollectionMappingExample())
- .add(new NestedListRequirementsController())
- .add(new TwoStreamsMappingExample())
+ new ExamplesInit().init();
+ }
- .add(new AppendOutputController())
- .add(new CustomOutputController())
- .add(new FixedOutputController())
- .add(new CustomTransformOutputController())
- .add(new TransformOutputController())
- .add(new KeepOutputController())
- .add(new CollectionMappingGroupExample())
+ @Override
+ public SpServiceDefinition provideServiceDefinition() {
+ return SpServiceDefinitionBuilder.create("org.apache.streampipes.processors.examples.jvm",
+ "StreamPipes Code Examples",
+ "",
+ 8090)
+ .registerMessagingProtocols(new SpKafkaProtocolFactory(), new SpJmsProtocolFactory())
+ .registerMessagingFormats(new JsonDataFormatFactory())
+ .registerPipelineElement(new TextParameterExampleController())
+ .registerPipelineElement(new NumberParameterExampleController())
+ .registerPipelineElement(new NumberParameterWithRangeExampleController())
+ .registerPipelineElement(new UnaryMappingPropertyExampleController())
+ .registerPipelineElement(new NaryMappingPropertyExampleController())
+ .registerPipelineElement(new SingleValueSelectionExampleController())
+ .registerPipelineElement(new MultiValueSelectionExampleController())
+ .registerPipelineElement(new CollectionExampleController())
+ .registerPipelineElement(new RuntimeResolvableSingleValue())
+ .registerPipelineElement(new RuntimeResolvableAnyStaticPropertyController())
+ .registerPipelineElement(new StaticPropertyAlternativesController())
+ .registerPipelineElement(new SecretStaticPropertyExampleController())
+ .registerPipelineElement(new CodeInputExampleController())
+ .registerPipelineElement(new ColorPickerExampleController())
+ .registerPipelineElement(new CollectionMappingExample())
+ .registerPipelineElement(new NestedListRequirementsController())
+ .registerPipelineElement(new TwoStreamsMappingExample())
- .add(new ExampleExternalEngineController());
+ .registerPipelineElement(new AppendOutputController())
+ .registerPipelineElement(new CustomOutputController())
+ .registerPipelineElement(new FixedOutputController())
+ .registerPipelineElement(new CustomTransformOutputController())
+ .registerPipelineElement(new TransformOutputController())
+ .registerPipelineElement(new KeepOutputController())
+ .registerPipelineElement(new CollectionMappingGroupExample())
- DeclarersSingleton.getInstance().registerDataFormat(new JsonDataFormatFactory());
- DeclarersSingleton.getInstance().registerProtocol(new SpKafkaProtocolFactory());
- DeclarersSingleton.getInstance().registerProtocol(new SpJmsProtocolFactory());
+ .registerPipelineElement(new ExampleExternalEngineController())
+ .build();
- new ExamplesInit().init(ExamplesJvmConfig.INSTANCE);
+
}
}
diff --git a/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/config/ExamplesJvmConfig.java b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/config/ExamplesJvmConfig.java
index 345c777..d8c8a57 100644
--- a/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/config/ExamplesJvmConfig.java
+++ b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/config/ExamplesJvmConfig.java
@@ -19,8 +19,9 @@
package org.apache.streampipes.pe.examples.jvm.config;
-import org.apache.streampipes.config.SpConfig;
import org.apache.streampipes.container.model.PeConfig;
+import org.apache.streampipes.svcdiscovery.SpServiceDiscovery;
+import org.apache.streampipes.svcdiscovery.api.SpConfig;
public enum ExamplesJvmConfig implements PeConfig {
INSTANCE;
@@ -35,7 +36,7 @@
private final static String service_container_name = "processors-examples-jvm";
ExamplesJvmConfig() {
- config = SpConfig.getSpConfig(service_id);
+ config = SpServiceDiscovery.getSpConfig(service_id);
config.register(ConfigKeys.HOST, service_container_name, "Hostname for the pe esper");
config.register(ConfigKeys.PORT, 8090, "Port for the pe esper");
diff --git a/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/CodeInputExampleController.java b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/CodeInputExampleController.java
index 85a6d6d..a1285ec 100644
--- a/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/CodeInputExampleController.java
+++ b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/CodeInputExampleController.java
@@ -14,6 +14,8 @@
public class CodeInputExampleController extends StandaloneEventProcessingDeclarer<DummyParameters> {
private static final String CODE_KEY = "code-key";
+ private static final String CODE_PYTHON_KEY = "code-python-key";
+ private static final String CODE_EMPTY_KEY = "code-empty-key";
@Override
public DataProcessorDescription declareModel() {
@@ -28,7 +30,13 @@
.supportedFormats(SupportedFormats.jsonFormat())
// create a required code block
- .requiredCodeblock(Labels.from(CODE_KEY, "Code", ""), CodeLanguage.Javascript)
+ .requiredCodeblock(Labels.from(CODE_KEY, "Code JS", ""), CodeLanguage.Javascript)
+
+ // python example
+ .requiredCodeblock(Labels.from(CODE_PYTHON_KEY, "Code Python", ""), CodeLanguage.Python)
+
+ // no specific language
+ .requiredCodeblock(Labels.from(CODE_EMPTY_KEY, "Any Code", ""), CodeLanguage.None)
.build();
}
