This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch dev
in repository
https://gitbox.apache.org/repos/asf/incubator-streampipes-examples.git
The following commit(s) were added to refs/heads/dev by this push:
new e2ff61a Add StreamPipes client example
e2ff61a is described below
commit e2ff61a4725f448dfc44e730f12999619ee9dbb1
Author: Dominik Riemer <[email protected]>
AuthorDate: Thu Feb 11 22:14:32 2021 +0100
Add StreamPipes client example
---
pom.xml | 13 ++--
.../pom.xml | 8 ++-
.../client/example/StreamPipesClientExample.java | 69 ++++++++++++++++++++++
.../streampipes/pe/examples/jvm/ExamplesInit.java | 2 +
.../NestedListRequirementsController.java | 56 ++++++++++++++++++
.../CollectionMappingGroupExample.java | 22 ++++---
6 files changed, 157 insertions(+), 13 deletions(-)
diff --git a/pom.xml b/pom.xml
index c66b0e5..95104b2 100644
--- a/pom.xml
+++ b/pom.xml
@@ -25,18 +25,18 @@
<parent>
<groupId>org.apache.streampipes</groupId>
<artifactId>streampipes-parent</artifactId>
- <version>0.67.0-SNAPSHOT</version>
+ <version>0.68.0-SNAPSHOT</version>
</parent>
<artifactId>streampipes-pipeline-elements-examples</artifactId>
<packaging>pom</packaging>
- <version>0.67.0-SNAPSHOT</version>
+ <version>0.68.0-SNAPSHOT</version>
<modules>
<module>streampipes-pipeline-elements-examples-processors-jvm</module>
</modules>
<properties>
- <streampipes.version>0.67.0-SNAPSHOT</streampipes.version>
+ <streampipes.version>0.68.0-SNAPSHOT</streampipes.version>
<lightcouch.version>0.1.8</lightcouch.version>
<maven-shade-plugin.version>3.0.0</maven-shade-plugin.version>
@@ -46,6 +46,11 @@
<dependencies>
<dependency>
<groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-client</artifactId>
+ <version>${streampipes.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
<artifactId>streampipes-commons</artifactId>
<version>${streampipes.version}</version>
</dependency>
@@ -251,4 +256,4 @@
</scm>
-</project>
\ No newline at end of file
+</project>
diff --git a/streampipes-pipeline-elements-examples-processors-jvm/pom.xml
b/streampipes-pipeline-elements-examples-processors-jvm/pom.xml
index 4593856..ca7c5ff 100644
--- a/streampipes-pipeline-elements-examples-processors-jvm/pom.xml
+++ b/streampipes-pipeline-elements-examples-processors-jvm/pom.xml
@@ -23,7 +23,7 @@
<parent>
<artifactId>streampipes-pipeline-elements-examples</artifactId>
<groupId>org.apache.streampipes</groupId>
- <version>0.67.0-SNAPSHOT</version>
+ <version>0.68.0-SNAPSHOT</version>
</parent>
<modelVersion>4.0.0</modelVersion>
@@ -58,6 +58,10 @@
<groupId>org.apache.streampipes</groupId>
<artifactId>streampipes-messaging-jms</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-client</artifactId>
+ </dependency>
</dependencies>
<build>
@@ -76,4 +80,4 @@
</plugins>
<finalName>streampipes-processors-examples-jvm</finalName>
</build>
-</project>
\ No newline at end of file
+</project>
diff --git
a/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/client/example/StreamPipesClientExample.java
b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/client/example/StreamPipesClientExample.java
new file mode 100644
index 0000000..6d1e15d
--- /dev/null
+++
b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/client/example/StreamPipesClientExample.java
@@ -0,0 +1,69 @@
+/*
+ * 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.client.live.KafkaConfig;
+import org.apache.streampipes.model.SpDataStream;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.pipeline.Pipeline;
+import org.apache.streampipes.model.pipeline.PipelineOperationStatus;
+import org.apache.streampipes.model.template.PipelineElementTemplate;
+
+import java.util.List;
+
+public class StreamPipesClientExample {
+
+ public static void main(String[] args) {
+
+ // First, go to the StreamPipes UI and create an API key (user -> profile
in the upper right corner)
+
+ // Create credentials by providing a user (the email) and the API key
+ 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 pipeliens
+ List<Pipeline> pipelines = client.pipelines().all();
+
+ // Start a pipeline
+ PipelineOperationStatus message =
client.pipelines().start(pipelines.get(0));
+
+ // Get all pipeline element templates
+ List<PipelineElementTemplate> templates =
client.pipelineElementTemplates().all();
+
+ // Get all data sinks
+ List<DataSinkInvocation> dataSinks = client.sinks().all();
+
+ // 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()));
+
+ // Subscribe to a data stream and provide an additional Kafka config
(e.g., for access from outside the StreamPipes network)
+ client.streams().subscribe(dataStreams.get(0),
KafkaConfig.create("localhost", 9094), 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 7069afe..1bdabb3 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
@@ -30,6 +30,7 @@ import
org.apache.streampipes.pe.examples.jvm.outputstrategy.CustomTransformOutp
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.requirements.NestedListRequirementsController;
import org.apache.streampipes.pe.examples.jvm.staticproperty.*;
public class ExamplesInit extends StandaloneModelSubmitter {
@@ -52,6 +53,7 @@ public class ExamplesInit extends StandaloneModelSubmitter {
.add(new CodeInputExampleController())
.add(new ColorPickerExampleController())
.add(new CollectionMappingExample())
+ .add(new NestedListRequirementsController())
.add(new AppendOutputController())
.add(new CustomOutputController())
diff --git
a/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/requirements/NestedListRequirementsController.java
b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/requirements/NestedListRequirementsController.java
new file mode 100644
index 0000000..6a32659
--- /dev/null
+++
b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/requirements/NestedListRequirementsController.java
@@ -0,0 +1,56 @@
+/*
+ * 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.pe.examples.jvm.requirements;
+
+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.pe.examples.jvm.base.DummyParameters;
+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.OutputStrategies;
+import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
+import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
+
+public class NestedListRequirementsController extends
StandaloneEventProcessingDeclarer<DummyParameters> {
+
+ @Override
+ public DataProcessorDescription declareModel() {
+ return
ProcessingElementBuilder.create("org.apache.streampipes.examples.requirements.nestedlist",
+ "Nested list mapping example", "")
+ .requiredStream(StreamRequirementsBuilder.
+ create()
+
.requiredPropertyWithUnaryMapping(EpRequirements.nestedListRequirement(EpRequirements.stringReq()),
+ Labels.from("key-value", "Key", ""),
+ PropertyScope.NONE)
+
.requiredPropertyWithUnaryMapping(EpRequirements.nestedListRequirement(EpRequirements.integerReq()),
+ Labels.from("count-value", "Count", ""),
+ PropertyScope.NONE)
+ .build())
+ .outputStrategy(OutputStrategies.keep())
+ .build();
+ }
+
+ @Override
+ public ConfiguredEventProcessor<DummyParameters>
onInvocation(DataProcessorInvocation graph, ProcessingElementParameterExtractor
extractor) {
+ return null;
+ }
+}
diff --git
a/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/CollectionMappingGroupExample.java
b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/CollectionMappingGroupExample.java
index fff7229..670b035 100644
---
a/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/CollectionMappingGroupExample.java
+++
b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/CollectionMappingGroupExample.java
@@ -20,19 +20,14 @@ package
org.apache.streampipes.pe.examples.jvm.staticproperty;
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.model.staticproperty.FreeTextStaticProperty;
-import org.apache.streampipes.model.staticproperty.MappingPropertyUnary;
-import org.apache.streampipes.model.staticproperty.StaticPropertyGroup;
+import org.apache.streampipes.model.staticproperty.*;
import org.apache.streampipes.pe.examples.jvm.base.DummyEngine;
import org.apache.streampipes.pe.examples.jvm.base.DummyParameters;
import org.apache.streampipes.sdk.StaticProperties;
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.OutputStrategies;
-import org.apache.streampipes.sdk.helpers.RequirementsSelector;
+import org.apache.streampipes.sdk.helpers.*;
import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
@@ -44,6 +39,7 @@ public class CollectionMappingGroupExample extends
private static final String MAPPING_PROPERTY_ID = "mapping-property";
private static final String FIELDS_KEY = "fields";
+ private static final String COMPARATOR_ID = "comparator";
@Override
public DataProcessorDescription declareModel() {
@@ -55,6 +51,8 @@ public class CollectionMappingGroupExample extends
.build())
.requiredCollection(Labels.from(FIELDS_KEY, "Field Mappings", ""),
StaticProperties.group(Labels.from("group", "Group", ""),
false,
+
StaticProperties.singleValueSelection(Labels.from(COMPARATOR_ID, "Comparator",
""),
+ Options.from("<", "<=", ">", ">=", "==",
"*")),
StaticProperties.mappingPropertyUnary(Labels.from(MAPPING_PROPERTY_ID, "Field",
""),
RequirementsSelector.FIRST_INPUT_STREAM,
PropertyScope.NONE),
@@ -85,6 +83,16 @@ public class CollectionMappingGroupExample extends
.map(Double::parseDouble)
.collect(Collectors.toList());
+ List<String> comparators = groupItems
+ .stream()
+ .map(group -> (extractor
+ .extractGroupMember(COMPARATOR_ID, group)
+ .as(OneOfStaticProperty.class))
+ .getOptions()
+ .stream()
+ .filter(Option::isSelected).findFirst().get().getName())
+ .collect(Collectors.toList());
+
return new ConfiguredEventProcessor<>(new DummyParameters(graph),
DummyEngine::new);
}
}
