This is an automated email from the ASF dual-hosted git repository. bossenti pushed a commit to branch chore/fully-support-nats in repository https://gitbox.apache.org/repos/asf/streampipes.git
commit c777ed98e486f4f53a00ac36bd343a4f6aaef9be Author: bossenti <[email protected]> AuthorDate: Sun May 28 21:38:47 2023 +0200 chore: add missing support of NATS as messaging protocol --- .../main/resources/archetype-resources/src/main/java/Init.java | 4 +++- .../main/resources/archetype-resources/src/main/java/Init.java | 4 +++- .../main/resources/archetype-resources/src/main/java/Init.java | 4 +++- .../extensions/connectors/influx/InfluxConnectorsInit.java | 4 +++- .../streampipes/pe/flink/AllFlinkPipelineElementsInit.java | 4 +++- .../org/apache/streampipes/pe/jvm/AllPipelineElementsInit.java | 4 +++- .../processors/aggregation/flink/AggregationFlinkInit.java | 4 +++- .../processors/changedetection/jvm/ChangeDetectionJvmInit.java | 4 +++- .../processors/enricher/flink/EnricherFlinkInit.java | 4 +++- .../streampipes/processors/enricher/jvm/EnricherJvmInit.java | 4 +++- .../streampipes/processors/filters/jvm/FiltersJvmInit.java | 4 +++- .../streampipes/processors/siddhi/FiltersSiddhiInit.java | 4 +++- .../apache/streampipes/processor/geo/flink/GeoFlinkInit.java | 4 +++- .../org/apache/streampipes/processors/geo/jvm/GeoJvmInit.java | 4 +++- .../processors/imageprocessing/jvm/ImageProcessingJvmInit.java | 4 +++- .../pattern/detection/flink/PatternDetectionFlinkInit.java | 4 +++- .../processors/statistics/flink/StatisticsFlinkInit.java | 4 +++- .../processors/textmining/flink/TextMiningFlinkInit.java | 4 +++- .../processors/textmining/jvm/TextMiningJvmInit.java | 4 +++- .../transformation/flink/TransformationFlinkInit.java | 4 +++- .../processors/transformation/jvm/TransformationJvmInit.java | 4 +++- .../apache/streampipes/sinks/brokers/jvm/BrokersJvmInit.java | 4 +++- .../streampipes/sinks/databases/flink/DatabasesFlinkInit.java | 4 +++- .../streampipes/sinks/databases/jvm/DatabasesJvmInit.java | 4 +++- .../streampipes/sinks/internal/jvm/SinksInternalJvmInit.java | 4 +++- .../sinks/notifications/jvm/SinksNotificationsJvmInit.java | 10 +++++++--- 26 files changed, 82 insertions(+), 28 deletions(-) diff --git a/archetypes/streampipes-archetype-extensions-jvm/src/main/resources/archetype-resources/src/main/java/Init.java b/archetypes/streampipes-archetype-extensions-jvm/src/main/resources/archetype-resources/src/main/java/Init.java index 74c27462b..d6c46481d 100644 --- a/archetypes/streampipes-archetype-extensions-jvm/src/main/resources/archetype-resources/src/main/java/Init.java +++ b/archetypes/streampipes-archetype-extensions-jvm/src/main/resources/archetype-resources/src/main/java/Init.java @@ -30,6 +30,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter; import ${package}.pe.${packageName}.${classNamePrefix}DataProcessor; @@ -56,7 +57,8 @@ public class Init extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .build(); } } diff --git a/archetypes/streampipes-archetype-pe-processors-flink/src/main/resources/archetype-resources/src/main/java/Init.java b/archetypes/streampipes-archetype-pe-processors-flink/src/main/resources/archetype-resources/src/main/java/Init.java index dc52194fb..aa1df4036 100644 --- a/archetypes/streampipes-archetype-pe-processors-flink/src/main/resources/archetype-resources/src/main/java/Init.java +++ b/archetypes/streampipes-archetype-pe-processors-flink/src/main/resources/archetype-resources/src/main/java/Init.java @@ -34,6 +34,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter; public class Init extends ExtensionsModelSubmitter { @@ -57,7 +58,8 @@ public class Init extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .addConfig(ConfigKeys.FLINK_HOST, "jobmanager", "Hostname of the Flink Jobmanager") .addConfig(ConfigKeys.FLINK_PORT, 8081, "Port of the Flink Jobmanager") .addConfig(ConfigKeys.DEBUG, false, "Debug/Mini cluster mode of Flink program") diff --git a/archetypes/streampipes-archetype-pe-sinks-flink/src/main/resources/archetype-resources/src/main/java/Init.java b/archetypes/streampipes-archetype-pe-sinks-flink/src/main/resources/archetype-resources/src/main/java/Init.java index ef8c40914..51e7ddfa7 100644 --- a/archetypes/streampipes-archetype-pe-sinks-flink/src/main/resources/archetype-resources/src/main/java/Init.java +++ b/archetypes/streampipes-archetype-pe-sinks-flink/src/main/resources/archetype-resources/src/main/java/Init.java @@ -30,6 +30,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter; import ${package}.config.ConfigKeys; @@ -56,7 +57,8 @@ public class Init extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .addConfig(ConfigKeys.FLINK_HOST, "jobmanager", "Hostname of the Flink Jobmanager") .addConfig(ConfigKeys.FLINK_PORT, 8081, "Port of the Flink Jobmanager") .addConfig(ConfigKeys.DEBUG, false, "Debug/Mini cluster mode of Flink program") diff --git a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/InfluxConnectorsInit.java b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/InfluxConnectorsInit.java index 64ff312aa..3a19d3dd9 100644 --- a/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/InfluxConnectorsInit.java +++ b/streampipes-extensions/streampipes-connectors-influx/src/main/java/org/apache/streampipes/extensions/connectors/influx/InfluxConnectorsInit.java @@ -29,6 +29,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter; public class InfluxConnectorsInit extends ExtensionsModelSubmitter { @@ -55,7 +56,8 @@ public class InfluxConnectorsInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .build(); } } diff --git a/streampipes-extensions/streampipes-pipeline-elements-all-flink/src/main/java/org/apache/streampipes/pe/flink/AllFlinkPipelineElementsInit.java b/streampipes-extensions/streampipes-pipeline-elements-all-flink/src/main/java/org/apache/streampipes/pe/flink/AllFlinkPipelineElementsInit.java index 35403c322..7c4339d73 100644 --- a/streampipes-extensions/streampipes-pipeline-elements-all-flink/src/main/java/org/apache/streampipes/pe/flink/AllFlinkPipelineElementsInit.java +++ b/streampipes-extensions/streampipes-pipeline-elements-all-flink/src/main/java/org/apache/streampipes/pe/flink/AllFlinkPipelineElementsInit.java @@ -26,6 +26,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.processor.geo.flink.GeoFlinkInit; import org.apache.streampipes.processors.aggregation.flink.AggregationFlinkInit; import org.apache.streampipes.processors.enricher.flink.EnricherFlinkInit; @@ -65,7 +66,8 @@ public class AllFlinkPipelineElementsInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .build(); } } diff --git a/streampipes-extensions/streampipes-pipeline-elements-all-jvm/src/main/java/org/apache/streampipes/pe/jvm/AllPipelineElementsInit.java b/streampipes-extensions/streampipes-pipeline-elements-all-jvm/src/main/java/org/apache/streampipes/pe/jvm/AllPipelineElementsInit.java index 5ce90d4a0..8a25c3e5f 100644 --- a/streampipes-extensions/streampipes-pipeline-elements-all-jvm/src/main/java/org/apache/streampipes/pe/jvm/AllPipelineElementsInit.java +++ b/streampipes-extensions/streampipes-pipeline-elements-all-jvm/src/main/java/org/apache/streampipes/pe/jvm/AllPipelineElementsInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.processors.changedetection.jvm.ChangeDetectionJvmInit; import org.apache.streampipes.processors.enricher.jvm.EnricherJvmInit; import org.apache.streampipes.processors.filters.jvm.FiltersJvmInit; @@ -74,7 +75,8 @@ public class AllPipelineElementsInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .build(); } } diff --git a/streampipes-extensions/streampipes-processors-aggregation-flink/src/main/java/org/apache/streampipes/processors/aggregation/flink/AggregationFlinkInit.java b/streampipes-extensions/streampipes-processors-aggregation-flink/src/main/java/org/apache/streampipes/processors/aggregation/flink/AggregationFlinkInit.java index 9596c9772..e65423e20 100644 --- a/streampipes-extensions/streampipes-processors-aggregation-flink/src/main/java/org/apache/streampipes/processors/aggregation/flink/AggregationFlinkInit.java +++ b/streampipes-extensions/streampipes-processors-aggregation-flink/src/main/java/org/apache/streampipes/processors/aggregation/flink/AggregationFlinkInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.processors.aggregation.flink.config.ConfigKeys; import org.apache.streampipes.processors.aggregation.flink.processor.aggregation.AggregationController; import org.apache.streampipes.processors.aggregation.flink.processor.count.CountController; @@ -60,7 +61,8 @@ public class AggregationFlinkInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .addConfig(ConfigKeys.FLINK_HOST, "jobmanager", "Hostname of the Flink Jobmanager") .addConfig(ConfigKeys.FLINK_PORT, 8081, "Port of the Flink Jobmanager") .addConfig(ConfigKeys.DEBUG, false, "Debug/Mini cluster mode of Flink program") diff --git a/streampipes-extensions/streampipes-processors-change-detection-jvm/src/main/java/org/apache/streampipes/processors/changedetection/jvm/ChangeDetectionJvmInit.java b/streampipes-extensions/streampipes-processors-change-detection-jvm/src/main/java/org/apache/streampipes/processors/changedetection/jvm/ChangeDetectionJvmInit.java index 40db44e1c..aa90a6943 100644 --- a/streampipes-extensions/streampipes-processors-change-detection-jvm/src/main/java/org/apache/streampipes/processors/changedetection/jvm/ChangeDetectionJvmInit.java +++ b/streampipes-extensions/streampipes-processors-change-detection-jvm/src/main/java/org/apache/streampipes/processors/changedetection/jvm/ChangeDetectionJvmInit.java @@ -28,6 +28,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.processors.changedetection.jvm.welford.WelfordChangeDetection; import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter; @@ -54,7 +55,8 @@ public class ChangeDetectionJvmInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .build(); } } diff --git a/streampipes-extensions/streampipes-processors-enricher-flink/src/main/java/org/apache/streampipes/processors/enricher/flink/EnricherFlinkInit.java b/streampipes-extensions/streampipes-processors-enricher-flink/src/main/java/org/apache/streampipes/processors/enricher/flink/EnricherFlinkInit.java index db55e09a7..0f7cf58f1 100644 --- a/streampipes-extensions/streampipes-processors-enricher-flink/src/main/java/org/apache/streampipes/processors/enricher/flink/EnricherFlinkInit.java +++ b/streampipes-extensions/streampipes-processors-enricher-flink/src/main/java/org/apache/streampipes/processors/enricher/flink/EnricherFlinkInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.processors.enricher.flink.config.ConfigKeys; import org.apache.streampipes.processors.enricher.flink.processor.timestamp.TimestampController; import org.apache.streampipes.processors.enricher.flink.processor.urldereferencing.UrlDereferencingController; @@ -54,7 +55,8 @@ public class EnricherFlinkInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .addConfig(ConfigKeys.FLINK_HOST, "jobmanager", "Hostname of the Flink Jobmanager") .addConfig(ConfigKeys.FLINK_PORT, 8081, "Port of the Flink Jobmanager") .addConfig(ConfigKeys.DEBUG, false, "Debug/Mini cluster mode of Flink program") diff --git a/streampipes-extensions/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/EnricherJvmInit.java b/streampipes-extensions/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/EnricherJvmInit.java index 7fbb7368a..fa48a77fc 100644 --- a/streampipes-extensions/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/EnricherJvmInit.java +++ b/streampipes-extensions/streampipes-processors-enricher-jvm/src/main/java/org/apache/streampipes/processors/enricher/jvm/EnricherJvmInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.processors.enricher.jvm.processor.jseval.JSEvalProcessor; import org.apache.streampipes.processors.enricher.jvm.processor.math.MathOpProcessor; import org.apache.streampipes.processors.enricher.jvm.processor.math.staticmathop.StaticMathOpProcessor; @@ -61,7 +62,8 @@ public class EnricherJvmInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .build(); } } diff --git a/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/FiltersJvmInit.java b/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/FiltersJvmInit.java index cd9b5eeca..c88ca5ac3 100644 --- a/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/FiltersJvmInit.java +++ b/streampipes-extensions/streampipes-processors-filters-jvm/src/main/java/org/apache/streampipes/processors/filters/jvm/FiltersJvmInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.processors.filters.jvm.processor.booleanfilter.BooleanFilterProcessor; import org.apache.streampipes.processors.filters.jvm.processor.compose.ComposeProcessor; import org.apache.streampipes.processors.filters.jvm.processor.enrich.MergeByEnrichProcessor; @@ -78,7 +79,8 @@ public class FiltersJvmInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .build(); } } diff --git a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/FiltersSiddhiInit.java b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/FiltersSiddhiInit.java index 06e727215..d7099f379 100644 --- a/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/FiltersSiddhiInit.java +++ b/streampipes-extensions/streampipes-processors-filters-siddhi/src/main/java/org/apache/streampipes/processors/siddhi/FiltersSiddhiInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.processors.siddhi.count.CountAggregation; import org.apache.streampipes.processors.siddhi.filter.NumericalFilterSiddhiProcessor; import org.apache.streampipes.processors.siddhi.listcollector.ListCollector; @@ -63,7 +64,8 @@ public class FiltersSiddhiInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .build(); } } diff --git a/streampipes-extensions/streampipes-processors-geo-flink/src/main/java/org/apache/streampipes/processor/geo/flink/GeoFlinkInit.java b/streampipes-extensions/streampipes-processors-geo-flink/src/main/java/org/apache/streampipes/processor/geo/flink/GeoFlinkInit.java index a50e4252d..9aa5917a4 100644 --- a/streampipes-extensions/streampipes-processors-geo-flink/src/main/java/org/apache/streampipes/processor/geo/flink/GeoFlinkInit.java +++ b/streampipes-extensions/streampipes-processors-geo-flink/src/main/java/org/apache/streampipes/processor/geo/flink/GeoFlinkInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.processor.geo.flink.config.ConfigKeys; import org.apache.streampipes.processor.geo.flink.processor.gridenricher.SpatialGridEnrichmentController; import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter; @@ -52,7 +53,8 @@ public class GeoFlinkInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .addConfig(ConfigKeys.FLINK_HOST, "jobmanager", "Hostname of the Flink Jobmanager") .addConfig(ConfigKeys.FLINK_PORT, 8081, "Port of the Flink Jobmanager") .addConfig(ConfigKeys.DEBUG, false, "Debug/Mini cluster mode of Flink program") diff --git a/streampipes-extensions/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/GeoJvmInit.java b/streampipes-extensions/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/GeoJvmInit.java index d3024ff10..e391ed874 100644 --- a/streampipes-extensions/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/GeoJvmInit.java +++ b/streampipes-extensions/streampipes-processors-geo-jvm/src/main/java/org/apache/streampipes/processors/geo/jvm/GeoJvmInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.processors.geo.jvm.config.ConfigKeys; import org.apache.streampipes.processors.geo.jvm.jts.processor.buffergeometry.BufferGeomProcessor; import org.apache.streampipes.processors.geo.jvm.jts.processor.bufferpoint.BufferPointProcessor; @@ -87,7 +88,8 @@ public class GeoJvmInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .addConfig(ConfigKeys.GOOGLE_API_KEY, "", "Google Maps API key") .build(); } diff --git a/streampipes-extensions/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/ImageProcessingJvmInit.java b/streampipes-extensions/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/ImageProcessingJvmInit.java index 645ad3a8a..b9aa41e4f 100644 --- a/streampipes-extensions/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/ImageProcessingJvmInit.java +++ b/streampipes-extensions/streampipes-processors-image-processing-jvm/src/main/java/org/apache/streampipes/processors/imageprocessing/jvm/ImageProcessingJvmInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.processors.imageprocessing.jvm.processor.genericclassification.GenericImageClassificationProcessor; import org.apache.streampipes.processors.imageprocessing.jvm.processor.imagecropper.ImageCropperProcessor; import org.apache.streampipes.processors.imageprocessing.jvm.processor.imageenrichment.ImageEnrichmentProcessor; @@ -54,7 +55,8 @@ public class ImageProcessingJvmInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .build(); } } diff --git a/streampipes-extensions/streampipes-processors-pattern-detection-flink/src/main/java/org/apache/streampipes/processors/pattern/detection/flink/PatternDetectionFlinkInit.java b/streampipes-extensions/streampipes-processors-pattern-detection-flink/src/main/java/org/apache/streampipes/processors/pattern/detection/flink/PatternDetectionFlinkInit.java index 37718f464..cbcf3e9a4 100644 --- a/streampipes-extensions/streampipes-processors-pattern-detection-flink/src/main/java/org/apache/streampipes/processors/pattern/detection/flink/PatternDetectionFlinkInit.java +++ b/streampipes-extensions/streampipes-processors-pattern-detection-flink/src/main/java/org/apache/streampipes/processors/pattern/detection/flink/PatternDetectionFlinkInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.processors.pattern.detection.flink.config.ConfigKeys; import org.apache.streampipes.processors.pattern.detection.flink.processor.absence.AbsenceController; import org.apache.streampipes.processors.pattern.detection.flink.processor.and.AndController; @@ -60,7 +61,8 @@ public class PatternDetectionFlinkInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .addConfig(ConfigKeys.FLINK_HOST, "jobmanager", "Hostname of the Flink Jobmanager") .addConfig(ConfigKeys.FLINK_PORT, 8081, "Port of the Flink Jobmanager") .addConfig(ConfigKeys.DEBUG, false, "Debug/Mini cluster mode of Flink program") diff --git a/streampipes-extensions/streampipes-processors-statistics-flink/src/main/java/org/apache/streampipes/processors/statistics/flink/StatisticsFlinkInit.java b/streampipes-extensions/streampipes-processors-statistics-flink/src/main/java/org/apache/streampipes/processors/statistics/flink/StatisticsFlinkInit.java index 039796c2f..e61911cd2 100644 --- a/streampipes-extensions/streampipes-processors-statistics-flink/src/main/java/org/apache/streampipes/processors/statistics/flink/StatisticsFlinkInit.java +++ b/streampipes-extensions/streampipes-processors-statistics-flink/src/main/java/org/apache/streampipes/processors/statistics/flink/StatisticsFlinkInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.processors.statistics.flink.config.ConfigKeys; import org.apache.streampipes.processors.statistics.flink.processor.stat.summary.StatisticsSummaryController; import org.apache.streampipes.processors.statistics.flink.processor.stat.window.StatisticsSummaryControllerWindow; @@ -54,7 +55,8 @@ public class StatisticsFlinkInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .addConfig(ConfigKeys.FLINK_HOST, "jobmanager", "Hostname of the Flink Jobmanager") .addConfig(ConfigKeys.FLINK_PORT, 8081, "Port of the Flink Jobmanager") .addConfig(ConfigKeys.DEBUG, false, "Debug/Mini cluster mode of Flink program") diff --git a/streampipes-extensions/streampipes-processors-text-mining-flink/src/main/java/org/apache/streampipes/processors/textmining/flink/TextMiningFlinkInit.java b/streampipes-extensions/streampipes-processors-text-mining-flink/src/main/java/org/apache/streampipes/processors/textmining/flink/TextMiningFlinkInit.java index e72d50b96..333b07985 100644 --- a/streampipes-extensions/streampipes-processors-text-mining-flink/src/main/java/org/apache/streampipes/processors/textmining/flink/TextMiningFlinkInit.java +++ b/streampipes-extensions/streampipes-processors-text-mining-flink/src/main/java/org/apache/streampipes/processors/textmining/flink/TextMiningFlinkInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.processors.textmining.flink.config.ConfigKeys; import org.apache.streampipes.processors.textmining.flink.processor.wordcount.WordCountController; import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter; @@ -52,7 +53,8 @@ public class TextMiningFlinkInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .addConfig(ConfigKeys.FLINK_HOST, "jobmanager", "Hostname of the Flink Jobmanager") .addConfig(ConfigKeys.FLINK_PORT, 8081, "Port of the Flink Jobmanager") .addConfig(ConfigKeys.DEBUG, false, "Debug/Mini cluster mode of Flink program") diff --git a/streampipes-extensions/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/TextMiningJvmInit.java b/streampipes-extensions/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/TextMiningJvmInit.java index b4ead4b46..e9c1ed796 100644 --- a/streampipes-extensions/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/TextMiningJvmInit.java +++ b/streampipes-extensions/streampipes-processors-text-mining-jvm/src/main/java/org/apache/streampipes/processors/textmining/jvm/TextMiningJvmInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.processors.textmining.jvm.processor.chunker.ChunkerProcessor; import org.apache.streampipes.processors.textmining.jvm.processor.language.LanguageDetectionProcessor; import org.apache.streampipes.processors.textmining.jvm.processor.namefinder.NameFinderProcessor; @@ -61,7 +62,8 @@ public class TextMiningJvmInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .build(); } } diff --git a/streampipes-extensions/streampipes-processors-transformation-flink/src/main/java/org/apache/streampipes/processors/transformation/flink/TransformationFlinkInit.java b/streampipes-extensions/streampipes-processors-transformation-flink/src/main/java/org/apache/streampipes/processors/transformation/flink/TransformationFlinkInit.java index cc7a9ec8e..4fb19dfe1 100644 --- a/streampipes-extensions/streampipes-processors-transformation-flink/src/main/java/org/apache/streampipes/processors/transformation/flink/TransformationFlinkInit.java +++ b/streampipes-extensions/streampipes-processors-transformation-flink/src/main/java/org/apache/streampipes/processors/transformation/flink/TransformationFlinkInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.processors.transformation.flink.config.ConfigKeys; import org.apache.streampipes.processors.transformation.flink.processor.boilerplate.BoilerplateController; import org.apache.streampipes.processors.transformation.flink.processor.converter.FieldConverterController; @@ -64,7 +65,8 @@ public class TransformationFlinkInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .addConfig(ConfigKeys.FLINK_HOST, "jobmanager", "Hostname of the Flink Jobmanager") .addConfig(ConfigKeys.FLINK_PORT, 8081, "Port of the Flink Jobmanager") .addConfig(ConfigKeys.DEBUG, false, "Debug/Mini cluster mode of Flink program") diff --git a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/TransformationJvmInit.java b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/TransformationJvmInit.java index 41ec49a8c..a5cd3615b 100644 --- a/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/TransformationJvmInit.java +++ b/streampipes-extensions/streampipes-processors-transformation-jvm/src/main/java/org/apache/streampipes/processors/transformation/jvm/TransformationJvmInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.processors.transformation.jvm.processor.array.count.CountArrayProcessor; import org.apache.streampipes.processors.transformation.jvm.processor.array.split.SplitArrayProcessor; import org.apache.streampipes.processors.transformation.jvm.processor.booloperator.counter.BooleanCounterProcessor; @@ -96,7 +97,8 @@ public class TransformationJvmInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .build(); } } diff --git a/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/BrokersJvmInit.java b/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/BrokersJvmInit.java index f3920b0e6..ec3560787 100644 --- a/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/BrokersJvmInit.java +++ b/streampipes-extensions/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/BrokersJvmInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter; import org.apache.streampipes.sinks.brokers.jvm.bufferrest.BufferRestPublisherSink; import org.apache.streampipes.sinks.brokers.jvm.jms.JmsPublisherSink; @@ -72,7 +73,8 @@ public class BrokersJvmInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .build(); } } diff --git a/streampipes-extensions/streampipes-sinks-databases-flink/src/main/java/org/apache/streampipes/sinks/databases/flink/DatabasesFlinkInit.java b/streampipes-extensions/streampipes-sinks-databases-flink/src/main/java/org/apache/streampipes/sinks/databases/flink/DatabasesFlinkInit.java index 0d314299e..e409bd737 100644 --- a/streampipes-extensions/streampipes-sinks-databases-flink/src/main/java/org/apache/streampipes/sinks/databases/flink/DatabasesFlinkInit.java +++ b/streampipes-extensions/streampipes-sinks-databases-flink/src/main/java/org/apache/streampipes/sinks/databases/flink/DatabasesFlinkInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter; import org.apache.streampipes.sinks.databases.flink.config.ConfigKeys; import org.apache.streampipes.sinks.databases.flink.elasticsearch.ElasticSearchController; @@ -52,7 +53,8 @@ public class DatabasesFlinkInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .addConfig(ConfigKeys.FLINK_HOST, "jobmanager", "Hostname of the Flink Jobmanager") .addConfig(ConfigKeys.FLINK_PORT, 8081, "Port of the Flink Jobmanager") .addConfig(ConfigKeys.DEBUG, false, "Debug/Mini cluster mode of Flink program") diff --git a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/DatabasesJvmInit.java b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/DatabasesJvmInit.java index 2c8cddb88..478780c4e 100644 --- a/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/DatabasesJvmInit.java +++ b/streampipes-extensions/streampipes-sinks-databases-jvm/src/main/java/org/apache/streampipes/sinks/databases/jvm/DatabasesJvmInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter; import org.apache.streampipes.sinks.databases.jvm.couchdb.CouchDbSink; import org.apache.streampipes.sinks.databases.jvm.ditto.DittoSink; @@ -62,7 +63,8 @@ public class DatabasesJvmInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .build(); } } diff --git a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/SinksInternalJvmInit.java b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/SinksInternalJvmInit.java index 9b9641366..38961624c 100644 --- a/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/SinksInternalJvmInit.java +++ b/streampipes-extensions/streampipes-sinks-internal-jvm/src/main/java/org/apache/streampipes/sinks/internal/jvm/SinksInternalJvmInit.java @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter; import org.apache.streampipes.sinks.internal.jvm.datalake.DataLakeSink; import org.apache.streampipes.sinks.internal.jvm.notification.NotificationProducer; @@ -50,7 +51,8 @@ public class SinksInternalJvmInit extends ExtensionsModelSubmitter { .registerMessagingProtocols( new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .build(); 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 1caeaa7a0..9f8995213 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 @@ -27,6 +27,7 @@ import org.apache.streampipes.extensions.management.model.SpServiceDefinitionBui 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.messaging.nats.SpNatsProtocolFactory; import org.apache.streampipes.service.extensions.ExtensionsModelSubmitter; import org.apache.streampipes.sinks.notifications.jvm.config.ConfigKeys; import org.apache.streampipes.sinks.notifications.jvm.email.EmailSink; @@ -47,13 +48,16 @@ public class SinksNotificationsJvmInit extends ExtensionsModelSubmitter { new OneSignalSink(), new SlackNotificationSink(), new TelegramSink()) - .registerMessagingFormats(new JsonDataFormatFactory(), + .registerMessagingFormats( + new JsonDataFormatFactory(), new CborDataFormatFactory(), new SmileDataFormatFactory(), new FstDataFormatFactory()) - .registerMessagingProtocols(new SpKafkaProtocolFactory(), + .registerMessagingProtocols( + new SpKafkaProtocolFactory(), new SpJmsProtocolFactory(), - new SpMqttProtocolFactory()) + new SpMqttProtocolFactory(), + new SpNatsProtocolFactory()) .addConfig(ConfigKeys.WEBSOCKET_PROTOCOL, "ws", "") .build(); }
