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/streampipes-website.git
The following commit(s) were added to refs/heads/dev by this push:
new c7c89b519 Update tutorials (#137)
c7c89b519 is described below
commit c7c89b519e5aebd31d520084b9488937078546d4
Author: Dominik Riemer <[email protected]>
AuthorDate: Sun Nov 26 21:40:42 2023 +0100
Update tutorials (#137)
---
docs/06_extend-tutorial-adapters.md | 2 +-
docs/06_extend-tutorial-data-processors.md | 287 +++++++++++++++++------------
docs/06_extend-tutorial-data-sinks.md | 103 ++++++-----
3 files changed, 229 insertions(+), 163 deletions(-)
diff --git a/docs/06_extend-tutorial-adapters.md
b/docs/06_extend-tutorial-adapters.md
index da8d4dc16..24cbf161d 100644
--- a/docs/06_extend-tutorial-adapters.md
+++ b/docs/06_extend-tutorial-adapters.md
@@ -52,7 +52,7 @@ that [Apache Maven](https://maven.apache.org/install.html)
isinstalled):
```bash
mvn archetype:generate \
-DarchetypeGroupId=org.apache.streampipes
-DarchetypeArtifactId=streampipes-archetype-extensions-jvm \
--DarchetypeVersion=0.93.0-SNAPSHOT -DgroupId=org.apache.streampipes \
+-DarchetypeVersion=0.93.0 -DgroupId=org.apache.streampipes \
-DartifactId=streampipes-archetype-extensions-jvm
-DclassNamePrefix=ParcelControlStation -DpackageName=parcelcontrol
```
diff --git a/docs/06_extend-tutorial-data-processors.md
b/docs/06_extend-tutorial-data-processors.md
index a4170ccb6..4c2d5ee25 100644
--- a/docs/06_extend-tutorial-data-processors.md
+++ b/docs/06_extend-tutorial-data-processors.md
@@ -38,7 +38,7 @@ Enter the following command in a command line of your choice
(Apache Maven needs
```
mvn archetype:generate \
-DarchetypeGroupId=org.apache.streampipes
-DarchetypeArtifactId=streampipes-archetype-extensions-jvm \
--DarchetypeVersion=0.70.0 -DgroupId=my.groupId \
+-DarchetypeVersion=0.93.0 -DgroupId=my.groupId \
-DartifactId=my-example -DclassNamePrefix=MyExample -DpackageName=mypackagename
```
@@ -61,47 +61,55 @@ Create a new class `GeofencingProcessor` which should look
as follows:
```java
package org.apache.streampipes.pe.example;
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.model.DataProcessorType;
-import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.extensions.api.pe.IStreamPipesDataProcessor;
+import
org.apache.streampipes.extensions.api.pe.config.IDataProcessorConfiguration;
+import
org.apache.streampipes.extensions.api.pe.context.EventProcessorRuntimeContext;
+import org.apache.streampipes.extensions.api.pe.param.IDataProcessorParameters;
+import org.apache.streampipes.extensions.api.pe.routing.SpOutputCollector;
import org.apache.streampipes.model.runtime.Event;
import org.apache.streampipes.sdk.builder.ProcessingElementBuilder;
import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+import org.apache.streampipes.sdk.builder.processor.DataProcessorConfiguration;
+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.context.EventProcessorRuntimeContext;
-import org.apache.streampipes.wrapper.routing.SpOutputCollector;
-import org.apache.streampipes.wrapper.standalone.ProcessorParams;
-import org.apache.streampipes.wrapper.standalone.StreamPipesDataProcessor;
+import org.apache.streampipes.sdk.helpers.SupportedFormats;
+import org.apache.streampipes.sdk.helpers.SupportedProtocols;
+import org.apache.streampipes.vocabulary.SO;
-public class GeofencingProcessor extends StreamPipesDataProcessor {
+public class GeofencingProcessor implements IStreamPipesDataProcessor {
private static final String LATITUDE_CENTER = "latitude-center";
private static final String LONGITUDE_CENTER = "longitude-center";
- @Override
- public DataProcessorDescription declareModel() {
- return
ProcessingElementBuilder.create("org.apache.streampipes.tutorial-geofencing")
- .category(DataProcessorType.ENRICH)
- .withAssets(Assets.DOCUMENTATION, Assets.ICON)
- .build();
+
+ public IDataProcessorConfiguration declareConfig() {
+ return DataProcessorConfiguration.create(
+ GeofencingProcessor::new,
+ ProcessingElementBuilder.create(
+ "org.apache.streampipes.tutorial-geofencing"
+ )
+ .category(DataProcessorType.ENRICH)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .build());
}
@Override
- public void onInvocation(ProcessorParams parameters, SpOutputCollector
spOutputCollector, EventProcessorRuntimeContext runtimeContext) throws
SpRuntimeException {
+ public void onPipelineStarted(IDataProcessorParameters params,
+ SpOutputCollector collector,
+ EventProcessorRuntimeContext runtimeContext) {
}
@Override
- public void onEvent(Event event, SpOutputCollector collector) throws
SpRuntimeException {
+ public void onEvent(Event event,
+ SpOutputCollector collector) {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() {
}
}
@@ -109,31 +117,33 @@ public class GeofencingProcessor extends
StreamPipesDataProcessor {
```
-In this class, we need to implement three methods: The `declareModel` method
is used to define abstract stream
+In this class, we need to implement three methods: The `declareConfig` method
is used to define abstract stream
requirements such as event properties that must be present in any input stream
that is later connected to the element
using the StreamPipes UI.
-The second method, `onInvocation` is triggered once a pipeline is started.
Finally, the `onEvent` method
+The second method, `onPipelineStarted` is triggered once a pipeline is
started.
+The `onEvent` method is called for every incoming event.
+Finally, the `onPipelineStopped` method is called once the pipeline is stopped.
Similar to data sources, the SDK provides a builder class to generate the
description for data processors.
-Delete the content within the ``declareModel`` method and add the following
lines to the `declareModel` method:
-
-```java
-return
ProcessingElementBuilder.create("org.apache.streampipes.tutorial.geofencing","Geofencing","A
simple geofencing data processor")
-```
-This creates a new data processor with the ID, title and description assigned
to the element builder.
-Next, we add some _stream requirements_ to the description. As we'd like to
develop a generic pipeline element that
+The current code within the `declareConfig` method creates a new data
processor with the ID.
+The ID is used as the internal ID of the data processor, but also used to
reference additional assets in the `resources` folder, such as a `strings.en`
file, used to configure labels and description, and a `documentation.md` file,
which will later servce as a markdown documentation in the UI.
+But first, we will add some _stream requirements_ to the description. As we'd
like to develop a generic pipeline element that
works with any event that provides a lat/lng pair, we define two stream
requirements as stated below:
```java
.requiredStream(StreamRequirementsBuilder
.create()
-
.requiredPropertyWithUnaryMapping(EpRequirements.domainPropertyReq(Geo.lat),
- Labels.from("latitude-field","Latitude","The event "+
- "property containing the latitude
value"),PropertyScope.MEASUREMENT_PROPERTY)
-
.requiredPropertyWithUnaryMapping(EpRequirements.domainPropertyReq(Geo.lng),
- Labels.from("longitude-field","Longitude","The event "+
- "property containing the longitude
value"),PropertyScope.MEASUREMENT_PROPERTY)
+ .requiredPropertyWithUnaryMapping(
+ EpRequirements.domainPropertyReq(Geo.LAT),
+ Labels.from("latitude-field","Latitude","The event property containing
the latitude value"),
+ PropertyScope.MEASUREMENT_PROPERTY
+ )
+ .requiredPropertyWithUnaryMapping(
+ EpRequirements.domainPropertyReq(Geo.LNG),
+ Labels.from("longitude-field","Longitude","The event property
containing the longitude value"),
+ PropertyScope.MEASUREMENT_PROPERTY
+ )
.build())
```
@@ -142,11 +152,11 @@ stream requirements would create elements with multiple
input connectors in Stre
Stream requirements can be assigned by using the `StreamRequirementsBuilder`
class.
In our example, we define two requirements, so-called _domain property
requirements_. In contrast to _data type
requirements_ where we'd expect an event property with a field of a specific
data type (e.g., float), domain property
-requirements expect a specific domain property, e.g., from a vocabulary such
as the WGS84 Geo vocab.
+requirements expect a specific semantic type (called domain property), e.g.,
from a vocabulary such as the WGS84 Geo vocab.
Once a pipeline is deployed, we are interested in the actual field (and its
field name) that contains the latitude and
longitude values.
-In some cases, there might me more than one field that satisfies a property
requirement and we would like users to
+In some cases, there might be more than one field that satisfies a property
requirement, and we would like users to
select the property the geofencing component should operate on.
Therefore, our example uses the method `requiredPropertyWithUnaryMapping`,
which will map a requirement to a real event
property of an input stream and let the user choose the appropriate field in
the StreamPipes UI when pipelines are
@@ -155,6 +165,50 @@ defined.
Finally, the `PropertyScope` indicates that the required property is a
measurement value (in contrast to a dimension
value). This allows us later to provide improved user guidance in the pipeline
editor.
+Similar to mapping properties, text parameters have an internalId (radius), a
label and a description.
+In addition, we can assign a _value specification_ to the parameter indicating
the value range we support.
+Our example supports a radius value between 0 and 1000 with a granularity of 1.
+In the StreamPipes UI, a required text parameter is rendered as a text input
field, in case we provide an optional value
+specification, a slider input is automatically generated.
+
+For now, we've assigned parameters with an internal ID, a label and a
description.
+To decouple human-readable labels and description from the actual data
processor description, it is possible to extract the strings to a properties
file.
+In the `resources` folder, switch to a folder with the same name as the data
processor's ID. If you've used the Maven archetype to build our project, there
should be a `strings.en` file.
+In this file, we can configure labels and descriptions. For instance, instead
of writing
+
+```java
+
+.requiredPropertyWithUnaryMapping(
+ EpRequirements.domainPropertyReq(Geo.LAT),
+ Labels.from("latitude-field","Latitude","The event property containing
the latitude value"),
+ PropertyScope.MEASUREMENT_PROPERTY
+ )
+
+```
+
+it is recommended to write
+
+```java
+
+.requiredPropertyWithUnaryMapping(
+ EpRequirements.domainPropertyReq(Geo.LAT),
+ Labels.withId("latitude-field"),
+ PropertyScope.MEASUREMENT_PROPERTY
+ )
+
+```
+
+and add the following line to the `strings.en` file:
+
+```properties
+
+latitude-field.title=Latitude
+latitute-field.description=The event property containing the latitude value
+
+```
+
+This feature will also ease future internationalization efforts.
+
Besides requirements, users should be able to define the center coordinate of
the Geofence and the size of the fence
defined as a radius around the center in meters.
The radius can be defined by adding a simple required text field to the
description:
@@ -163,19 +217,13 @@ The radius can be defined by adding a simple required
text field to the descript
.requiredIntegerParameter("radius","Geofence Size","The size of the circular
geofence in meters.",0,1000,1)
```
-Similar to mapping properties, text parameters have an internalId (radius), a
label and a description.
-In addition, we can assign a _value specification_ to the parameter indicating
the value range we support.
-Our example supports a radius value between 0 and 1000 with a granularity of 1.
-In the StreamPipes UI, a required text parameter is rendered as a text input
field, in case we provide an optional value
-specification, a slider input is automatically generated.
-
Such user-defined parameters are called _static properties_. There are many
different types of static properties (see
-the [Processor SDK](06_extend-sdk-static-properties.md) for an overview).
+the [Processor SDK](06_extend-sdk-static-properties.md) for an overview).
Similar to stream requirements, it is also recommended to type
`Labels.withId("radius")` and move labels and descriptions to the resource file.
In this example, we'll further add two very simple input fields to let users
provide latitude and longitude of the
geofence center.
-Add the following line to the `declareModel` method:
+Add the following line to the `declareConfig` method:
```java
.requiredFloatParameter(Labels.from(LATITUDE_KEY,"Latitude","The latitude
value"))
@@ -205,16 +253,16 @@ pipeline was created.
## Pipeline element invocation
-Once users start a pipeline that uses our geofencing component, the
_onInvocation_ method in our class is called. The
-class `ProcessorParams` includes convenient access to user-configured
parameters a users has selected in the pipeline
-editor and information on the acutal streams that are connected to the
pipeline element.
+Once users start a pipeline that uses our geofencing component, the
_onPipelineStarted_ method in our class is called. The
+interface `IDataProcessorParameters` includes convenient access to
user-configured parameters a users has selected in the pipeline
+editor and information on the actual streams that are connected to the
pipeline element.
Next, we are interested in the fields of the input event stream that contains
the latitude and longitude value we would
like to compute against the geofence center location as follows:
```java
-String
latitudeFieldName=parameters.extractor().mappingPropertyValue("latitude-field");
- String
longitudeFieldName=parameters.extractor().mappingPropertyValue("longitude-field");
+ String latitudeFieldName =
params.extractor().mappingPropertyValue("latitude-field");
+ String longitudeFieldName =
params.extractor().mappingPropertyValue("longitude-field");
```
We use the same `internalId` we've used to define the mapping property
requirements in the `declareModel` method.
@@ -223,41 +271,41 @@ Next, for extracting the geofence center coordinates, add
to class variables cen
assign the selected values using the following statements:
```java
-this.centerLatitude=parameters.extractor().singleValueParameter(LATITUDE_CENTER,Float.class);
-
this.centerLongitude=parameters.extractor().singleValueParameter(LONGITUDE_CENTER,Float.class);
+ this.centerLatitude =
params.extractor().singleValueParameter(LATITUDE_CENTER,Float.class);
+ this.centerLongitude =
params.extractor().singleValueParameter(LONGITUDE_CENTER,Float.class);
```
The radius value can be extracted as follows:
```java
-int radius=parameters.extractor().singleValueParameter("radius",Float.class);
+ int radius = params.extractor().singleValueParameter("radius",Float.class);
```
Great! That's all we need to describe a data processor for usage in
StreamPipes. Your processor class should look as
follows:
```java
+
package org.apache.streampipes.pe.example;
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.model.DataProcessorType;
-import org.apache.streampipes.model.graph.DataProcessorDescription;
+import org.apache.streampipes.extensions.api.pe.IStreamPipesDataProcessor;
+import
org.apache.streampipes.extensions.api.pe.config.IDataProcessorConfiguration;
+import
org.apache.streampipes.extensions.api.pe.context.EventProcessorRuntimeContext;
+import org.apache.streampipes.extensions.api.pe.param.IDataProcessorParameters;
+import org.apache.streampipes.extensions.api.pe.routing.SpOutputCollector;
import org.apache.streampipes.model.runtime.Event;
-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.builder.processor.DataProcessorConfiguration;
+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.vocabulary.Geo;
-import org.apache.streampipes.wrapper.context.EventProcessorRuntimeContext;
-import org.apache.streampipes.wrapper.routing.SpOutputCollector;
-import org.apache.streampipes.wrapper.standalone.ProcessorParams;
-import org.apache.streampipes.wrapper.standalone.StreamPipesDataProcessor;
+import org.apache.streampipes.sdk.helpers.SupportedFormats;
+import org.apache.streampipes.sdk.helpers.SupportedProtocols;
+import org.apache.streampipes.vocabulary.SO;
-public class GeofencingProcessor extends StreamPipesDataProcessor {
+public class GeofencingProcessor implements IStreamPipesDataProcessor {
private static final String LATITUDE_CENTER = "latitude-center";
private static final String LONGITUDE_CENTER = "longitude-center";
@@ -269,51 +317,53 @@ public class GeofencingProcessor extends
StreamPipesDataProcessor {
private int radius;
- @Override
- public DataProcessorDescription declareModel() {
- return
ProcessingElementBuilder.create("org.streampipes.tutorial-geofencing")
- .category(DataProcessorType.ENRICH)
- .withAssets(Assets.DOCUMENTATION, Assets.ICON)
- .withLocales(Locales.EN)
- .requiredStream(StreamRequirementsBuilder
- .create()
-
.requiredPropertyWithUnaryMapping(EpRequirements.domainPropertyReq(Geo.lat),
- Labels.from("latitude-field", "Latitude", "The event " +
- "property containing the latitude value"),
PropertyScope.MEASUREMENT_PROPERTY)
-
.requiredPropertyWithUnaryMapping(EpRequirements.domainPropertyReq(Geo.lng),
- Labels.from("longitude-field", "Longitude", "The event " +
- "property containing the longitude value"),
PropertyScope.MEASUREMENT_PROPERTY)
- .build())
- .outputStrategy(OutputStrategies.keep())
- .requiredIntegerParameter("radius", "Geofence Size", "The size of the
circular geofence in meters.", 0, 1000, 1)
- .requiredFloatParameter(Labels.from(LATITUDE_CENTER, "Latitude", "The
latitude value"))
- .requiredFloatParameter(Labels.from(LONGITUDE_CENTER, "Longitude",
"The longitude value"))
- .build();
+ public IDataProcessorConfiguration declareConfig() {
+ return DataProcessorConfiguration.create(
+ GeofencingProcessor::new,
+ ProcessingElementBuilder.create("org.streampipes.tutorial-geofencing")
+ .category(DataProcessorType.ENRICH)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .withLocales(Locales.EN)
+ .requiredStream(StreamRequirementsBuilder
+ .create()
+
.requiredPropertyWithUnaryMapping(EpRequirements.domainPropertyReq(Geo.lat),
+ Labels.from("latitude-field", "Latitude", "The event " +
+ "property containing the latitude value"),
PropertyScope.MEASUREMENT_PROPERTY)
+
.requiredPropertyWithUnaryMapping(EpRequirements.domainPropertyReq(Geo.lng),
+ Labels.from("longitude-field", "Longitude", "The event " +
+ "property containing the longitude value"),
PropertyScope.MEASUREMENT_PROPERTY)
+ .build())
+ .outputStrategy(OutputStrategies.keep())
+ .requiredIntegerParameter("radius", "Geofence Size", "The size of
the circular geofence in meters.", 0, 1000, 1)
+ .requiredFloatParameter(Labels.from(LATITUDE_CENTER, "Latitude",
"The latitude value"))
+ .requiredFloatParameter(Labels.from(LONGITUDE_CENTER, "Longitude",
"The longitude value"))
+ .build()
+ );
}
@Override
- public void onInvocation(ProcessorParams parameters,
- SpOutputCollector spOutputCollector,
- EventProcessorRuntimeContext runtimeContext) throws
SpRuntimeException {
- this.centerLatitude =
parameters.extractor().singleValueParameter(LATITUDE_CENTER, Float.class);
- this.centerLongitude =
parameters.extractor().singleValueParameter(LONGITUDE_CENTER, Float.class);
- this.latitudeFieldName =
parameters.extractor().mappingPropertyValue("latitude-field");
- this.longitudeFieldName =
parameters.extractor().mappingPropertyValue("longitude-field");
- this.radius = parameters.extractor().singleValueParameter("radius",
Integer.class);
+ public void onPipelineStarted(IDataProcessorParameters params,
+ SpOutputCollector collector,
+ EventProcessorRuntimeContext runtimeContext) {
+ this.centerLatitude =
params.extractor().singleValueParameter(LATITUDE_CENTER, Float.class);
+ this.centerLongitude =
params.extractor().singleValueParameter(LONGITUDE_CENTER, Float.class);
+ this.latitudeFieldName =
params.extractor().mappingPropertyValue("latitude-field");
+ this.longitudeFieldName =
params.extractor().mappingPropertyValue("longitude-field");
+ this.radius = params.extractor().singleValueParameter("radius",
Integer.class);
}
@Override
- public void onEvent(Event event, SpOutputCollector collector) throws
SpRuntimeException {
+ public void onEvent(Event event,
+ SpOutputCollector collector) {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() {
}
}
-
```
## Adding an implementation
@@ -324,29 +374,30 @@ Add the following piece of code to the onEvent method,
which realizes the Geofen
```java
-@Override
-public void onEvent(Event event,SpOutputCollector collector)throws
SpRuntimeException{
- float
latitude=event.getFieldBySelector(latitudeFieldName).getAsPrimitive().getAsFloat();
- float
longitude=event.getFieldBySelector(longitudeFieldName).getAsPrimitive().getAsFloat();
-
- float distance=distFrom(latitude,longitude,centerLatitude,centerLongitude);
-
- if(distance<=radius){
- collector.collect(event);
- }
+ @Override
+ public void onEvent(Event event,
+ SpOutputCollector collector) {
+ float latitude =
event.getFieldBySelector(latitudeFieldName).getAsPrimitive().getAsFloat();
+ float longitude =
event.getFieldBySelector(longitudeFieldName).getAsPrimitive().getAsFloat();
+
+ float distance = distFrom(latitude,longitude, centerLatitude,
centerLongitude);
+
+ if(distance <= radius){
+ collector.collect(event);
+ }
}
-
-public static float distFrom(float lat1,float lng1,float lat2,float lng2){
- double earthRadius=6371000;
- double dLat=Math.toRadians(lat2-lat1);
- double dLng=Math.toRadians(lng2-lng1);
- double a=Math.sin(dLat/2)*Math.sin(dLat/2)+
- Math.cos(Math.toRadians(lat1))*Math.cos(Math.toRadians(lat2))*
- Math.sin(dLng/2)*Math.sin(dLng/2);
- double c=2*Math.atan2(Math.sqrt(a),Math.sqrt(1-a));
- return(float)(earthRadius*c);
- }
-
+
+ public static float distFrom(float lat1, float lng1, float lat2, float
lng2) {
+ double earthRadius = 6371000;
+ double dLat = Math.toRadians(lat2-lat1);
+ double dLng = Math.toRadians(lng2-lng1);
+ double a = Math.sin(dLat/2)*Math.sin(dLat/2) +
+ Math.cos(Math.toRadians(lat1))*Math.cos(Math.toRadians(lat2)) *
+ Math.sin(dLng/2)*Math.sin(dLng/2);
+
+ double c = 2*Math.atan2(Math.sqrt(a),Math.sqrt(1-a));
+
+ return(float)(earthRadius*c);
}
```
diff --git a/docs/06_extend-tutorial-data-sinks.md
b/docs/06_extend-tutorial-data-sinks.md
index 69ada8553..09baeff71 100644
--- a/docs/06_extend-tutorial-data-sinks.md
+++ b/docs/06_extend-tutorial-data-sinks.md
@@ -26,7 +26,7 @@ Enter the following command in a command line of your choice
(Apache Maven needs
```
mvn archetype:generate -DarchetypeGroupId=org.apache.streampipes \
--DarchetypeArtifactId=streampipes-archetype-extensions-jvm
-DarchetypeVersion=0.70.0 \
+-DarchetypeArtifactId=streampipes-archetype-extensions-jvm
-DarchetypeVersion=0.93.0 \
-DgroupId=org.streampipes.tutorial -DartifactId=sink-tutorial
-DclassNamePrefix=Rest -DpackageName=mypackage
```
@@ -49,63 +49,66 @@ Create a class `RestSink` which should look as follows:
```java
package org.apache.streampipes.pe.example;
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.extensions.api.pe.IStreamPipesDataSink;
+import org.apache.streampipes.extensions.api.pe.config.IDataSinkConfiguration;
+import
org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
+import org.apache.streampipes.extensions.api.pe.param.IDataSinkParameters;
import org.apache.streampipes.model.DataSinkType;
-import org.apache.streampipes.model.graph.DataSinkDescription;
import org.apache.streampipes.model.runtime.Event;
import org.apache.streampipes.model.schema.PropertyScope;
import org.apache.streampipes.sdk.builder.DataSinkBuilder;
import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+import org.apache.streampipes.sdk.builder.sink.DataSinkConfiguration;
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.utils.Assets;
-import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
-import org.apache.streampipes.wrapper.standalone.SinkParams;
-import org.apache.streampipes.wrapper.standalone.StreamPipesDataSink;
-public class RestSink extends StreamPipesDataSink {
+public class RestSink implements IStreamPipesDataSink {
@Override
- public DataSinkDescription declareModel() {
- return
DataSinkBuilder.create("org.apache.streampipes.tutorial.pe.sink.rest")
- .category(DataSinkType.NOTIFICATION)
- .withAssets(Assets.DOCUMENTATION, Assets.ICON)
- .withLocales(Locales.EN)
- .requiredStream(StreamRequirementsBuilder
- .create()
- .requiredPropertyWithNaryMapping(EpRequirements.anyProperty(),
Labels.withId(
- "fields-to-send"), PropertyScope.NONE)
- .build())
- .build();
+ public IDataSinkConfiguration declareConfig() {
+ return DataSinkConfiguration.create(
+ RestSink::new,
+
DataSinkBuilder.create("org.apache.streampipes.tutorial.pe.sink.rest")
+ .category(DataSinkType.NOTIFICATION)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .withLocales(Locales.EN)
+ .requiredStream(StreamRequirementsBuilder
+ .create()
+
.requiredPropertyWithNaryMapping(EpRequirements.anyProperty(), Labels.withId(
+ "fields-to-send"), PropertyScope.NONE)
+ .build())
+ .build()
+ );
}
@Override
- public void onInvocation(SinkParams parameters, EventSinkRuntimeContext
runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(IDataSinkParameters params,
+ EventSinkRuntimeContext
eventSinkRuntimeContext) {
}
@Override
- public void onEvent(Event event) throws SpRuntimeException {
+ public void onEvent(Event event) {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() {
}
-}
```
-In this class, we need to implement three methods: The `declareModel` method
is used to define abstract stream
+In this class, we need to implement three methods: The `declareConfig` method
is used to define abstract stream
requirements such as event properties that must be present in any input stream
that is later connected to the element
using the StreamPipes UI.
-The second method, `onInvocation` is called once a pipeline using this sink is
started. The third method, `onEvent`, is
+The second method, `onPipelineStarted` is called once a pipeline using this
sink is started. The third method, `onEvent`, is
called for every incoming event.
-The ``declareModel`` method describes the properties of our data sink:
+The `DataSinkBuilder` within the ``declareConfig`` method describes the
properties of our data sink:
* ``category`` defines a category for this sink.
* ``withAssets`` denotes that we will provide an external documentation file
and an icon, which can be found in
@@ -116,7 +119,7 @@ The ``declareModel`` method describes the properties of our
data sink:
display a list of available fields from the connected input event, where
users can select a subset. This is defined by
defining a Mapping from the empty requirement. This will later on render a
selection dialog in the pipeline editor.
-The ``onInvocation`` method is called when a pipeline containing the sink is
started. Once a pipeline is started, we
+The ``onPipelineStarted`` method is called when a pipeline containing the sink
is started. Once a pipeline is started, we
would like to extract user-defined parameters.
In this example, we simply extract the fields selected by users that should be
forwarded to the REST sink. Finally, we
return a new configured event sink containing the parameters.
@@ -124,8 +127,8 @@ return a new configured event sink containing the
parameters.
## Pipeline element invocation
Once users start a pipeline that uses our geofencing component, the
_onInvocation_ method in our class is called. The
-class `SinkParams` includes a graph that contains information on the
configuration parameters a users has selected in
-the pipeline editor and information on the acutal streams that are connected
to the pipeline element.
+interface `IDataSinkParameters` includes methods to extract the configuration
parameters a user has selected in
+the pipeline editor and information on the actual streams that are connected
to the pipeline element.
## Adding an implementation
@@ -136,27 +139,27 @@ Our final class should look as follows:
```java
package org.apache.streampipes.pe.example;
-import com.google.common.base.Charsets;
-import org.apache.http.client.fluent.Request;
-import org.apache.http.entity.StringEntity;
-
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
import org.apache.streampipes.dataformat.SpDataFormatDefinition;
import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
+import org.apache.streampipes.extensions.api.pe.IStreamPipesDataSink;
+import org.apache.streampipes.extensions.api.pe.config.IDataSinkConfiguration;
+import
org.apache.streampipes.extensions.api.pe.context.EventSinkRuntimeContext;
+import org.apache.streampipes.extensions.api.pe.param.IDataSinkParameters;
import org.apache.streampipes.model.DataSinkType;
-import org.apache.streampipes.model.graph.DataSinkDescription;
import org.apache.streampipes.model.runtime.Event;
import org.apache.streampipes.model.schema.PropertyScope;
import org.apache.streampipes.sdk.builder.DataSinkBuilder;
import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+import org.apache.streampipes.sdk.builder.sink.DataSinkConfiguration;
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.utils.Assets;
-import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
-import org.apache.streampipes.wrapper.standalone.SinkParams;
-import org.apache.streampipes.wrapper.standalone.StreamPipesDataSink;
+import com.google.common.base.Charsets;
+import org.apache.http.client.fluent.Request;
+import org.apache.http.entity.StringEntity;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -164,7 +167,7 @@ import java.io.IOException;
import java.util.List;
import java.util.Map;
-public class RestSink extends StreamPipesDataSink {
+public class RestSink implements IStreamPipesDataSink {
private static final Logger LOG = LoggerFactory.getLogger(RestSink.class);
@@ -173,19 +176,31 @@ public class RestSink extends StreamPipesDataSink {
private SpDataFormatDefinition dataFormatDefinition;
@Override
- public DataSinkDescription declareModel() {
- ...
+ public IDataSinkConfiguration declareConfig() {
+ return DataSinkConfiguration.create(
+ RestSink::new,
+
DataSinkBuilder.create("org.apache.streampipes.tutorial.pe.sink.rest")
+ .category(DataSinkType.NOTIFICATION)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .withLocales(Locales.EN)
+ .requiredStream(StreamRequirementsBuilder
+ .create()
+
.requiredPropertyWithNaryMapping(EpRequirements.anyProperty(), Labels.withId(
+ "fields-to-send"), PropertyScope.NONE)
+ .build())
+ .build()
+ );
}
@Override
- public void onInvocation(SinkParams parameters, EventSinkRuntimeContext
runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(IDataSinkParameters params,
+ EventSinkRuntimeContext
eventSinkRuntimeContext) {
this.dataFormatDefinition = new JsonDataFormatDefinition();
- this.fieldsToSend =
parameters.extractor().mappingPropertyValues("fields-to-send");
-
+ this.fieldsToSend =
params.extractor().mappingPropertyValues("fields-to-send");
}
@Override
- public void onEvent(Event event) throws SpRuntimeException {
+ public void onEvent(Event event) {
Map<String, Object> outEventMap = event.getSubset(fieldsToSend).getRaw();
try {
String json = new String(dataFormatDefinition.fromMap(outEventMap));
@@ -198,7 +213,7 @@ public class RestSink extends StreamPipesDataSink {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() {
}
}
