This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch
1620-migrate-data-sinks-in-notifications-jvm-module
in repository https://gitbox.apache.org/repos/asf/streampipes.git
The following commit(s) were added to
refs/heads/1620-migrate-data-sinks-in-notifications-jvm-module by this push:
new 0ce61bfa3 Migrate data sinks in notifications-jvm module (#1620)
0ce61bfa3 is described below
commit 0ce61bfa3a027a44c40c3fb07aa4a9f9651ab1fd
Author: Dominik Riemer <[email protected]>
AuthorDate: Fri May 26 23:14:40 2023 +0200
Migrate data sinks in notifications-jvm module (#1620)
---
.../jvm/SinksNotificationsJvmInit.java | 16 +--
.../notifications/jvm/email/EmailParameters.java | 58 ---------
.../notifications/jvm/email/EmailPublisher.java | 74 -----------
.../email/{EmailController.java => EmailSink.java} | 62 +++++++--
.../jvm/onesignal/OneSignalController.java | 69 ----------
.../jvm/onesignal/OneSignalParameters.java | 49 -------
.../{OneSignalProducer.java => OneSignalSink.java} | 47 +++++--
.../notifications/jvm/slack/SlackNotification.java | 99 --------------
.../jvm/slack/SlackNotificationController.java | 76 -----------
.../jvm/slack/SlackNotificationParameters.java | 58 ---------
.../jvm/slack/SlackNotificationSink.java | 142 +++++++++++++++++++++
.../jvm/telegram/TelegramController.java | 65 ----------
.../jvm/telegram/TelegramParameters.java | 52 --------
.../{TelegramPublisher.java => TelegramSink.java} | 52 ++++++--
14 files changed, 281 insertions(+), 638 deletions(-)
diff --git
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/SinksNotificationsJvmInit.java
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/SinksNotificationsJvmInit.java
index 69b35d403..1caeaa7a0 100644
---
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/SinksNotificationsJvmInit.java
+++
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/SinksNotificationsJvmInit.java
@@ -29,10 +29,10 @@ import
org.apache.streampipes.messaging.kafka.SpKafkaProtocolFactory;
import org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory;
import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter;
import org.apache.streampipes.sinks.notifications.jvm.config.ConfigKeys;
-import org.apache.streampipes.sinks.notifications.jvm.email.EmailController;
-import
org.apache.streampipes.sinks.notifications.jvm.onesignal.OneSignalController;
-import
org.apache.streampipes.sinks.notifications.jvm.slack.SlackNotificationController;
-import
org.apache.streampipes.sinks.notifications.jvm.telegram.TelegramController;
+import org.apache.streampipes.sinks.notifications.jvm.email.EmailSink;
+import org.apache.streampipes.sinks.notifications.jvm.onesignal.OneSignalSink;
+import
org.apache.streampipes.sinks.notifications.jvm.slack.SlackNotificationSink;
+import org.apache.streampipes.sinks.notifications.jvm.telegram.TelegramSink;
public class SinksNotificationsJvmInit extends ExtensionsModelSubmitter {
@@ -43,10 +43,10 @@ public class SinksNotificationsJvmInit extends
ExtensionsModelSubmitter {
"",
8090)
.registerPipelineElements(
- new EmailController(),
- new OneSignalController(),
- new SlackNotificationController(),
- new TelegramController())
+ new EmailSink(),
+ new OneSignalSink(),
+ new SlackNotificationSink(),
+ new TelegramSink())
.registerMessagingFormats(new JsonDataFormatFactory(),
new CborDataFormatFactory(),
new SmileDataFormatFactory(),
diff --git
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/email/EmailParameters.java
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/email/EmailParameters.java
deleted file mode 100644
index 123e68670..000000000
---
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/email/EmailParameters.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.streampipes.sinks.notifications.jvm.email;
-
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.wrapper.params.binding.EventSinkBindingParams;
-
-public class EmailParameters extends EventSinkBindingParams {
-
- private String toEmailAddress;
- private String subject;
- private String content;
- private Integer silentPeriod;
-
- public EmailParameters(DataSinkInvocation graph,
- String toEmailAddress,
- String subject,
- String content,
- Integer silentPeriod) {
- super(graph);
- this.toEmailAddress = toEmailAddress;
- this.subject = subject;
- this.content = content;
- this.silentPeriod = silentPeriod;
- }
-
- public String getToEmailAddress() {
- return toEmailAddress;
- }
-
- public String getSubject() {
- return subject;
- }
-
- public String getContent() {
- return content;
- }
-
- public Integer getSilentPeriod() {
- return silentPeriod;
- }
-}
diff --git
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/email/EmailPublisher.java
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/email/EmailPublisher.java
deleted file mode 100644
index 3ca8478f8..000000000
---
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/email/EmailPublisher.java
+++ /dev/null
@@ -1,74 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.streampipes.sinks.notifications.jvm.email;
-
-import org.apache.streampipes.client.StreamPipesClient;
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.model.mail.SpEmail;
-import org.apache.streampipes.model.runtime.Event;
-import org.apache.streampipes.pe.shared.PlaceholderExtractor;
-import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
-import org.apache.streampipes.wrapper.runtime.EventSink;
-
-import java.time.Instant;
-import java.util.Collections;
-
-public class EmailPublisher implements EventSink<EmailParameters> {
-
- private SpEmail preparedEmail;
- private long silentPeriodInSeconds;
- private long lastMailEpochSecond = -1;
-
- private String originalContent;
-
- private StreamPipesClient client;
-
- @Override
- public void onInvocation(EmailParameters parameters, EventSinkRuntimeContext
runtimeContext) {
- this.preparedEmail = new SpEmail();
-
this.preparedEmail.setRecipients(Collections.singletonList(parameters.getToEmailAddress()));
- this.preparedEmail.setSubject(parameters.getSubject());
-
- this.silentPeriodInSeconds = parameters.getSilentPeriod() * 60;
- this.client = runtimeContext.getStreamPipesClient();
- this.originalContent = parameters.getContent();
- }
-
- @Override
- public void onEvent(Event inputEvent) {
- if (shouldSendMail()) {
- String message = PlaceholderExtractor.replacePlaceholders(inputEvent,
this.originalContent);
- this.preparedEmail.setMessage(message);
- this.client.deliverEmail(this.preparedEmail);
- this.lastMailEpochSecond = Instant.now().getEpochSecond();
- }
- }
-
- private boolean shouldSendMail() {
- if (this.lastMailEpochSecond == -1) {
- return true;
- } else {
- return Instant.now().getEpochSecond() >= (this.lastMailEpochSecond +
this.silentPeriodInSeconds);
- }
- }
-
- @Override
- public void onDetach() throws SpRuntimeException {
- }
-}
diff --git
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/email/EmailController.java
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/email/EmailSink.java
similarity index 56%
rename from
streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/email/EmailController.java
rename to
streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/email/EmailSink.java
index 7c4251cef..46ed25e6a 100644
---
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/email/EmailController.java
+++
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/email/EmailSink.java
@@ -18,26 +18,40 @@
package org.apache.streampipes.sinks.notifications.jvm.email;
+import org.apache.streampipes.client.StreamPipesClient;
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
import org.apache.streampipes.model.DataSinkType;
import org.apache.streampipes.model.graph.DataSinkDescription;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.mail.SpEmail;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.pe.shared.PlaceholderExtractor;
import org.apache.streampipes.sdk.builder.DataSinkBuilder;
import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
-import org.apache.streampipes.sdk.extractor.DataSinkParameterExtractor;
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.standalone.ConfiguredEventSink;
-import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventSinkDeclarer;
+import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
+import org.apache.streampipes.wrapper.standalone.SinkParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataSink;
-public class EmailController extends
StandaloneEventSinkDeclarer<EmailParameters> {
+import java.time.Instant;
+import java.util.Collections;
+
+public class EmailSink extends StreamPipesDataSink {
private static final String TO_EMAIL_ADRESS = "to_email";
private static final String EMAIL_SUBJECT = "email_subject";
private static final String EMAIL_CONTENT = "email_content";
private static final String SILENT_PERIOD = "silent-period";
+ private SpEmail preparedEmail;
+ private long silentPeriodInSeconds;
+ private long lastMailEpochSecond = -1;
+
+ private String originalContent;
+
+ private StreamPipesClient client;
@Override
public DataSinkDescription declareModel() {
@@ -57,17 +71,43 @@ public class EmailController extends
StandaloneEventSinkDeclarer<EmailParameters
}
@Override
- public ConfiguredEventSink<EmailParameters> onInvocation(DataSinkInvocation
graph,
-
DataSinkParameterExtractor extractor) {
-
+ public void onInvocation(SinkParams parameters,
+ EventSinkRuntimeContext runtimeContext) throws
SpRuntimeException {
+ var extractor = parameters.extractor();
String toEmail = extractor.singleValueParameter(TO_EMAIL_ADRESS,
String.class);
String subject = extractor.singleValueParameter(EMAIL_SUBJECT,
String.class);
- String content = extractor.singleValueParameter(EMAIL_CONTENT,
String.class);
Integer silentPeriod = extractor.singleValueParameter(SILENT_PERIOD,
Integer.class);
- EmailParameters params = new EmailParameters(graph, toEmail, subject,
content, silentPeriod);
+ this.preparedEmail = new SpEmail();
+ this.preparedEmail.setRecipients(Collections.singletonList(toEmail));
+ this.preparedEmail.setSubject(subject);
+
+ this.silentPeriodInSeconds = silentPeriod * 60;
+ this.client = runtimeContext.getStreamPipesClient();
+ this.originalContent = extractor.singleValueParameter(EMAIL_CONTENT,
String.class);
+ }
+
+ @Override
+ public void onEvent(Event event) throws SpRuntimeException {
+ if (shouldSendMail()) {
+ String message = PlaceholderExtractor.replacePlaceholders(event,
this.originalContent);
+ this.preparedEmail.setMessage(message);
+ this.client.deliverEmail(this.preparedEmail);
+ this.lastMailEpochSecond = Instant.now().getEpochSecond();
+ }
+ }
+
+ @Override
+ public void onDetach() throws SpRuntimeException {
+
+ }
- return new ConfiguredEventSink<>(params, EmailPublisher::new);
+ private boolean shouldSendMail() {
+ if (this.lastMailEpochSecond == -1) {
+ return true;
+ } else {
+ return Instant.now().getEpochSecond() >= (this.lastMailEpochSecond +
this.silentPeriodInSeconds);
+ }
}
}
diff --git
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/onesignal/OneSignalController.java
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/onesignal/OneSignalController.java
deleted file mode 100644
index 30e8d0df1..000000000
---
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/onesignal/OneSignalController.java
+++ /dev/null
@@ -1,69 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.streampipes.sinks.notifications.jvm.onesignal;
-
-import org.apache.streampipes.model.DataSinkType;
-import org.apache.streampipes.model.graph.DataSinkDescription;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.sdk.builder.DataSinkBuilder;
-import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
-import org.apache.streampipes.sdk.extractor.DataSinkParameterExtractor;
-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.standalone.ConfiguredEventSink;
-import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventSinkDeclarer;
-
-public class OneSignalController extends
StandaloneEventSinkDeclarer<OneSignalParameters> {
-
- private static final String CONTENT_KEY = "content";
- private static final String APP_ID = "app_id";
- private static final String REST_API_KEY = "api_key";
-
- @Override
- public DataSinkDescription declareModel() {
- return
DataSinkBuilder.create("org.apache.streampipes.sinks.notifications.jvm.onesignal")
- .withLocales(Locales.EN)
- .withAssets(Assets.DOCUMENTATION, Assets.ICON)
- .category(DataSinkType.NOTIFICATION)
- .requiredStream(StreamRequirementsBuilder
- .create()
- .requiredProperty(EpRequirements.anyProperty())
- .build())
- .requiredHtmlInputParameter(Labels.withId(CONTENT_KEY))
- .requiredTextParameter(Labels.withId(APP_ID))
- .requiredTextParameter(Labels.withId(REST_API_KEY))
- .build();
- }
-
- @Override
- public ConfiguredEventSink<OneSignalParameters>
onInvocation(DataSinkInvocation graph,
-
DataSinkParameterExtractor extractor) {
-
- String content = extractor.singleValueParameter(CONTENT_KEY, String.class);
- String appId = extractor.singleValueParameter(APP_ID, String.class);
- String apiKey = extractor.singleValueParameter(REST_API_KEY, String.class);
-
- OneSignalParameters params = new OneSignalParameters(graph, content,
appId, apiKey);
-
- return new ConfiguredEventSink<>(params, OneSignalProducer::new);
- }
-
-}
diff --git
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/onesignal/OneSignalParameters.java
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/onesignal/OneSignalParameters.java
deleted file mode 100644
index 9ef4c11ee..000000000
---
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/onesignal/OneSignalParameters.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.streampipes.sinks.notifications.jvm.onesignal;
-
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.wrapper.params.binding.EventSinkBindingParams;
-
-public class OneSignalParameters extends EventSinkBindingParams {
-
- private String title;
- private String content;
- private String appId;
- private String apiKey;
-
- public OneSignalParameters(DataSinkInvocation graph, String content, String
appId, String apiKey) {
- super(graph);
- this.content = content;
- this.appId = appId;
- this.apiKey = apiKey;
- }
-
- public String getContent() {
- return content;
- }
-
- public String getAppId() {
- return appId;
- }
-
- public String getApiKey() {
- return apiKey;
- }
-}
diff --git
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/onesignal/OneSignalProducer.java
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/onesignal/OneSignalSink.java
similarity index 56%
rename from
streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/onesignal/OneSignalProducer.java
rename to
streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/onesignal/OneSignalSink.java
index da4cb01d6..7481ce39c 100644
---
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/onesignal/OneSignalProducer.java
+++
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/onesignal/OneSignalSink.java
@@ -19,9 +19,18 @@
package org.apache.streampipes.sinks.notifications.jvm.onesignal;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.DataSinkType;
+import org.apache.streampipes.model.graph.DataSinkDescription;
import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.sdk.builder.DataSinkBuilder;
+import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+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.runtime.EventSink;
+import org.apache.streampipes.wrapper.standalone.SinkParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataSink;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
@@ -33,23 +42,43 @@ import org.apache.http.impl.client.HttpClients;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
-public class OneSignalProducer implements EventSink<OneSignalParameters> {
+public class OneSignalSink extends StreamPipesDataSink {
+
+ private static final String CONTENT_KEY = "content";
+ private static final String APP_ID = "app_id";
+ private static final String REST_API_KEY = "api_key";
private String content;
private String appId;
private String apiKey;
@Override
- public void onInvocation(OneSignalParameters parameters,
EventSinkRuntimeContext runtimeContext)
- throws SpRuntimeException {
- this.content = parameters.getContent();
- this.appId = parameters.getAppId();
- this.apiKey = parameters.getApiKey();
+ public DataSinkDescription declareModel() {
+ return
DataSinkBuilder.create("org.apache.streampipes.sinks.notifications.jvm.onesignal")
+ .withLocales(Locales.EN)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .category(DataSinkType.NOTIFICATION)
+ .requiredStream(StreamRequirementsBuilder
+ .create()
+ .requiredProperty(EpRequirements.anyProperty())
+ .build())
+ .requiredHtmlInputParameter(Labels.withId(CONTENT_KEY))
+ .requiredTextParameter(Labels.withId(APP_ID))
+ .requiredTextParameter(Labels.withId(REST_API_KEY))
+ .build();
}
@Override
- public void onEvent(Event inputEvent) {
+ public void onInvocation(SinkParams parameters,
+ EventSinkRuntimeContext runtimeContext) throws
SpRuntimeException {
+ var extractor = parameters.extractor();
+ content = extractor.singleValueParameter(CONTENT_KEY, String.class);
+ appId = extractor.singleValueParameter(APP_ID, String.class);
+ apiKey = extractor.singleValueParameter(REST_API_KEY, String.class);
+ }
+ @Override
+ public void onEvent(Event event) throws SpRuntimeException {
String jsondata =
"{\"app_id\": \"" + appId + "\",\"contents\": {\"en\": \"" + content +
"\"}, \"included_segments\":[\"All\"]}";
@@ -74,10 +103,10 @@ public class OneSignalProducer implements
EventSink<OneSignalParameters> {
e.printStackTrace();
}
HttpEntity entity = response.getEntity();
-
}
@Override
public void onDetach() throws SpRuntimeException {
+
}
}
diff --git
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/slack/SlackNotification.java
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/slack/SlackNotification.java
deleted file mode 100644
index bad7fa25e..000000000
---
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/slack/SlackNotification.java
+++ /dev/null
@@ -1,99 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.streampipes.sinks.notifications.jvm.slack;
-
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.model.runtime.Event;
-import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
-import org.apache.streampipes.wrapper.runtime.EventSink;
-
-import com.ullink.slack.simpleslackapi.SlackChannel;
-import com.ullink.slack.simpleslackapi.SlackSession;
-import com.ullink.slack.simpleslackapi.SlackUser;
-import com.ullink.slack.simpleslackapi.impl.SlackSessionFactory;
-
-import java.io.IOException;
-
-
-public class SlackNotification implements
EventSink<SlackNotificationParameters> {
-
- private static final String HASHTAG = "#";
-
- private SlackNotificationParameters params;
- private SlackSession session;
- private Boolean sendToUser;
-
- @Override
- public void onInvocation(SlackNotificationParameters parameters,
EventSinkRuntimeContext runtimeContext)
- throws SpRuntimeException {
- this.params = parameters;
- // Initialize slack session
- this.session =
SlackSessionFactory.createWebSocketSlackSession(params.getAuthToken());
- try {
- this.session.connect();
- } catch (IOException e) {
- e.printStackTrace();
- throw new SpRuntimeException("Could not connect to Slack");
- }
-
- this.sendToUser = params.getChannelType().equals("User");
-
- if (this.sendToUser) {
- SlackUser user = session.findUserByUserName(params.getUserChannel());
- if (user == null) {
- throw new SpRuntimeException("The user: '" + params.getUserChannel() +
"' does not exists");
- }
- } else {
- SlackChannel channel =
session.findChannelByName(params.getUserChannel());
- if (channel == null || channel.getId() == null) {
- throw new SpRuntimeException("The channel: '" +
params.getUserChannel() + "' does not "
- + "exists or "
- + "the bot has no rights to access it");
- }
- }
- }
-
- @Override
- public void onEvent(Event event) {
- String message = replacePlaceholders(event, params.getMessage());
- if (this.sendToUser) {
- this.session.sendMessageToUser(params.getUserChannel(),
- message, null);
- } else {
- SlackChannel channel =
this.session.findChannelByName(params.getUserChannel());
- this.session.sendMessage(channel, message);
- }
- }
-
- @Override
- public void onDetach() throws SpRuntimeException {
- try {
- this.session.disconnect();
- } catch (IOException e) {
- throw new SpRuntimeException("Could not disconnect");
- }
- }
-
- private String replacePlaceholders(Event event, String content) {
- for (String key : event.getRaw().keySet()) {
- content = content.replaceAll(HASHTAG + key + HASHTAG,
event.getRaw().get(key).toString());
- }
- return content;
- }
-}
diff --git
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/slack/SlackNotificationController.java
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/slack/SlackNotificationController.java
deleted file mode 100644
index c199afb9a..000000000
---
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/slack/SlackNotificationController.java
+++ /dev/null
@@ -1,76 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.streampipes.sinks.notifications.jvm.slack;
-
-import org.apache.streampipes.model.DataSinkType;
-import org.apache.streampipes.model.graph.DataSinkDescription;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.sdk.builder.DataSinkBuilder;
-import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
-import org.apache.streampipes.sdk.extractor.DataSinkParameterExtractor;
-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.Options;
-import org.apache.streampipes.sdk.utils.Assets;
-import org.apache.streampipes.wrapper.standalone.ConfiguredEventSink;
-import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventSinkDeclarer;
-
-public class SlackNotificationController extends
StandaloneEventSinkDeclarer<SlackNotificationParameters> {
-
- private static final String CHANNEL_TYPE = "channel-type";
- private static final String RECEIVER = "receiver";
- private static final String CONTENT = "content";
- private static final String AUTH_TOKEN = "auth-token";
-
- @Override
- public DataSinkDescription declareModel() {
-
- return
DataSinkBuilder.create("org.apache.streampipes.sinks.notifications.jvm.slack")
- .withLocales(Locales.EN)
- .withAssets(Assets.DOCUMENTATION, Assets.ICON)
- .category(DataSinkType.NOTIFICATION)
- .requiredStream(StreamRequirementsBuilder
- .create()
- .requiredProperty(EpRequirements.anyProperty())
- .build())
- .requiredTextParameter(Labels.withId(RECEIVER))
- .requiredTextParameter(Labels.withId(CONTENT), false, true)
- .requiredSingleValueSelection(Labels.withId(CHANNEL_TYPE),
- Options.from("User", "Channel"))
- .requiredSecret(Labels.withId(AUTH_TOKEN))
- .build();
- }
-
-
- @Override
- public ConfiguredEventSink<SlackNotificationParameters>
onInvocation(DataSinkInvocation graph,
-
DataSinkParameterExtractor extractor) {
-
- String userChannel = extractor.singleValueParameter(RECEIVER,
String.class);
- String channelType = extractor.selectedSingleValue(CHANNEL_TYPE,
String.class);
- String message = extractor.singleValueParameter(CONTENT, String.class);
- String authToken = extractor.secretValue(AUTH_TOKEN);
-
- SlackNotificationParameters params = new
SlackNotificationParameters(graph, authToken,
- channelType, userChannel, message);
-
- return new ConfiguredEventSink<>(params, SlackNotification::new);
- }
-}
diff --git
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/slack/SlackNotificationParameters.java
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/slack/SlackNotificationParameters.java
deleted file mode 100644
index 15bb645ee..000000000
---
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/slack/SlackNotificationParameters.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.streampipes.sinks.notifications.jvm.slack;
-
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.wrapper.params.binding.EventSinkBindingParams;
-
-public class SlackNotificationParameters extends EventSinkBindingParams {
- private String authToken;
- private String userChannel;
- private String channelType;
- private String message;
-
- public SlackNotificationParameters(DataSinkInvocation graph,
- String authToken,
- String channelType,
- String userChannel,
- String message) {
- super(graph);
- this.authToken = authToken;
- this.userChannel = userChannel;
- this.message = message;
- this.channelType = channelType;
- }
-
- public String getAuthToken() {
- return authToken;
- }
-
- public String getUserChannel() {
- return userChannel;
- }
-
- public String getMessage() {
- return message;
- }
-
- public String getChannelType() {
- return channelType;
- }
-}
-
diff --git
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/slack/SlackNotificationSink.java
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/slack/SlackNotificationSink.java
new file mode 100644
index 000000000..60bce1e06
--- /dev/null
+++
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/slack/SlackNotificationSink.java
@@ -0,0 +1,142 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.sinks.notifications.jvm.slack;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.DataSinkType;
+import org.apache.streampipes.model.graph.DataSinkDescription;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.sdk.builder.DataSinkBuilder;
+import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+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.Options;
+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.ullink.slack.simpleslackapi.SlackChannel;
+import com.ullink.slack.simpleslackapi.SlackSession;
+import com.ullink.slack.simpleslackapi.SlackUser;
+import com.ullink.slack.simpleslackapi.impl.SlackSessionFactory;
+
+import java.io.IOException;
+
+public class SlackNotificationSink extends StreamPipesDataSink {
+
+ private static final String CHANNEL_TYPE = "channel-type";
+ private static final String RECEIVER = "receiver";
+ private static final String CONTENT = "content";
+ private static final String AUTH_TOKEN = "auth-token";
+
+ private static final String HASHTAG = "#";
+
+ private SlackSession session;
+ private Boolean sendToUser;
+
+ private String userChannel;
+ private String originalMessage;
+
+ @Override
+ public DataSinkDescription declareModel() {
+
+ return
DataSinkBuilder.create("org.apache.streampipes.sinks.notifications.jvm.slack")
+ .withLocales(Locales.EN)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .category(DataSinkType.NOTIFICATION)
+ .requiredStream(StreamRequirementsBuilder
+ .create()
+ .requiredProperty(EpRequirements.anyProperty())
+ .build())
+ .requiredTextParameter(Labels.withId(RECEIVER))
+ .requiredTextParameter(Labels.withId(CONTENT), false, true)
+ .requiredSingleValueSelection(Labels.withId(CHANNEL_TYPE),
+ Options.from("User", "Channel"))
+ .requiredSecret(Labels.withId(AUTH_TOKEN))
+ .build();
+ }
+
+ @Override
+ public void onInvocation(SinkParams parameters,
+ EventSinkRuntimeContext runtimeContext) throws
SpRuntimeException {
+ var extractor = parameters.extractor();
+
+ userChannel = extractor.singleValueParameter(RECEIVER, String.class);
+ originalMessage = extractor.singleValueParameter(CONTENT, String.class);
+
+ String channelType = extractor.selectedSingleValue(CHANNEL_TYPE,
String.class);
+ String authToken = extractor.secretValue(AUTH_TOKEN);
+
+ // Initialize slack session
+ this.session = SlackSessionFactory.createWebSocketSlackSession(authToken);
+ try {
+ this.session.connect();
+ } catch (IOException e) {
+ e.printStackTrace();
+ throw new SpRuntimeException("Could not connect to Slack");
+ }
+
+ this.sendToUser = channelType.equals("User");
+
+ if (this.sendToUser) {
+ SlackUser user = session.findUserByUserName(userChannel);
+ if (user == null) {
+ throw new SpRuntimeException("The user: '" + userChannel + "' does not
exists");
+ }
+ } else {
+ SlackChannel channel = session.findChannelByName(userChannel);
+ if (channel == null || channel.getId() == null) {
+ throw new SpRuntimeException("The channel: '" + userChannel + "' does
not "
+ + "exists or "
+ + "the bot has no rights to access it");
+ }
+ }
+
+ }
+
+ @Override
+ public void onEvent(Event event) throws SpRuntimeException {
+ String message = replacePlaceholders(event, originalMessage);
+ if (this.sendToUser) {
+ this.session.sendMessageToUser(userChannel,
+ message, null);
+ } else {
+ SlackChannel channel = this.session.findChannelByName(userChannel);
+ this.session.sendMessage(channel, message);
+ }
+ }
+
+ @Override
+ public void onDetach() throws SpRuntimeException {
+ try {
+ this.session.disconnect();
+ } catch (IOException e) {
+ throw new SpRuntimeException("Could not disconnect");
+ }
+ }
+
+ private String replacePlaceholders(Event event, String content) {
+ for (String key : event.getRaw().keySet()) {
+ content = content.replaceAll(HASHTAG + key + HASHTAG,
event.getRaw().get(key).toString());
+ }
+ return content;
+ }
+}
diff --git
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/telegram/TelegramController.java
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/telegram/TelegramController.java
deleted file mode 100644
index 0a84fd4bb..000000000
---
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/telegram/TelegramController.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.streampipes.sinks.notifications.jvm.telegram;
-
-import org.apache.streampipes.model.DataSinkType;
-import org.apache.streampipes.model.graph.DataSinkDescription;
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.sdk.builder.DataSinkBuilder;
-import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
-import org.apache.streampipes.sdk.extractor.DataSinkParameterExtractor;
-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.standalone.ConfiguredEventSink;
-import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventSinkDeclarer;
-
-public class TelegramController extends
StandaloneEventSinkDeclarer<TelegramParameters> {
- private static final String CHANNEL_NAME_OR_CHAT_ID = "channel-chat-name";
- private static final String MESSAGE_TEXT = "message-text";
- private static final String BOT_API_KEY = "api-key";
-
- @Override
- public DataSinkDescription declareModel() {
- return
DataSinkBuilder.create("org.apache.streampipes.sinks.notifications.jvm.telegram")
- .withLocales(Locales.EN)
- .withAssets(Assets.DOCUMENTATION, Assets.ICON)
- .category(DataSinkType.NOTIFICATION)
- .requiredStream(StreamRequirementsBuilder
- .create()
- .requiredProperty(EpRequirements.anyProperty())
- .build())
- .requiredSecret(Labels.withId(BOT_API_KEY))
- .requiredTextParameter(Labels.withId(CHANNEL_NAME_OR_CHAT_ID))
- .requiredTextParameter(Labels.withId(MESSAGE_TEXT), true, true, true)
- .build();
- }
-
-
- @Override
- public ConfiguredEventSink<TelegramParameters>
onInvocation(DataSinkInvocation graph,
-
DataSinkParameterExtractor extractor) {
- String apiKey = extractor.secretValue(BOT_API_KEY);
- String channelOrChatId =
extractor.singleValueParameter(CHANNEL_NAME_OR_CHAT_ID, String.class);
- String message = extractor.singleValueParameter(MESSAGE_TEXT,
String.class);
- TelegramParameters params = new TelegramParameters(graph, apiKey,
channelOrChatId, message);
- return new ConfiguredEventSink<>(params, TelegramPublisher::new);
- }
-}
diff --git
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/telegram/TelegramParameters.java
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/telegram/TelegramParameters.java
deleted file mode 100644
index c92caf004..000000000
---
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/telegram/TelegramParameters.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- *
- */
-
-package org.apache.streampipes.sinks.notifications.jvm.telegram;
-
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.wrapper.params.binding.EventSinkBindingParams;
-
-public class TelegramParameters extends EventSinkBindingParams {
- private String apiKey;
- private String channelOrChatId;
- private String message;
-
- public TelegramParameters(DataSinkInvocation graph,
- String apiKey,
- String channelOrChatId,
- String message) {
- super(graph);
- this.apiKey = apiKey;
- this.channelOrChatId = channelOrChatId;
- this.message = message;
- }
-
- public String getApiKey() {
- return apiKey;
- }
-
- public String getChannelOrChatId() {
- return channelOrChatId;
- }
-
- public String getMessage() {
- return message;
- }
-
-}
-
diff --git
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/telegram/TelegramPublisher.java
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/telegram/TelegramSink.java
similarity index 62%
rename from
streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/telegram/TelegramPublisher.java
rename to
streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/telegram/TelegramSink.java
index 4abcd4fde..5dffe1c78 100644
---
a/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/telegram/TelegramPublisher.java
+++
b/streampipes-extensions/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/telegram/TelegramSink.java
@@ -19,9 +19,18 @@
package org.apache.streampipes.sinks.notifications.jvm.telegram;
import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.DataSinkType;
+import org.apache.streampipes.model.graph.DataSinkDescription;
import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.sdk.builder.DataSinkBuilder;
+import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder;
+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.runtime.EventSink;
+import org.apache.streampipes.wrapper.standalone.SinkParams;
+import org.apache.streampipes.wrapper.standalone.StreamPipesDataSink;
import okhttp3.OkHttpClient;
import okhttp3.Request;
@@ -33,21 +42,44 @@ import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.util.Map;
-public class TelegramPublisher implements EventSink<TelegramParameters> {
+public class TelegramSink extends StreamPipesDataSink {
+ private static final String CHANNEL_NAME_OR_CHAT_ID = "channel-chat-name";
+ private static final String MESSAGE_TEXT = "message-text";
+ private static final String BOT_API_KEY = "api-key";
+
private static final String ENDPOINT =
"https://api.telegram.org/bot%s/sendMessage?chat_id=%s&text=%s&parse_mode=%s";;
private static final OkHttpClient HTTP_CLIENT = new OkHttpClient();
private static final String HASH_TAG = "#";
private static final String HTML = "HTML";
+
private String apiKey;
private String channelOrChatId;
private String message;
@Override
- public void onInvocation(TelegramParameters parameters,
- EventSinkRuntimeContext runtimeContext) {
- this.apiKey = parameters.getApiKey();
- this.channelOrChatId = parameters.getChannelOrChatId();
- this.message = parameters.getMessage();
+ public DataSinkDescription declareModel() {
+ return
DataSinkBuilder.create("org.apache.streampipes.sinks.notifications.jvm.telegram")
+ .withLocales(Locales.EN)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .category(DataSinkType.NOTIFICATION)
+ .requiredStream(StreamRequirementsBuilder
+ .create()
+ .requiredProperty(EpRequirements.anyProperty())
+ .build())
+ .requiredSecret(Labels.withId(BOT_API_KEY))
+ .requiredTextParameter(Labels.withId(CHANNEL_NAME_OR_CHAT_ID))
+ .requiredTextParameter(Labels.withId(MESSAGE_TEXT), true, true, true)
+ .build();
+ }
+
+
+ @Override
+ public void onInvocation(SinkParams parameters,
+ EventSinkRuntimeContext runtimeContext) throws
SpRuntimeException {
+ var extractor = parameters.extractor();
+ apiKey = extractor.secretValue(BOT_API_KEY);
+ channelOrChatId = extractor.singleValueParameter(CHANNEL_NAME_OR_CHAT_ID,
String.class);
+ message = extractor.singleValueParameter(MESSAGE_TEXT, String.class);
}
@Override
@@ -55,7 +87,7 @@ public class TelegramPublisher implements
EventSink<TelegramParameters> {
try {
String content = replacePlaceholders(event, this.message);
content = trimHTML(content);
- content = URLEncoder.encode(content, StandardCharsets.UTF_8.toString());
+ content = URLEncoder.encode(content, StandardCharsets.UTF_8);
String url = String.format(ENDPOINT, this.apiKey, this.channelOrChatId,
content, HTML);
Request request = new Request.Builder().url(url).build();
try (Response response = HTTP_CLIENT.newCall(request).execute()) {
@@ -71,8 +103,8 @@ public class TelegramPublisher implements
EventSink<TelegramParameters> {
}
@Override
- public void onDetach() {
- // do nothing.
+ public void onDetach() throws SpRuntimeException {
+ // Do nothing
}
private String replacePlaceholders(Event event, String content) {
