This is an automated email from the ASF dual-hosted git repository. riemer pushed a commit to branch STREAMPIPES-319 in repository https://gitbox.apache.org/repos/asf/incubator-streampipes-extensions.git
commit 8f013da0a44ada433ccfb7be1e79b1fffaa01c5d Author: Dominik Riemer <[email protected]> AuthorDate: Mon May 10 13:50:43 2021 +0200 [STREAMPIPES-319] Migrate pipeline element descriptions to servicedefinition builder --- streampipes-extensions-all-jvm/pom.xml | 2 +- .../processors/filters/jvm/FiltersJvmInit.java | 67 ++++++++++++---------- .../jvm/processor/compose/ComposeController.java | 5 +- .../textmining/jvm/TextMiningJvmInit.java | 42 +++++++------- .../sinks/internal/jvm/SinksInternalJvmInit.java | 45 ++++++++------- .../jvm/SinksNotificationsJvmInit.java | 44 ++++++++------ .../notifications/jvm/email/EmailPublisher.java | 30 +++++----- 7 files changed, 122 insertions(+), 113 deletions(-) diff --git a/streampipes-extensions-all-jvm/pom.xml b/streampipes-extensions-all-jvm/pom.xml index 358f0ba..577bfb3 100644 --- a/streampipes-extensions-all-jvm/pom.xml +++ b/streampipes-extensions-all-jvm/pom.xml @@ -71,4 +71,4 @@ <finalName>streampipes-extensions-all-jvm</finalName> </build> -</project> \ No newline at end of file +</project> diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/FiltersJvmInit.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/FiltersJvmInit.java index bd6cfdc..41929fb 100644 --- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/FiltersJvmInit.java +++ b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/FiltersJvmInit.java @@ -18,51 +18,56 @@ package org.apache.streampipes.processors.filters.jvm; -import org.apache.streampipes.container.init.DeclarersSingleton; +import org.apache.streampipes.container.model.SpServiceDefinition; +import org.apache.streampipes.container.model.SpServiceDefinitionBuilder; import org.apache.streampipes.container.standalone.init.StandaloneModelSubmitter; import org.apache.streampipes.dataformat.cbor.CborDataFormatFactory; -import org.apache.streampipes.dataformat.fst.FstDataFormatFactory; import org.apache.streampipes.dataformat.json.JsonDataFormatFactory; -import org.apache.streampipes.dataformat.smile.SmileDataFormatFactory; import org.apache.streampipes.messaging.jms.SpJmsProtocolFactory; import org.apache.streampipes.messaging.kafka.SpKafkaProtocolFactory; -import org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory; -import org.apache.streampipes.processors.filters.jvm.config.FiltersJvmConfig; -import org.apache.streampipes.processors.filters.jvm.processor.compose.ComposeController; -import org.apache.streampipes.processors.filters.jvm.processor.enrich.MergeByEnrichController; -import org.apache.streampipes.processors.filters.jvm.processor.limit.RateLimitController; -import org.apache.streampipes.processors.filters.jvm.processor.merge.MergeByTimeController; import org.apache.streampipes.processors.filters.jvm.processor.numericalfilter.NumericalFilterController; -import org.apache.streampipes.processors.filters.jvm.processor.numericaltextfilter.NumericalTextFilterController; -import org.apache.streampipes.processors.filters.jvm.processor.projection.ProjectionController; import org.apache.streampipes.processors.filters.jvm.processor.textfilter.TextFilterController; import org.apache.streampipes.processors.filters.jvm.processor.threshold.ThresholdDetectionController; public class FiltersJvmInit extends StandaloneModelSubmitter { public static void main(String[] args) { - DeclarersSingleton.getInstance() - .add(new NumericalFilterController()) - .add(new ThresholdDetectionController()) - .add(new TextFilterController()) - .add(new ProjectionController()) - .add(new MergeByEnrichController()) - .add(new MergeByTimeController()) - .add(new ComposeController()) - .add(new NumericalTextFilterController()) - .add(new RateLimitController()); - DeclarersSingleton.getInstance().registerDataFormats( - new JsonDataFormatFactory(), - new CborDataFormatFactory(), - new SmileDataFormatFactory(), - new FstDataFormatFactory()); + SpServiceDefinition serviceDef = SpServiceDefinitionBuilder.create("org.apache.streampipes.processors.filters.jvm", + "StreamPipes Processors Filters (JVM)", + "", + 8090) +// .registerAdapters(new MqttProtocol(), +// new IssAdapter()) + .registerPipelineElements(new TextFilterController(), + new NumericalFilterController(), + new ThresholdDetectionController()) + .registerMessagingFormats(new JsonDataFormatFactory(), new CborDataFormatFactory()) + .registerMessagingProtocols(new SpKafkaProtocolFactory(), new SpJmsProtocolFactory()) + .build(); - DeclarersSingleton.getInstance().registerProtocols( - new SpKafkaProtocolFactory(), - new SpMqttProtocolFactory(), - new SpJmsProtocolFactory()); +// DeclarersSingleton.getInstance() +// .add(new NumericalFilterController()) +// .add(new ThresholdDetectionController()) +// .add(new TextFilterController()) +// .add(new ProjectionController()) +// .add(new MergeByEnrichController()) +// .add(new MergeByTimeController()) +// .add(new ComposeController()) +// .add(new NumericalTextFilterController()) +// .add(new RateLimitController()); +// +// DeclarersSingleton.getInstance().registerDataFormats( +// new JsonDataFormatFactory(), +// new CborDataFormatFactory(), +// new SmileDataFormatFactory(), +// new FstDataFormatFactory()); +// +// DeclarersSingleton.getInstance().registerProtocols( +// new SpKafkaProtocolFactory(), +// new SpMqttProtocolFactory(), +// new SpJmsProtocolFactory()); - new FiltersJvmInit().init(FiltersJvmConfig.INSTANCE); + new FiltersJvmInit().init(serviceDef); } } diff --git a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/ComposeController.java b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/ComposeController.java index d04f7dd..aa22d9f 100644 --- a/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/ComposeController.java +++ b/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/processor/compose/ComposeController.java @@ -20,7 +20,6 @@ package org.apache.streampipes.processors.filters.jvm.processor.compose; import org.apache.streampipes.model.DataProcessorType; import org.apache.streampipes.model.graph.DataProcessorDescription; import org.apache.streampipes.model.graph.DataProcessorInvocation; -import org.apache.streampipes.processors.filters.jvm.config.FiltersJvmConfig; import org.apache.streampipes.sdk.builder.ProcessingElementBuilder; import org.apache.streampipes.sdk.builder.StreamRequirementsBuilder; import org.apache.streampipes.sdk.extractor.ProcessingElementParameterExtractor; @@ -54,8 +53,8 @@ public class ComposeController extends StandaloneEventProcessingDeclarer<Compose } @Override - public ConfiguredEventProcessor<ComposeParameters> - onInvocation(DataProcessorInvocation graph, ProcessingElementParameterExtractor extractor) { + public ConfiguredEventProcessor<ComposeParameters> onInvocation(DataProcessorInvocation graph, + ProcessingElementParameterExtractor extractor) { List<String> outputKeySelectors = extractor.outputKeySelectors(); diff --git a/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/TextMiningJvmInit.java b/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/TextMiningJvmInit.java index a97ad02..cf630f5 100644 --- a/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/TextMiningJvmInit.java +++ b/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/TextMiningJvmInit.java @@ -18,7 +18,8 @@ package org.apache.streampipes.processors.textmining.jvm; -import org.apache.streampipes.container.init.DeclarersSingleton; +import org.apache.streampipes.container.model.SpServiceDefinition; +import org.apache.streampipes.container.model.SpServiceDefinitionBuilder; import org.apache.streampipes.container.standalone.init.StandaloneModelSubmitter; import org.apache.streampipes.dataformat.cbor.CborDataFormatFactory; import org.apache.streampipes.dataformat.fst.FstDataFormatFactory; @@ -27,7 +28,6 @@ import org.apache.streampipes.dataformat.smile.SmileDataFormatFactory; import org.apache.streampipes.messaging.jms.SpJmsProtocolFactory; import org.apache.streampipes.messaging.kafka.SpKafkaProtocolFactory; import org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory; -import org.apache.streampipes.processors.textmining.jvm.config.TextMiningJvmConfig; import org.apache.streampipes.processors.textmining.jvm.processor.chunker.ChunkerController; import org.apache.streampipes.processors.textmining.jvm.processor.language.LanguageDetectionController; import org.apache.streampipes.processors.textmining.jvm.processor.namefinder.NameFinderController; @@ -36,26 +36,24 @@ import org.apache.streampipes.processors.textmining.jvm.processor.sentencedetect import org.apache.streampipes.processors.textmining.jvm.processor.tokenizer.TokenizerController; public class TextMiningJvmInit extends StandaloneModelSubmitter { - public static void main(String[] args) { - DeclarersSingleton.getInstance() - .add(new LanguageDetectionController()) - .add(new TokenizerController()) - .add(new PartOfSpeechController()) - .add(new ChunkerController()) - .add(new NameFinderController()) - .add(new SentenceDetectionController()); + public static void main(String[] args) { - DeclarersSingleton.getInstance().registerDataFormats( - new JsonDataFormatFactory(), - new CborDataFormatFactory(), - new SmileDataFormatFactory(), - new FstDataFormatFactory()); + SpServiceDefinition serviceDef = SpServiceDefinitionBuilder.create("org.apache.streampipes.processors.textmining.jvm", "Processors Text Mining JVM", "", 8090) + .registerPipelineElements(new LanguageDetectionController(), + new TokenizerController(), + new PartOfSpeechController(), + new ChunkerController(), + new NameFinderController(), + new SentenceDetectionController()) + .registerMessagingFormats(new JsonDataFormatFactory(), + new CborDataFormatFactory(), + new SmileDataFormatFactory(), + new FstDataFormatFactory()) + .registerMessagingProtocols(new SpKafkaProtocolFactory(), + new SpJmsProtocolFactory(), + new SpMqttProtocolFactory()) + .build(); - DeclarersSingleton.getInstance().registerProtocols( - new SpKafkaProtocolFactory(), - new SpMqttProtocolFactory(), - new SpJmsProtocolFactory()); - - new TextMiningJvmInit().init(TextMiningJvmConfig.INSTANCE); - } + new TextMiningJvmInit().init(serviceDef); + } } diff --git a/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/SinksInternalJvmInit.java b/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/SinksInternalJvmInit.java index 9ad9d16..8ba9e29 100644 --- a/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/SinksInternalJvmInit.java +++ b/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/SinksInternalJvmInit.java @@ -18,16 +18,13 @@ package org.apache.streampipes.sinks.internal.jvm; -import org.apache.streampipes.container.init.DeclarersSingleton; +import org.apache.streampipes.container.model.SpServiceDefinition; +import org.apache.streampipes.container.model.SpServiceDefinitionBuilder; import org.apache.streampipes.container.standalone.init.StandaloneModelSubmitter; import org.apache.streampipes.dataformat.cbor.CborDataFormatFactory; -import org.apache.streampipes.dataformat.fst.FstDataFormatFactory; import org.apache.streampipes.dataformat.json.JsonDataFormatFactory; -import org.apache.streampipes.dataformat.smile.SmileDataFormatFactory; import org.apache.streampipes.messaging.jms.SpJmsProtocolFactory; import org.apache.streampipes.messaging.kafka.SpKafkaProtocolFactory; -import org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory; -import org.apache.streampipes.sinks.internal.jvm.config.SinksInternalJvmConfig; import org.apache.streampipes.sinks.internal.jvm.dashboard.DashboardController; import org.apache.streampipes.sinks.internal.jvm.datalake.DataLakeController; import org.apache.streampipes.sinks.internal.jvm.notification.NotificationController; @@ -35,22 +32,30 @@ import org.apache.streampipes.sinks.internal.jvm.notification.NotificationContro public class SinksInternalJvmInit extends StandaloneModelSubmitter { public static void main(String[] args) { - DeclarersSingleton.getInstance() - .add(new NotificationController()) - .add(new DataLakeController()) - .add(new DashboardController()); - DeclarersSingleton.getInstance().registerDataFormats( - new JsonDataFormatFactory(), - new CborDataFormatFactory(), - new SmileDataFormatFactory(), - new FstDataFormatFactory()); + SpServiceDefinition serviceDef = SpServiceDefinitionBuilder.create("org.apache.streampipes.sinks.internal.jvm", "StreamPipes Sinks Internal JVM", "",8090) + .registerPipelineElements(new DashboardController(), + new DataLakeController(), + new NotificationController()) + .registerMessagingFormats(new JsonDataFormatFactory(), new CborDataFormatFactory()) + .registerMessagingProtocols(new SpKafkaProtocolFactory(), new SpJmsProtocolFactory()) + .build(); +// DeclarersSingleton.getInstance() +// .add(new NotificationController()) +// .add(new DataLakeController()) +// .add(new DashboardController()); +// +// DeclarersSingleton.getInstance().registerDataFormats( +// new JsonDataFormatFactory(), +// new CborDataFormatFactory(), +// new SmileDataFormatFactory(), +// new FstDataFormatFactory()); +// +// DeclarersSingleton.getInstance().registerProtocols( +// new SpKafkaProtocolFactory(), +// new SpMqttProtocolFactory(), +// new SpJmsProtocolFactory()); - DeclarersSingleton.getInstance().registerProtocols( - new SpKafkaProtocolFactory(), - new SpMqttProtocolFactory(), - new SpJmsProtocolFactory()); - - new SinksInternalJvmInit().init(SinksInternalJvmConfig.INSTANCE); + new SinksInternalJvmInit().init(serviceDef); } } diff --git a/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/SinksNotificationsJvmInit.java b/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/SinksNotificationsJvmInit.java index 04414fd..c4534d0 100644 --- a/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/SinksNotificationsJvmInit.java +++ b/streampipes-sinks-notifications-jvm/src/main/java/org/apache/streampipes/sinks/notifications/jvm/SinksNotificationsJvmInit.java @@ -18,7 +18,8 @@ package org.apache.streampipes.sinks.notifications.jvm; -import org.apache.streampipes.container.init.DeclarersSingleton; +import org.apache.streampipes.container.model.SpServiceDefinition; +import org.apache.streampipes.container.model.SpServiceDefinitionBuilder; import org.apache.streampipes.container.standalone.init.StandaloneModelSubmitter; import org.apache.streampipes.dataformat.cbor.CborDataFormatFactory; import org.apache.streampipes.dataformat.fst.FstDataFormatFactory; @@ -27,7 +28,7 @@ import org.apache.streampipes.dataformat.smile.SmileDataFormatFactory; import org.apache.streampipes.messaging.jms.SpJmsProtocolFactory; import org.apache.streampipes.messaging.kafka.SpKafkaProtocolFactory; import org.apache.streampipes.messaging.mqtt.SpMqttProtocolFactory; -import org.apache.streampipes.sinks.notifications.jvm.config.SinksNotificationsJvmConfig; +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; @@ -36,23 +37,28 @@ import org.apache.streampipes.sinks.notifications.jvm.telegram.TelegramControlle public class SinksNotificationsJvmInit extends StandaloneModelSubmitter { public static void main(String[] args) { - DeclarersSingleton.getInstance() - .add(new EmailController()) - .add(new OneSignalController()) - .add(new SlackNotificationController()) - .add(new TelegramController()); + SpServiceDefinition serviceDef = SpServiceDefinitionBuilder.create("org.apache.streampipes.sinks.notifications.jvm", "Sinks Notifications JVM", "", 8096) + .registerPipelineElements(new EmailController(), + new OneSignalController(), + new SlackNotificationController(), + new TelegramController()) + .registerMessagingFormats(new JsonDataFormatFactory(), + new CborDataFormatFactory(), + new SmileDataFormatFactory(), + new FstDataFormatFactory()) + .registerMessagingProtocols(new SpKafkaProtocolFactory(), + new SpJmsProtocolFactory(), + new SpMqttProtocolFactory()) + .addConfig(ConfigKeys.WEBSOCKET_PROTOCOL, "ws", "") + .addConfig(ConfigKeys.EMAIL_FROM, "", "The sender address of the email") + .addConfig(ConfigKeys.EMAIL_USERNAME, "", "The username of the email account") + .addConfig(ConfigKeys.EMAIL_PASSWORD, "", "The password of the email account") + .addConfig(ConfigKeys.EMAIL_SMTP_HOST, "", "The SMTP Host") + .addConfig(ConfigKeys.EMAIL_SMTP_PORT, "", "The SMTP Port") + .addConfig(ConfigKeys.EMAIL_STARTTLS, false, "Use startls?") + .addConfig(ConfigKeys.EMAIL_SLL, false, "Use SLL?") + .build(); - DeclarersSingleton.getInstance().registerDataFormats( - new JsonDataFormatFactory(), - new CborDataFormatFactory(), - new SmileDataFormatFactory(), - new FstDataFormatFactory()); - - DeclarersSingleton.getInstance().registerProtocols( - new SpKafkaProtocolFactory(), - new SpMqttProtocolFactory(), - new SpJmsProtocolFactory()); - - new SinksNotificationsJvmInit().init(SinksNotificationsJvmConfig.INSTANCE); + new SinksNotificationsJvmInit().init(serviceDef); } } 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..1a086f1 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 @@ -19,24 +19,19 @@ package org.apache.streampipes.sinks.notifications.jvm.email; import org.apache.streampipes.commons.exceptions.SpRuntimeException; +import org.apache.streampipes.config.SpConfig; import org.apache.streampipes.logging.api.Logger; import org.apache.streampipes.model.runtime.Event; import org.apache.streampipes.model.runtime.EventConverter; -import org.apache.streampipes.sinks.notifications.jvm.config.SinksNotificationsJvmConfig; +import org.apache.streampipes.sinks.notifications.jvm.config.ConfigKeys; import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext; import org.apache.streampipes.wrapper.runtime.EventSink; -import java.util.Map; -import java.util.Properties; - -import javax.mail.Authenticator; -import javax.mail.Message; -import javax.mail.MessagingException; -import javax.mail.PasswordAuthentication; -import javax.mail.Session; -import javax.mail.Transport; +import javax.mail.*; import javax.mail.internet.InternetAddress; import javax.mail.internet.MimeMessage; +import java.util.Map; +import java.util.Properties; public class EmailPublisher implements EventSink<EmailParameters> { @@ -49,16 +44,17 @@ public class EmailPublisher implements EventSink<EmailParameters> { public void onInvocation(EmailParameters parameters, EventSinkRuntimeContext runtimeContext) { LOG = parameters.getGraph().getLogger(EmailPublisher.class); - String from = SinksNotificationsJvmConfig.INSTANCE.getEmailFrom(); + SpConfig config = runtimeContext.getConfigStore().getConfig(); + String from = config.getString(ConfigKeys.EMAIL_FROM); String to = parameters.getToEmailAddress(); String subject = parameters.getSubject(); this.content = parameters.getContent(); - String username = SinksNotificationsJvmConfig.INSTANCE.getEmailUsername(); - String password = SinksNotificationsJvmConfig.INSTANCE.getEmailPassword(); - String host = SinksNotificationsJvmConfig.INSTANCE.getEmailSmtpHost(); - int port = SinksNotificationsJvmConfig.INSTANCE.getEmailSmtpPort(); - boolean starttls = SinksNotificationsJvmConfig.INSTANCE.useEmailStarttls(); - boolean ssl = SinksNotificationsJvmConfig.INSTANCE.useEmailSll(); + String username = config.getString(ConfigKeys.EMAIL_USERNAME); + String password = config.getString(ConfigKeys.EMAIL_PASSWORD); + String host = config.getString(ConfigKeys.EMAIL_SMTP_HOST); + int port = config.getInteger(ConfigKeys.EMAIL_SMTP_PORT); + boolean starttls = config.getBoolean(ConfigKeys.EMAIL_STARTTLS); + boolean ssl = config.getBoolean(ConfigKeys.EMAIL_SLL); Properties properties = new Properties(); properties.setProperty("mail.smtp.host", host);
