You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ri...@apache.org on 2020/08/25 20:00:26 UTC
[incubator-streampipes-examples] branch dev updated: Add examples
for ColorPicker and Collections
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 e2c8978 Add examples for ColorPicker and Collections
new c9b35d4 Merge branch 'dev' of https://github.com/apache/incubator-streampipes-examples into dev
e2c8978 is described below
commit e2c8978b02949ce466b41e4c98f4d19eb45e5a99
Author: Dominik Riemer <ri...@fzi.de>
AuthorDate: Tue Aug 25 21:59:59 2020 +0200
Add examples for ColorPicker and Collections
---
.../streampipes/pe/examples/jvm/ExamplesInit.java | 3 +
.../staticproperty/CollectionMappingExample.java | 65 +++++++++++++++
.../ColorPickerExampleController.java | 63 ++++++++++++++
...ntimeResolvableAnyStaticPropertyController.java | 97 ++++++++++++++++++++++
4 files changed, 228 insertions(+)
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 6a6550c..70fbba2 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
@@ -46,9 +46,12 @@ public class ExamplesInit extends StandaloneModelSubmitter {
.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 AppendOutputController())
.add(new CustomOutputController())
diff --git a/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/CollectionMappingExample.java b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/CollectionMappingExample.java
new file mode 100644
index 0000000..8efbd6c
--- /dev/null
+++ b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/CollectionMappingExample.java
@@ -0,0 +1,65 @@
+/*
+ * 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.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.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.*;
+import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
+import org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
+
+import java.util.List;
+
+public class CollectionMappingExample extends
+ StandaloneEventProcessingDeclarer<DummyParameters> {
+
+ private static final String MAPPING_PROPERTY_ID = "mapping-property";
+ private static final String FIELDS_KEY = "fields";
+
+ @Override
+ public DataProcessorDescription declareModel() {
+ return ProcessingElementBuilder.create("org.apache.streampipes.examples.collection.mapping",
+ "Collection with mapping properties", "")
+ .requiredStream(StreamRequirementsBuilder.
+ create()
+ .requiredProperty(EpRequirements.withMappingPropertyId(MAPPING_PROPERTY_ID, EpRequirements.numberReq()))
+ .build())
+ .requiredCollection(Labels.from(FIELDS_KEY, "Field Mappings", ""),
+ StaticProperties.mappingPropertyUnary(Labels.from(MAPPING_PROPERTY_ID, "Field", ""),
+ RequirementsSelector.FIRST_INPUT_STREAM,
+ PropertyScope.NONE))
+ .outputStrategy(OutputStrategies.keep())
+ .build();
+ }
+
+ @Override
+ public ConfiguredEventProcessor<DummyParameters> onInvocation(DataProcessorInvocation graph, ProcessingElementParameterExtractor extractor) {
+
+ // Extract the mapping property value
+ List<String> selectedMappings = extractor.getUnaryMappingsFromCollection(FIELDS_KEY);
+
+ return new ConfiguredEventProcessor<>(new DummyParameters(graph), DummyEngine::new);
+ }
+}
diff --git a/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/ColorPickerExampleController.java b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/ColorPickerExampleController.java
new file mode 100644
index 0000000..68c150a
--- /dev/null
+++ b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/ColorPickerExampleController.java
@@ -0,0 +1,63 @@
+/*
+ * 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.staticproperty;
+
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.pe.examples.jvm.base.DummyEngine;
+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.*;
+import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
+import org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
+
+public class ColorPickerExampleController extends StandaloneEventProcessingDeclarer<DummyParameters> {
+
+ private static final String COLOR_PICKER_KEY = "color-key";
+
+ @Override
+ public DataProcessorDescription declareModel() {
+ return ProcessingElementBuilder.create("org.apache.streampipes.examples.staticproperty" +
+ ".colorpicker", "Color Picker Example", "")
+ .requiredStream(StreamRequirementsBuilder.
+ create()
+ .requiredProperty(EpRequirements.anyProperty())
+ .build())
+ .outputStrategy(OutputStrategies.userDefined())
+ .supportedProtocols(SupportedProtocols.kafka())
+ .supportedFormats(SupportedFormats.jsonFormat())
+
+ // create a required code block
+ .requiredColorParameter(Labels.from(COLOR_PICKER_KEY, "Color", "Select color"))
+
+ .build();
+ }
+
+ @Override
+ public ConfiguredEventProcessor<DummyParameters> onInvocation(DataProcessorInvocation graph, ProcessingElementParameterExtractor extractor) {
+
+ // Extract the code parameter value
+ String color = extractor.selectedColor(COLOR_PICKER_KEY);
+
+ // now the text parameter would be added to a parameter class (omitted for this example)
+
+ return new ConfiguredEventProcessor<>(new DummyParameters(graph), DummyEngine::new);
+ }
+}
diff --git a/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/RuntimeResolvableAnyStaticPropertyController.java b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/RuntimeResolvableAnyStaticPropertyController.java
new file mode 100644
index 0000000..48e6b36
--- /dev/null
+++ b/streampipes-pipeline-elements-examples-processors-jvm/src/main/java/org/apache/streampipes/pe/examples/jvm/staticproperty/RuntimeResolvableAnyStaticPropertyController.java
@@ -0,0 +1,97 @@
+/*
+ * 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.staticproperty;
+
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.streampipes.container.api.ResolvesContainerProvidedOptions;
+import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.model.graph.DataProcessorInvocation;
+import org.apache.streampipes.model.staticproperty.Option;
+import org.apache.streampipes.pe.examples.jvm.base.DummyEngine;
+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.extractor.StaticPropertyExtractor;
+import org.apache.streampipes.sdk.helpers.*;
+import org.apache.streampipes.wrapper.standalone.ConfiguredEventProcessor;
+import org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventProcessingDeclarer;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Properties;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class RuntimeResolvableAnyStaticPropertyController extends
+ StandaloneEventProcessingDeclarer<DummyParameters> implements ResolvesContainerProvidedOptions {
+
+ private static final String KafkaHost = "kafka-host";
+ private static final String KafkaPort = "kafka-port";
+
+ @Override
+ public DataProcessorDescription declareModel() {
+ return ProcessingElementBuilder.create("org.apache.streampipes.examples.staticproperty" +
+ ".runtimeresolvablemulti", "Runtime-resolvable multi value example", "")
+ .requiredStream(StreamRequirementsBuilder.
+ create()
+ .requiredProperty(EpRequirements.anyProperty())
+ .build())
+ .outputStrategy(OutputStrategies.keep())
+ .supportedProtocols(SupportedProtocols.kafka())
+ .supportedFormats(SupportedFormats.jsonFormat())
+ .requiredTextParameter(Labels.from(KafkaHost, "Kafka Host", ""))
+ .requiredIntegerParameter(Labels.from(KafkaPort, "Kafka Port", ""))
+
+ // create a single value selection parameter that is resolved at runtime
+ .requiredMultiValueSelectionFromContainer(Labels.from("id", "Example Name", "Example " +
+ "Description"), Arrays.asList(KafkaHost, KafkaPort))
+
+ .build();
+ }
+
+ @Override
+ public ConfiguredEventProcessor<DummyParameters> onInvocation(DataProcessorInvocation graph, ProcessingElementParameterExtractor extractor) {
+
+ // Extract the text parameter value
+ List<String> selectedSingleValue = extractor.selectedMultiValues("id", String.class);
+
+ // now the text parameter would be added to a parameter class (omitted for this example)
+
+ return new ConfiguredEventProcessor<>(new DummyParameters(graph), DummyEngine::new);
+ }
+
+ @Override
+ public List<Option> resolveOptions(String requestId, StaticPropertyExtractor extractor) {
+ String host = extractor.singleValueParameter(KafkaHost, String.class);
+ Integer port = extractor.singleValueParameter(KafkaPort, Integer.class);
+
+ String kafkaAddress = host + ":" + port;
+
+ Properties props = new Properties();
+ props.put("bootstrap.servers", kafkaAddress);
+ props.put("group.id", "test-consumer-group");
+ props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+ props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+
+ KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
+ Set<String> topics = consumer.listTopics().keySet();
+ consumer.close();
+ return topics.stream().map(Option::new).collect(Collectors.toList());
+ }
+}