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/02/23 11:14:19 UTC
[incubator-streampipes-extensions] branch STREAMPIPES-81 updated:
STREAMPIPES-81: Rename wrapper interfaces
This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch STREAMPIPES-81
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes-extensions.git
The following commit(s) were added to refs/heads/STREAMPIPES-81 by this push:
new 420f6f3 STREAMPIPES-81: Rename wrapper interfaces
420f6f3 is described below
commit 420f6f3e3f6e0865ec0e334308ea6e0a3da8b799
Author: Dominik Riemer <ri...@fzi.de>
AuthorDate: Sun Feb 23 12:14:02 2020 +0100
STREAMPIPES-81: Rename wrapper interfaces
---
.../enricher/jvm/processor/sizemeasure/SizeMeasure.java | 4 ++--
.../processors/filters/jvm/processor/compose/Compose.java | 4 ++--
.../filters/jvm/processor/numericalfilter/NumericalFilter.java | 4 ++--
.../processors/filters/jvm/processor/projection/Projection.java | 4 ++--
.../processors/filters/jvm/processor/textfilter/TextFilter.java | 4 ++--
.../filters/jvm/processor/threshold/ThresholdDetection.java | 4 ++--
.../streampipes/processors/siddhi/trend/TestTrendProcessor.java | 2 +-
.../geo/jvm/processor/distancecalculator/DistanceCalculator.java | 4 ++--
.../geo/jvm/processor/geocoder/GoogleMapsGeocoding.java | 4 ++--
.../geo/jvm/processor/revgeocoder/ReverseGeocoding.java | 4 ++--
.../jvm/processor/staticgeocoder/StaticGoogleMapsGeocoder.java | 4 ++--
.../genericclassification/GenericImageClassification.java | 4 ++--
.../imageprocessing/jvm/processor/imagecropper/ImageCropper.java | 4 ++--
.../jvm/processor/imageenrichment/ImageEnricher.java | 4 ++--
.../imageprocessing/jvm/processor/qrreader/QrCodeReader.java | 4 ++--
.../processors/textmining/jvm/processor/chunker/Chunker.java | 8 ++++----
.../textmining/jvm/processor/language/LanguageDetection.java | 8 ++++----
.../textmining/jvm/processor/namefinder/NameFinder.java | 8 ++++----
.../textmining/jvm/processor/partofspeech/PartOfSpeech.java | 8 ++++----
.../jvm/processor/sentencedetection/SentenceDetection.java | 8 ++++----
.../processors/textmining/jvm/processor/tokenizer/Tokenizer.java | 8 ++++----
.../transformation/jvm/processor/array/count/CountArray.java | 4 ++--
.../transformation/jvm/processor/array/split/SplitArray.java | 4 ++--
.../jvm/processor/booloperator/counter/BooleanCounter.java | 8 ++++----
.../jvm/processor/booloperator/inverter/BooleanInverter.java | 8 ++++----
.../processor/booloperator/timekeeping/BooleanTimekeeping.java | 8 ++++----
.../jvm/processor/booloperator/timer/BooleanTimer.java | 8 ++++----
.../jvm/processor/csvmetadata/CsvMetadataEnrichment.java | 4 ++--
.../transformation/jvm/processor/task/TaskDuration.java | 4 ++--
.../jvm/processor/timestampextractor/TimestampExtractor.java | 4 ++--
.../jvm/processor/transformtoboolean/TransformToBoolean.java | 8 ++++----
.../jvm/processor/value/change/ChangedValueDetection.java | 8 ++++----
.../jvm/processor/value/duration/CalculateDuration.java | 8 ++++----
.../streampipes/sinks/brokers/jvm/bufferrest/BufferRest.java | 4 ++--
.../apache/streampipes/sinks/brokers/jvm/jms/JmsPublisher.java | 4 ++--
.../streampipes/sinks/brokers/jvm/kafka/KafkaPublisher.java | 4 ++--
.../org/apache/streampipes/sinks/brokers/jvm/pulsar/Pulsar.java | 6 +++---
.../streampipes/sinks/brokers/jvm/rabbitmq/RabbitMqConsumer.java | 4 ++--
.../apache/streampipes/sinks/brokers/jvm/rest/RestPublisher.java | 4 ++--
.../apache/streampipes/sinks/databases/jvm/couchdb/CouchDb.java | 4 ++--
.../org/apache/streampipes/sinks/databases/jvm/ditto/Ditto.java | 4 ++--
.../apache/streampipes/sinks/databases/jvm/influxdb/InfluxDb.java | 4 ++--
.../org/apache/streampipes/sinks/databases/jvm/iotdb/IotDb.java | 4 ++--
.../org/apache/streampipes/sinks/databases/jvm/opcua/OpcUa.java | 4 ++--
.../streampipes/sinks/databases/jvm/postgresql/PostgreSql.java | 4 ++--
.../streampipes/sinks/internal/jvm/dashboard/Dashboard.java | 4 ++--
.../apache/streampipes/sinks/internal/jvm/datalake/DataLake.java | 4 ++--
.../sinks/internal/jvm/notification/NotificationProducer.java | 4 ++--
.../streampipes/sinks/notifications/jvm/email/EmailPublisher.java | 4 ++--
.../sinks/notifications/jvm/onesignal/OneSignalProducer.java | 4 ++--
.../sinks/notifications/jvm/slack/SlackNotification.java | 4 ++--
51 files changed, 128 insertions(+), 128 deletions(-)
diff --git a/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/processor/sizemeasure/SizeMeasure.java b/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/processor/sizemeasure/SizeMeasure.java
index 66f545d..f4feda2 100644
--- a/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/processor/sizemeasure/SizeMeasure.java
+++ b/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/processor/sizemeasure/SizeMeasure.java
@@ -29,12 +29,12 @@ public class SizeMeasure implements EventProcessor<SizeMeasureParameters> {
private SizeMeasureParameters sizeMeasureParameters;
@Override
- public void onInvocation(SizeMeasureParameters sizeMeasureParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(SizeMeasureParameters sizeMeasureParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
this.sizeMeasureParameters = sizeMeasureParameters;
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
@Override
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/Compose.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/Compose.java
index 5cb9620..47cf726 100644
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/Compose.java
+++ b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/Compose.java
@@ -37,14 +37,14 @@ public class Compose implements EventProcessor<ComposeParameters> {
@Override
- public void onInvocation(ComposeParameters composeParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(ComposeParameters composeParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
this.outputSchema = composeParameters.getGraph().getOutputStream().getEventSchema();
this.outputKeySelectors = composeParameters.getOutputKeySelectors();
this.lastEvents = new HashMap<>();
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
this.lastEvents.clear();
}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericalfilter/NumericalFilter.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericalfilter/NumericalFilter.java
index ddc18db..aa5fa06 100644
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericalfilter/NumericalFilter.java
+++ b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/numericalfilter/NumericalFilter.java
@@ -28,7 +28,7 @@ public class NumericalFilter implements EventProcessor<NumericalFilterParameters
private NumericalFilterParameters params;
@Override
- public void onInvocation(NumericalFilterParameters numericalFilterParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext
+ public void onPipelineStarted(NumericalFilterParameters numericalFilterParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext
runtimeContext) {
this.params = numericalFilterParameters;
}
@@ -63,7 +63,7 @@ public class NumericalFilter implements EventProcessor<NumericalFilterParameters
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/projection/Projection.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/projection/Projection.java
index 30afb91..65f4f1e 100644
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/projection/Projection.java
+++ b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/projection/Projection.java
@@ -31,7 +31,7 @@ public class Projection implements EventProcessor<ProjectionParameters> {
@Override
- public void onInvocation(ProjectionParameters projectionParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(ProjectionParameters projectionParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
this.outputKeys = projectionParameters.getOutputKeys();
}
@@ -41,7 +41,7 @@ public class Projection implements EventProcessor<ProjectionParameters> {
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/textfilter/TextFilter.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/textfilter/TextFilter.java
index fb2447c..e265684 100644
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/textfilter/TextFilter.java
+++ b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/textfilter/TextFilter.java
@@ -28,7 +28,7 @@ public class TextFilter implements EventProcessor<TextFilterParameters> {
private TextFilterParameters params;
@Override
- public void onInvocation(TextFilterParameters textFilterParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(TextFilterParameters textFilterParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
this.params = textFilterParameters;
}
@@ -51,7 +51,7 @@ public class TextFilter implements EventProcessor<TextFilterParameters> {
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/threshold/ThresholdDetection.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/threshold/ThresholdDetection.java
index 3ffa7af..a8b0727 100644
--- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/threshold/ThresholdDetection.java
+++ b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/threshold/ThresholdDetection.java
@@ -28,7 +28,7 @@ public class ThresholdDetection implements EventProcessor<ThresholdDetectionPara
private ThresholdDetectionParameters params;
@Override
- public void onInvocation(ThresholdDetectionParameters numericalFilterParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext
+ public void onPipelineStarted(ThresholdDetectionParameters numericalFilterParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext
runtimeContext) {
this.params = numericalFilterParameters;
}
@@ -67,7 +67,7 @@ public class ThresholdDetection implements EventProcessor<ThresholdDetectionPara
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java b/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java
index 2dc2c4d..6894d10 100644
--- a/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java
+++ b/streampipes-processors-filters-siddhi/src/test/java/org/apache/streampipes/processors/siddhi/trend/TestTrendProcessor.java
@@ -110,7 +110,7 @@ public class TestTrendProcessor {
"::randomValue");
Trend trend = new Trend(event -> actualMatchCount[0]++);
- trend.onInvocation(params, null, null);
+ trend.onPipelineStarted(params, null, null);
sendEvents(trend);
LOG.info("Expected match count is {}", expectedMatchCount);
diff --git a/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/processor/distancecalculator/DistanceCalculator.java b/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/processor/distancecalculator/DistanceCalculator.java
index a4ba37e..08bf6f0 100644
--- a/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/processor/distancecalculator/DistanceCalculator.java
+++ b/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/processor/distancecalculator/DistanceCalculator.java
@@ -28,7 +28,7 @@ public class DistanceCalculator implements EventProcessor<DistanceCalculatorPara
private DistanceCalculatorParameters params;
@Override
- public void onInvocation(DistanceCalculatorParameters numericalFilterParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext
+ public void onPipelineStarted(DistanceCalculatorParameters numericalFilterParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext
runtimeContext) {
this.params = numericalFilterParameters;
}
@@ -49,7 +49,7 @@ public class DistanceCalculator implements EventProcessor<DistanceCalculatorPara
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
diff --git a/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/processor/geocoder/GoogleMapsGeocoding.java b/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/processor/geocoder/GoogleMapsGeocoding.java
index cf03b14..61072a5 100644
--- a/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/processor/geocoder/GoogleMapsGeocoding.java
+++ b/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/processor/geocoder/GoogleMapsGeocoding.java
@@ -35,7 +35,7 @@ public class GoogleMapsGeocoding implements EventProcessor<GoogleMapsGeocodingPa
private String placeField;
@Override
- public void onInvocation(GoogleMapsGeocodingParameters parameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(GoogleMapsGeocodingParameters parameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) throws SpRuntimeException {
this.placeField = parameters.getPlaceField();
String googleMapsApiKey = parameters.getGoogleMapsApiKey();
@@ -75,7 +75,7 @@ public class GoogleMapsGeocoding implements EventProcessor<GoogleMapsGeocodingPa
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
}
}
diff --git a/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/processor/revgeocoder/ReverseGeocoding.java b/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/processor/revgeocoder/ReverseGeocoding.java
index 59d42bd..08a9a8f 100644
--- a/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/processor/revgeocoder/ReverseGeocoding.java
+++ b/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/processor/revgeocoder/ReverseGeocoding.java
@@ -37,7 +37,7 @@ public class ReverseGeocoding implements EventProcessor<ReverseGeocodingParamete
private ReverseGeoCode reverseGeoCode;
@Override
- public void onInvocation(ReverseGeocodingParameters parameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(ReverseGeocodingParameters parameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) throws SpRuntimeException {
this.latitudeField = parameters.getLatitudeField();
this.longitudeField = parameters.getLongitudeField();
@@ -66,7 +66,7 @@ public class ReverseGeocoding implements EventProcessor<ReverseGeocodingParamete
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
}
}
diff --git a/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/processor/staticgeocoder/StaticGoogleMapsGeocoder.java b/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/processor/staticgeocoder/StaticGoogleMapsGeocoder.java
index b36adaf..0721721 100644
--- a/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/processor/staticgeocoder/StaticGoogleMapsGeocoder.java
+++ b/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/processor/staticgeocoder/StaticGoogleMapsGeocoder.java
@@ -35,7 +35,7 @@ public class StaticGoogleMapsGeocoder implements EventProcessor<StaticGoogleMaps
private Double longitude;
@Override
- public void onInvocation(StaticGoogleMapsGeocodingParameters parameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(StaticGoogleMapsGeocodingParameters parameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) throws SpRuntimeException {
String place = parameters.getPlace();
String googleMapsApiKey = parameters.getGoogleMapsApiKey();
@@ -68,7 +68,7 @@ public class StaticGoogleMapsGeocoder implements EventProcessor<StaticGoogleMaps
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
}
diff --git a/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/processor/genericclassification/GenericImageClassification.java b/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/processor/genericclassification/GenericImageClassification.java
index ba52613..ac21355 100644
--- a/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/processor/genericclassification/GenericImageClassification.java
+++ b/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/processor/genericclassification/GenericImageClassification.java
@@ -46,7 +46,7 @@ public class GenericImageClassification implements EventProcessor<GenericImageCl
private List<String> categories;
@Override
- public void onInvocation(GenericImageClassificationParameters genericImageClassificationParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(GenericImageClassificationParameters genericImageClassificationParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
this.params = genericImageClassificationParameters;
//this.cs = FactoryImageClassifier.vgg_cifar10(); // Test set 89.9% for 10 categories
ClassifierAndSource cs = FactoryImageClassifier.nin_imagenet(); // Test set 62.6% for 1000 categories
@@ -97,7 +97,7 @@ public class GenericImageClassification implements EventProcessor<GenericImageCl
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/processor/imagecropper/ImageCropper.java b/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/processor/imagecropper/ImageCropper.java
index 7c32ffd..4e3f5af 100644
--- a/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/processor/imagecropper/ImageCropper.java
+++ b/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/processor/imagecropper/ImageCropper.java
@@ -35,7 +35,7 @@ public class ImageCropper implements EventProcessor<ImageCropperParameters> {
private ImageCropperParameters params;
@Override
- public void onInvocation(ImageCropperParameters imageCropperParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(ImageCropperParameters imageCropperParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
this.params = imageCropperParameters;
}
@@ -66,7 +66,7 @@ public class ImageCropper implements EventProcessor<ImageCropperParameters> {
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/processor/imageenrichment/ImageEnricher.java b/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/processor/imageenrichment/ImageEnricher.java
index 7ee3240..7733717 100644
--- a/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/processor/imageenrichment/ImageEnricher.java
+++ b/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/processor/imageenrichment/ImageEnricher.java
@@ -34,7 +34,7 @@ public class ImageEnricher implements EventProcessor<ImageEnrichmentParameters>
private ImageEnrichmentParameters params;
@Override
- public void onInvocation(ImageEnrichmentParameters params, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(ImageEnrichmentParameters params, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
this.params = params;
}
@@ -75,7 +75,7 @@ public class ImageEnricher implements EventProcessor<ImageEnrichmentParameters>
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/processor/qrreader/QrCodeReader.java b/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/processor/qrreader/QrCodeReader.java
index c275809..64cdb3e 100644
--- a/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/processor/qrreader/QrCodeReader.java
+++ b/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/processor/qrreader/QrCodeReader.java
@@ -42,7 +42,7 @@ public class QrCodeReader implements EventProcessor<QrCodeReaderParameters> {
private static final Logger LOG = LoggerFactory.getLogger(QrCodeReader.class);
@Override
- public void onInvocation(QrCodeReaderParameters qrCodeReaderParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(QrCodeReaderParameters qrCodeReaderParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
this.params = qrCodeReaderParameters;
this.sendIfNoResult = qrCodeReaderParameters.getSendIfNoResult();
this.placeholderValue = qrCodeReaderParameters.getPlaceholderValue();
@@ -90,7 +90,7 @@ public class QrCodeReader implements EventProcessor<QrCodeReaderParameters> {
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/chunker/Chunker.java b/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/chunker/Chunker.java
index 96a31e9..73fe720 100644
--- a/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/chunker/Chunker.java
+++ b/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/chunker/Chunker.java
@@ -53,9 +53,9 @@ public class Chunker implements EventProcessor<ChunkerParameters> {
}
@Override
- public void onInvocation(ChunkerParameters chunkerParameters,
- SpOutputCollector spOutputCollector,
- EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(ChunkerParameters chunkerParameters,
+ SpOutputCollector spOutputCollector,
+ EventProcessorRuntimeContext runtimeContext) {
LOG = chunkerParameters.getGraph().getLogger(Chunker.class);
this.tags = chunkerParameters.getTags();
this.tokens = chunkerParameters.getTokens();
@@ -82,6 +82,6 @@ public class Chunker implements EventProcessor<ChunkerParameters> {
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/language/LanguageDetection.java b/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/language/LanguageDetection.java
index af03e70..ba6ffd3 100644
--- a/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/language/LanguageDetection.java
+++ b/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/language/LanguageDetection.java
@@ -48,9 +48,9 @@ public class LanguageDetection implements EventProcessor<LanguageDetectionParame
}
@Override
- public void onInvocation(LanguageDetectionParameters languageDetectionParameters,
- SpOutputCollector spOutputCollector,
- EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(LanguageDetectionParameters languageDetectionParameters,
+ SpOutputCollector spOutputCollector,
+ EventProcessorRuntimeContext runtimeContext) {
LOG = languageDetectionParameters.getGraph().getLogger(LanguageDetection.class);
this.detection = languageDetectionParameters.getDetectionName();
}
@@ -67,6 +67,6 @@ public class LanguageDetection implements EventProcessor<LanguageDetectionParame
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/namefinder/NameFinder.java b/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/namefinder/NameFinder.java
index 3b89056..0e4f0c4 100644
--- a/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/namefinder/NameFinder.java
+++ b/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/namefinder/NameFinder.java
@@ -48,9 +48,9 @@ public class NameFinder implements EventProcessor<NameFinderParameters> {
}
@Override
- public void onInvocation(NameFinderParameters nameFinderParameters,
- SpOutputCollector spOutputCollector,
- EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(NameFinderParameters nameFinderParameters,
+ SpOutputCollector spOutputCollector,
+ EventProcessorRuntimeContext runtimeContext) {
LOG = nameFinderParameters.getGraph().getLogger(NameFinder.class);
String modelPath = TextMiningJvmConfig.INSTANCE.getModelDirectory() + nameFinderParameters.getModel();
@@ -76,7 +76,7 @@ public class NameFinder implements EventProcessor<NameFinderParameters> {
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
private void loadModel(String modelFile) {
diff --git a/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/partofspeech/PartOfSpeech.java b/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/partofspeech/PartOfSpeech.java
index d0751ae..ea046f2 100644
--- a/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/partofspeech/PartOfSpeech.java
+++ b/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/partofspeech/PartOfSpeech.java
@@ -47,9 +47,9 @@ public class PartOfSpeech implements EventProcessor<PartOfSpeechParameters> {
}
@Override
- public void onInvocation(PartOfSpeechParameters partOfSpeechParameters,
- SpOutputCollector spOutputCollector,
- EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(PartOfSpeechParameters partOfSpeechParameters,
+ SpOutputCollector spOutputCollector,
+ EventProcessorRuntimeContext runtimeContext) {
LOG = partOfSpeechParameters.getGraph().getLogger(PartOfSpeech.class);
this.detection = partOfSpeechParameters.getDetectionName();
}
@@ -69,6 +69,6 @@ public class PartOfSpeech implements EventProcessor<PartOfSpeechParameters> {
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/sentencedetection/SentenceDetection.java b/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/sentencedetection/SentenceDetection.java
index c680a36..5c5a73b 100644
--- a/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/sentencedetection/SentenceDetection.java
+++ b/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/sentencedetection/SentenceDetection.java
@@ -47,9 +47,9 @@ public class SentenceDetection implements EventProcessor<SentenceDetectionParame
}
@Override
- public void onInvocation(SentenceDetectionParameters sentenceDetectionParameters,
- SpOutputCollector spOutputCollector,
- EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(SentenceDetectionParameters sentenceDetectionParameters,
+ SpOutputCollector spOutputCollector,
+ EventProcessorRuntimeContext runtimeContext) {
LOG = sentenceDetectionParameters.getGraph().getLogger(SentenceDetection.class);
this.detection = sentenceDetectionParameters.getDetectionName();
}
@@ -67,6 +67,6 @@ public class SentenceDetection implements EventProcessor<SentenceDetectionParame
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/tokenizer/Tokenizer.java b/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/tokenizer/Tokenizer.java
index d2e9bf3..97f7ee2 100644
--- a/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/tokenizer/Tokenizer.java
+++ b/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/processor/tokenizer/Tokenizer.java
@@ -47,9 +47,9 @@ public class Tokenizer implements EventProcessor<TokenizerParameters> {
}
@Override
- public void onInvocation(TokenizerParameters tokenizerParameters,
- SpOutputCollector spOutputCollector,
- EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(TokenizerParameters tokenizerParameters,
+ SpOutputCollector spOutputCollector,
+ EventProcessorRuntimeContext runtimeContext) {
LOG = tokenizerParameters.getGraph().getLogger(Tokenizer.class);
this.detection = tokenizerParameters.getDetectionName();
}
@@ -64,6 +64,6 @@ public class Tokenizer implements EventProcessor<TokenizerParameters> {
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/array/count/CountArray.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/array/count/CountArray.java
index efe2e81..48588da 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/array/count/CountArray.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/array/count/CountArray.java
@@ -34,7 +34,7 @@ public class CountArray implements EventProcessor<CountArrayParameters> {
private CountArrayParameters splitArrayParameters;
@Override
- public void onInvocation(CountArrayParameters params, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(CountArrayParameters params, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
LOG = params.getGraph().getLogger(CountArray.class);
this.splitArrayParameters = params;
@@ -53,6 +53,6 @@ public class CountArray implements EventProcessor<CountArrayParameters> {
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/array/split/SplitArray.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/array/split/SplitArray.java
index 5eb1f7a..00fbeec 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/array/split/SplitArray.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/array/split/SplitArray.java
@@ -39,7 +39,7 @@ public class SplitArray implements EventProcessor<SplitArrayParameters> {
@Override
- public void onInvocation(SplitArrayParameters splitArrayParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(SplitArrayParameters splitArrayParameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
LOG = splitArrayParameters.getGraph().getLogger(SplitArray.class);
this.splitArrayParameters = splitArrayParameters;
}
@@ -80,6 +80,6 @@ public class SplitArray implements EventProcessor<SplitArrayParameters> {
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/BooleanCounter.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/BooleanCounter.java
index 4613ce7..a058a16 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/BooleanCounter.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/counter/BooleanCounter.java
@@ -39,9 +39,9 @@ public class BooleanCounter implements EventProcessor<BooleanCounterParameters>
@Override
- public void onInvocation(BooleanCounterParameters booleanCounterParametersParameters,
- SpOutputCollector spOutputCollector,
- EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(BooleanCounterParameters booleanCounterParametersParameters,
+ SpOutputCollector spOutputCollector,
+ EventProcessorRuntimeContext runtimeContext) {
LOG = booleanCounterParametersParameters.getGraph().getLogger(BooleanCounter.class);
this.fieldName = booleanCounterParametersParameters.getInvertFieldName();
this.flankUp = booleanCounterParametersParameters.getFlankUp();
@@ -87,7 +87,7 @@ public class BooleanCounter implements EventProcessor<BooleanCounterParameters>
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/inverter/BooleanInverter.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/inverter/BooleanInverter.java
index 16758a4..c35a8b6 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/inverter/BooleanInverter.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/inverter/BooleanInverter.java
@@ -33,9 +33,9 @@ public class BooleanInverter implements EventProcessor<BooleanInverterParameters
@Override
- public void onInvocation(BooleanInverterParameters booleanInverterParameters,
- SpOutputCollector spOutputCollector,
- EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(BooleanInverterParameters booleanInverterParameters,
+ SpOutputCollector spOutputCollector,
+ EventProcessorRuntimeContext runtimeContext) {
LOG = booleanInverterParameters.getGraph().getLogger(BooleanInverter.class);
this.invertFieldName = booleanInverterParameters.getInvertFieldName();
}
@@ -50,6 +50,6 @@ public class BooleanInverter implements EventProcessor<BooleanInverterParameters
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/timekeeping/BooleanTimekeeping.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/timekeeping/BooleanTimekeeping.java
index 84b95ea..73e034e 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/timekeeping/BooleanTimekeeping.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/timekeeping/BooleanTimekeeping.java
@@ -42,9 +42,9 @@ public class BooleanTimekeeping implements EventProcessor<BooleanTimekeepingPara
private double outputDivisor;
@Override
- public void onInvocation(BooleanTimekeepingParameters booleanInverterParameters,
- SpOutputCollector spOutputCollector,
- EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(BooleanTimekeepingParameters booleanInverterParameters,
+ SpOutputCollector spOutputCollector,
+ EventProcessorRuntimeContext runtimeContext) {
LOG = booleanInverterParameters.getGraph().getLogger(BooleanTimekeeping.class);
this.leftFieldName = booleanInverterParameters.getLeftFieldName();
this.rightFieldName = booleanInverterParameters.getRightFieldName();
@@ -89,7 +89,7 @@ public class BooleanTimekeeping implements EventProcessor<BooleanTimekeepingPara
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/timer/BooleanTimer.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/timer/BooleanTimer.java
index 55fb9a9..69f917b 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/timer/BooleanTimer.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/booloperator/timer/BooleanTimer.java
@@ -37,9 +37,9 @@ public class BooleanTimer implements EventProcessor<BooleanTimerParameters> {
@Override
- public void onInvocation(BooleanTimerParameters booleanInverterParameters,
- SpOutputCollector spOutputCollector,
- EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(BooleanTimerParameters booleanInverterParameters,
+ SpOutputCollector spOutputCollector,
+ EventProcessorRuntimeContext runtimeContext) {
LOG = booleanInverterParameters.getGraph().getLogger(BooleanTimer.class);
this.fieldName = booleanInverterParameters.getFieldName();
this.measureTrue = booleanInverterParameters.isMeasureTrue();
@@ -71,7 +71,7 @@ public class BooleanTimer implements EventProcessor<BooleanTimerParameters> {
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
public static void main(String... args) {
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/csvmetadata/CsvMetadataEnrichment.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/csvmetadata/CsvMetadataEnrichment.java
index 9d955cd..6c39004 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/csvmetadata/CsvMetadataEnrichment.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/csvmetadata/CsvMetadataEnrichment.java
@@ -51,7 +51,7 @@ public class CsvMetadataEnrichment implements EventProcessor<CsvMetadataEnrichme
private Map<String, CSVRecord> columnMap;
@Override
- public void onInvocation(CsvMetadataEnrichmentParameters parameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(CsvMetadataEnrichmentParameters parameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) throws SpRuntimeException {
this.mappingFieldSelector = parameters.getMappingFieldSelector();
this.matchingColumn = parameters.getLookupField();
try {
@@ -115,7 +115,7 @@ public class CsvMetadataEnrichment implements EventProcessor<CsvMetadataEnrichme
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
this.columnMap = new HashMap<>();
}
}
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/task/TaskDuration.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/task/TaskDuration.java
index 10dd3b5..fa58cab 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/task/TaskDuration.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/task/TaskDuration.java
@@ -33,7 +33,7 @@ public class TaskDuration implements EventProcessor<TaskDurationParameters> {
private Double outputDivisor;
@Override
- public void onInvocation(TaskDurationParameters parameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(TaskDurationParameters parameters, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) throws SpRuntimeException {
this.taskFieldSelector = parameters.getTaskFieldSelector();
this.timestampFieldSelector = parameters.getTimestampFieldSelector();
this.outputDivisor = parameters.getOutputDivisor();
@@ -72,7 +72,7 @@ public class TaskDuration implements EventProcessor<TaskDurationParameters> {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
this.lastValue = null;
}
}
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/timestampextractor/TimestampExtractor.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/timestampextractor/TimestampExtractor.java
index 91016fd..8616d38 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/timestampextractor/TimestampExtractor.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/timestampextractor/TimestampExtractor.java
@@ -36,7 +36,7 @@ public class TimestampExtractor implements EventProcessor<TimestampExtractorPara
private List<String> outputFields;
@Override
- public void onInvocation(TimestampExtractorParameters params, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(TimestampExtractorParameters params, SpOutputCollector spOutputCollector, EventProcessorRuntimeContext runtimeContext) {
LOG = params.getGraph().getLogger(TimestampExtractor.class);
this.timestampField = params.getTimestampField();
@@ -97,6 +97,6 @@ public class TimestampExtractor implements EventProcessor<TimestampExtractorPara
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/transformtoboolean/TransformToBoolean.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/transformtoboolean/TransformToBoolean.java
index 34ccaaa..0193054 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/transformtoboolean/TransformToBoolean.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/transformtoboolean/TransformToBoolean.java
@@ -34,9 +34,9 @@ public class TransformToBoolean implements EventProcessor<TransformToBooleanPara
private List<String> transformFields;
@Override
- public void onInvocation(TransformToBooleanParameters transformToBooleanParameters,
- SpOutputCollector spOutputCollector,
- EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(TransformToBooleanParameters transformToBooleanParameters,
+ SpOutputCollector spOutputCollector,
+ EventProcessorRuntimeContext runtimeContext) {
LOG = transformToBooleanParameters.getGraph().getLogger(TransformToBoolean.class);
this.transformFields = transformToBooleanParameters.getTransformFields();
@@ -62,7 +62,7 @@ public class TransformToBoolean implements EventProcessor<TransformToBooleanPara
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
private Boolean toBoolean(Object value) throws SpRuntimeException {
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/value/change/ChangedValueDetection.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/value/change/ChangedValueDetection.java
index 5fc71e9..f2a27f4 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/value/change/ChangedValueDetection.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/value/change/ChangedValueDetection.java
@@ -33,9 +33,9 @@ public class ChangedValueDetection implements EventProcessor<ChangedValueDetecti
private Object lastObject = null;
@Override
- public void onInvocation(ChangedValueDetectionParameters changedValueDetectionParameters,
- SpOutputCollector spOutputCollector,
- EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(ChangedValueDetectionParameters changedValueDetectionParameters,
+ SpOutputCollector spOutputCollector,
+ EventProcessorRuntimeContext runtimeContext) {
LOG = changedValueDetectionParameters.getGraph().getLogger(ChangedValueDetection.class);
this.compareParameter = changedValueDetectionParameters.getCompareField();
this.changeFieldName = changedValueDetectionParameters.getChangeFieldName();
@@ -56,6 +56,6 @@ public class ChangedValueDetection implements EventProcessor<ChangedValueDetecti
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/value/duration/CalculateDuration.java b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/value/duration/CalculateDuration.java
index cd1b442..9375f21 100644
--- a/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/value/duration/CalculateDuration.java
+++ b/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/processor/value/duration/CalculateDuration.java
@@ -34,9 +34,9 @@ public class CalculateDuration implements EventProcessor<CalculateDurationParame
private String durationName;
@Override
- public void onInvocation(CalculateDurationParameters calculateDurationParameters,
- SpOutputCollector spOutputCollector,
- EventProcessorRuntimeContext runtimeContext) {
+ public void onPipelineStarted(CalculateDurationParameters calculateDurationParameters,
+ SpOutputCollector spOutputCollector,
+ EventProcessorRuntimeContext runtimeContext) {
LOG = calculateDurationParameters.getGraph().getLogger(
CalculateDuration.class);
@@ -66,6 +66,6 @@ public class CalculateDuration implements EventProcessor<CalculateDurationParame
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
}
}
diff --git a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/bufferrest/BufferRest.java b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/bufferrest/BufferRest.java
index caa479f..4962d43 100644
--- a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/bufferrest/BufferRest.java
+++ b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/bufferrest/BufferRest.java
@@ -50,7 +50,7 @@ public class BufferRest implements EventSink<BufferRestParameters>, BufferListen
}
@Override
- public void onInvocation(BufferRestParameters parameters, EventSinkRuntimeContext runtimeContext) {
+ public void onPipelineStarted(BufferRestParameters parameters, EventSinkRuntimeContext runtimeContext) {
this.fieldsToSend = parameters.getFieldsToSend();
this.restEndpointURI = parameters.getRestEndpointURI();
this.buffer = new MessageBuffer(parameters.getBufferSize());
@@ -69,7 +69,7 @@ public class BufferRest implements EventSink<BufferRestParameters>, BufferListen
}
@Override
- public void onDetach() {
+ public void onPipelineStopped() {
buffer.removeListener(this);
}
diff --git a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/jms/JmsPublisher.java b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/jms/JmsPublisher.java
index baf4539..50d505d 100644
--- a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/jms/JmsPublisher.java
+++ b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/jms/JmsPublisher.java
@@ -37,7 +37,7 @@ public class JmsPublisher implements EventSink<JmsParameters> {
}
@Override
- public void onInvocation(JmsParameters params, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(JmsParameters params, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
this.publisher = new ActiveMQPublisher(params.getJmsHost() + ":" + params.getJmsPort(), params.getTopic());
if (!this.publisher.isConnected()) {
throw new SpRuntimeException("Could not connect to JMS server " + params.getJmsHost() + " on Port: " + params.getJmsPort() + " to topic: " + params.getTopic());
@@ -58,7 +58,7 @@ public class JmsPublisher implements EventSink<JmsParameters> {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
this.publisher.disconnect();
}
}
diff --git a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/kafka/KafkaPublisher.java b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/kafka/KafkaPublisher.java
index 430e0a8..5866f31 100644
--- a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/kafka/KafkaPublisher.java
+++ b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/kafka/KafkaPublisher.java
@@ -37,7 +37,7 @@ public class KafkaPublisher implements EventSink<KafkaParameters> {
}
@Override
- public void onInvocation(KafkaParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(KafkaParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
this.producer = new SpKafkaProducer(parameters.getKafkaHost() + ":" + parameters.getKafkaPort(), parameters
.getTopic());
}
@@ -53,7 +53,7 @@ public class KafkaPublisher implements EventSink<KafkaParameters> {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
this.producer.disconnect();
}
}
diff --git a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/pulsar/Pulsar.java b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/pulsar/Pulsar.java
index c51ce93..ac4b860 100644
--- a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/pulsar/Pulsar.java
+++ b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/pulsar/Pulsar.java
@@ -43,8 +43,8 @@ public class Pulsar implements EventSink<PulsarParameters> {
}
@Override
- public void onInvocation(PulsarParameters params,
- EventSinkRuntimeContext eventSinkRuntimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(PulsarParameters params,
+ EventSinkRuntimeContext eventSinkRuntimeContext) throws SpRuntimeException {
try {
this.pulsarClient = PulsarClient.builder()
.serviceUrl(makePulsarUrl(params.getPulsarHost(), params.getPulsarPort()))
@@ -71,7 +71,7 @@ public class Pulsar implements EventSink<PulsarParameters> {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
try {
this.pulsarClient.close();
} catch (PulsarClientException e) {
diff --git a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/rabbitmq/RabbitMqConsumer.java b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/rabbitmq/RabbitMqConsumer.java
index a1ef054..16c9afe 100644
--- a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/rabbitmq/RabbitMqConsumer.java
+++ b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/rabbitmq/RabbitMqConsumer.java
@@ -42,7 +42,7 @@ public class RabbitMqConsumer implements EventSink<RabbitMqParameters> {
}
@Override
- public void onInvocation(RabbitMqParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(RabbitMqParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
this.publisher = new RabbitMqPublisher(parameters);
this.topic = parameters.getRabbitMqTopic();
}
@@ -59,7 +59,7 @@ public class RabbitMqConsumer implements EventSink<RabbitMqParameters> {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
publisher.cleanup();
}
}
diff --git a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/rest/RestPublisher.java b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/rest/RestPublisher.java
index 49a026f..c58ece2 100644
--- a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/rest/RestPublisher.java
+++ b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/rest/RestPublisher.java
@@ -36,7 +36,7 @@ public class RestPublisher implements EventSink<RestParameters> {
private JsonDataFormatDefinition jsonDataFormatDefinition;
@Override
- public void onInvocation(RestParameters params, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(RestParameters params, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
this.url = params.getUrl();
logger = params.getGraph().getLogger(RestPublisher.class);
jsonDataFormatDefinition = new JsonDataFormatDefinition();
@@ -65,6 +65,6 @@ public class RestPublisher implements EventSink<RestParameters> {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
}
}
diff --git a/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDb.java b/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDb.java
index eae1d3a..8214a9a 100644
--- a/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDb.java
+++ b/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/couchdb/CouchDb.java
@@ -31,7 +31,7 @@ public class CouchDb implements EventSink<CouchDbParameters> {
private CouchDbClient couchDbClient;
@Override
- public void onInvocation(CouchDbParameters parameters, EventSinkRuntimeContext runtimeContext) throws
+ public void onPipelineStarted(CouchDbParameters parameters, EventSinkRuntimeContext runtimeContext) throws
SpRuntimeException {
this.couchDbClient = new CouchDbClient(new CouchDbProperties(
parameters.getDatabaseName(),
@@ -50,7 +50,7 @@ public class CouchDb implements EventSink<CouchDbParameters> {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
this.couchDbClient.shutdown();
}
diff --git a/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/ditto/Ditto.java b/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/ditto/Ditto.java
index 8213660..56fffff 100644
--- a/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/ditto/Ditto.java
+++ b/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/ditto/Ditto.java
@@ -49,7 +49,7 @@ public class Ditto implements EventSink<DittoParameters> {
private List<String> selectedFields;
@Override
- public void onInvocation(DittoParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(DittoParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
this.thingId = parameters.getThingId();
this.featureId = parameters.getFeatureId();
@@ -112,7 +112,7 @@ public class Ditto implements EventSink<DittoParameters> {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
this.client.destroy();
}
}
diff --git a/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/influxdb/InfluxDb.java b/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/influxdb/InfluxDb.java
index 4eca4d7..a64bc8f 100644
--- a/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/influxdb/InfluxDb.java
+++ b/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/influxdb/InfluxDb.java
@@ -31,7 +31,7 @@ public class InfluxDb implements EventSink<InfluxDbParameters> {
private static Logger LOG;
@Override
- public void onInvocation(InfluxDbParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(InfluxDbParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
LOG = parameters.getGraph().getLogger(InfluxDb.class);
this.influxDbClient = new InfluxDbClient(
@@ -58,7 +58,7 @@ public class InfluxDb implements EventSink<InfluxDbParameters> {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
influxDbClient.stop();
}
}
diff --git a/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDb.java b/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDb.java
index 177270a..6ce89b0 100644
--- a/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDb.java
+++ b/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/iotdb/IotDb.java
@@ -38,7 +38,7 @@ public class IotDb extends JdbcClient implements EventSink<IotDbParameters> {
private String timestampField;
@Override
- public void onInvocation(IotDbParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(IotDbParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
LOG = parameters.getGraph().getLogger(IotDb.class);
timestampField = parameters.getTimestampField();
@@ -75,7 +75,7 @@ public class IotDb extends JdbcClient implements EventSink<IotDbParameters> {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
closeAll();
}
diff --git a/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/OpcUa.java b/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/OpcUa.java
index 871a6fb..ad66e6b 100644
--- a/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/OpcUa.java
+++ b/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/opcua/OpcUa.java
@@ -46,7 +46,7 @@ public class OpcUa implements EventSink<OpcUaParameters> {
private NodeId node;
@Override
- public void onInvocation(OpcUaParameters parameters, EventSinkRuntimeContext runtimeContext) throws
+ public void onPipelineStarted(OpcUaParameters parameters, EventSinkRuntimeContext runtimeContext) throws
SpRuntimeException {
LOG = parameters.getGraph().getLogger(OpcUa.class);
@@ -119,7 +119,7 @@ public class OpcUa implements EventSink<OpcUaParameters> {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
opcUaClient.disconnect();
}
diff --git a/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSql.java b/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSql.java
index 68ae54d..3049430 100644
--- a/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSql.java
+++ b/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/postgresql/PostgreSql.java
@@ -30,7 +30,7 @@ public class PostgreSql extends JdbcClient implements EventSink<PostgreSqlParame
private static Logger LOG;
@Override
- public void onInvocation(PostgreSqlParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(PostgreSqlParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
LOG = parameters.getGraph().getLogger(PostgreSql.class);
// get(0) because it is the only input stream of the sink (and not two)
@@ -62,7 +62,7 @@ public class PostgreSql extends JdbcClient implements EventSink<PostgreSqlParame
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
closeAll();
}
}
diff --git a/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/dashboard/Dashboard.java b/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/dashboard/Dashboard.java
index 618b251..edaa7dc 100644
--- a/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/dashboard/Dashboard.java
+++ b/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/dashboard/Dashboard.java
@@ -51,7 +51,7 @@ public class Dashboard implements EventSink<DashboardParameters> {
}
@Override
- public void onInvocation(DashboardParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(DashboardParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
if (!saveToCouchDB(parameters.getGraph(), parameters)) {
throw new SpRuntimeException("The schema couldn't be stored in the couchDB");
}
@@ -89,7 +89,7 @@ public class Dashboard implements EventSink<DashboardParameters> {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
this.publisher.disconnect();
if (!removeFromCouchDB()) {
throw new SpRuntimeException("There was an error while deleting pipeline: '" + pipelineId + "'");
diff --git a/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLake.java b/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLake.java
index a3254eb..51c01a2 100644
--- a/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLake.java
+++ b/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/datalake/DataLake.java
@@ -57,7 +57,7 @@ public class DataLake implements EventSink<DataLakeParameters> {
private String timestampField;
@Override
- public void onInvocation(DataLakeParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(DataLakeParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
LOG = parameters.getGraph().getLogger(DataLake.class);
this.timestampField = parameters.getTimestampField();
@@ -108,7 +108,7 @@ public class DataLake implements EventSink<DataLakeParameters> {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
influxDbClient.stop();
}
diff --git a/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/notification/NotificationProducer.java b/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/notification/NotificationProducer.java
index 9057a6d..73cb195 100644
--- a/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/notification/NotificationProducer.java
+++ b/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/notification/NotificationProducer.java
@@ -40,7 +40,7 @@ public class NotificationProducer implements EventSink<NotificationParameters> {
@Override
- public void onInvocation(NotificationParameters parameters, EventSinkRuntimeContext runtimeContext) throws
+ public void onPipelineStarted(NotificationParameters parameters, EventSinkRuntimeContext runtimeContext) throws
SpRuntimeException {
this.publisher = new ActiveMQPublisher(SinksInternalJvmConfig.INSTANCE.getJmsHost() + ":" + SinksInternalJvmConfig.INSTANCE.getJmsPort(),
"org.apache.streampipes.notifications");
@@ -62,7 +62,7 @@ public class NotificationProducer implements EventSink<NotificationParameters> {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
this.publisher.disconnect();
}
}
diff --git a/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/email/EmailPublisher.java b/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/email/EmailPublisher.java
index 7718689..4453d52 100644
--- a/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/email/EmailPublisher.java
+++ b/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/email/EmailPublisher.java
@@ -46,7 +46,7 @@ public class EmailPublisher implements EventSink<EmailParameters> {
private String content;
@Override
- public void onInvocation(EmailParameters parameters, EventSinkRuntimeContext runtimeContext) {
+ public void onPipelineStarted(EmailParameters parameters, EventSinkRuntimeContext runtimeContext) {
LOG = parameters.getGraph().getLogger(EmailPublisher.class);
String from = SinksNotificationsJvmConfig.INSTANCE.getEmailFrom();
@@ -108,6 +108,6 @@ public class EmailPublisher implements EventSink<EmailParameters> {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
}
}
diff --git a/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/onesignal/OneSignalProducer.java b/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/onesignal/OneSignalProducer.java
index c8ce665..4dc99a2 100644
--- a/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/onesignal/OneSignalProducer.java
+++ b/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/onesignal/OneSignalProducer.java
@@ -39,7 +39,7 @@ public class OneSignalProducer implements EventSink<OneSignalParameters> {
private String apiKey;
@Override
- public void onInvocation(OneSignalParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(OneSignalParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
this.content = parameters.getContent();
this.appId = parameters.getAppId();
this.apiKey = parameters.getApiKey();
@@ -75,6 +75,6 @@ public class OneSignalProducer implements EventSink<OneSignalParameters> {
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
}
}
diff --git a/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/slack/SlackNotification.java b/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/slack/SlackNotification.java
index 2ddd7ae..35e8766 100644
--- a/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/slack/SlackNotification.java
+++ b/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/slack/SlackNotification.java
@@ -31,7 +31,7 @@ public class SlackNotification implements EventSink<SlackNotificationParameters>
private SlackNotificationParameters params;
@Override
- public void onInvocation(SlackNotificationParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
+ public void onPipelineStarted(SlackNotificationParameters parameters, EventSinkRuntimeContext runtimeContext) throws SpRuntimeException {
this.params = parameters;
}
@@ -47,7 +47,7 @@ public class SlackNotification implements EventSink<SlackNotificationParameters>
}
@Override
- public void onDetach() throws SpRuntimeException {
+ public void onPipelineStopped() throws SpRuntimeException {
try {
params.getSession().disconnect();
} catch (IOException e) {