This is an automated email from the ASF dual-hosted git repository.
riemer pushed a commit to branch dev
in repository
https://gitbox.apache.org/repos/asf/incubator-streampipes-extensions.git
The following commit(s) were added to refs/heads/dev by this push:
new 56aa199 Add extension to send events to NATS broker
new 82478c7 Merge pull request #44 from mohanvive/dev
56aa199 is described below
commit 56aa1998cf860c4bd2c0d313630a8ae62b5ff54d
Author: mohanadarshan <[email protected]>
AuthorDate: Fri May 7 08:21:15 2021 +0530
Add extension to send events to NATS broker
---
pom.xml | 6 +
.../extensions/all/jvm/AllExtensionsInit.java | 2 +
.../pe/jvm/AllPipelineElementsInit.java | 2 +
streampipes-sinks-brokers-jvm/pom.xml | 4 +
.../sinks/brokers/jvm/BrokersJvmInit.java | 4 +-
.../sinks/brokers/jvm/nats/NatsController.java | 122 +++++++++++++++++++++
.../sinks/brokers/jvm/nats/NatsParameters.java | 61 +++++++++++
.../sinks/brokers/jvm/nats/NatsPublisher.java | 119 ++++++++++++++++++++
.../documentation.md | 72 ++++++++++++
.../icon.png | Bin 0 -> 6986 bytes
.../strings.en | 39 +++++++
11 files changed, 430 insertions(+), 1 deletion(-)
diff --git a/pom.xml b/pom.xml
index 10a767b..418c43b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -136,6 +136,7 @@
<protobuf.version>3.6.1</protobuf.version>
<slack.version>1.2.0</slack.version>
<poiji.version>2.7.0</poiji.version>
+ <nats.version>2.11.0</nats.version>
</properties>
@@ -355,6 +356,11 @@
</exclusion>
</exclusions>
</dependency>
+ <dependency>
+ <groupId>io.nats</groupId>
+ <artifactId>jnats</artifactId>
+ <version>${nats.version}</version>
+ </dependency>
<!-- StreamPipes dependencies -->
<dependency>
diff --git
a/streampipes-extensions-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java
b/streampipes-extensions-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java
index dd67855..76c90af 100644
---
a/streampipes-extensions-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java
+++
b/streampipes-extensions-all-jvm/src/main/java/org/apache/streampipes/extensions/all/jvm/AllExtensionsInit.java
@@ -107,6 +107,7 @@ import
org.apache.streampipes.sinks.brokers.jvm.bufferrest.BufferRestController;
import org.apache.streampipes.sinks.brokers.jvm.jms.JmsController;
import org.apache.streampipes.sinks.brokers.jvm.kafka.KafkaController;
import org.apache.streampipes.sinks.brokers.jvm.mqtt.MqttPublisherSink;
+import org.apache.streampipes.sinks.brokers.jvm.nats.NatsController;
import org.apache.streampipes.sinks.brokers.jvm.websocket.WebsocketServerSink;
import org.apache.streampipes.sinks.brokers.jvm.pulsar.PulsarController;
import org.apache.streampipes.sinks.brokers.jvm.rabbitmq.RabbitMqController;
@@ -202,6 +203,7 @@ public class AllExtensionsInit extends
ExtensionsModelSubmitter {
.add(new RabbitMqController())
.add(new PulsarController())
.add(new MqttPublisherSink())
+ .add(new NatsController())
// streampipes-sinks-databases-jvm
.add(new CouchDbController())
.add(new InfluxDbController())
diff --git
a/streampipes-pipeline-elements-all-jvm/src/main/java/org/apache/streampipes/pe/jvm/AllPipelineElementsInit.java
b/streampipes-pipeline-elements-all-jvm/src/main/java/org/apache/streampipes/pe/jvm/AllPipelineElementsInit.java
index 9ff7f88..bbbb1f3 100644
---
a/streampipes-pipeline-elements-all-jvm/src/main/java/org/apache/streampipes/pe/jvm/AllPipelineElementsInit.java
+++
b/streampipes-pipeline-elements-all-jvm/src/main/java/org/apache/streampipes/pe/jvm/AllPipelineElementsInit.java
@@ -90,6 +90,7 @@ import
org.apache.streampipes.sinks.brokers.jvm.bufferrest.BufferRestController;
import org.apache.streampipes.sinks.brokers.jvm.jms.JmsController;
import org.apache.streampipes.sinks.brokers.jvm.kafka.KafkaController;
import org.apache.streampipes.sinks.brokers.jvm.mqtt.MqttPublisherSink;
+import org.apache.streampipes.sinks.brokers.jvm.nats.NatsController;
import org.apache.streampipes.sinks.brokers.jvm.pulsar.PulsarController;
import org.apache.streampipes.sinks.brokers.jvm.rabbitmq.RabbitMqController;
import org.apache.streampipes.sinks.brokers.jvm.rest.RestController;
@@ -190,6 +191,7 @@ public class AllPipelineElementsInit extends
StandaloneModelSubmitter {
.add(new PulsarController())
.add(new MqttPublisherSink())
.add(new WebsocketServerSink())
+ .add(new NatsController())
// streampipes-sinks-databases-jvm
.add(new CouchDbController())
.add(new InfluxDbController())
diff --git a/streampipes-sinks-brokers-jvm/pom.xml
b/streampipes-sinks-brokers-jvm/pom.xml
index 9dcedf6..d45c116 100644
--- a/streampipes-sinks-brokers-jvm/pom.xml
+++ b/streampipes-sinks-brokers-jvm/pom.xml
@@ -112,6 +112,10 @@
<groupId>org.java-websocket</groupId>
<artifactId>Java-WebSocket</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.nats</groupId>
+ <artifactId>jnats</artifactId>
+ </dependency>
<!-- dependency convergence -->
<dependency>
diff --git
a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/BrokersJvmInit.java
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/BrokersJvmInit.java
index 17593d1..b27d8c1 100644
---
a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/BrokersJvmInit.java
+++
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/BrokersJvmInit.java
@@ -32,6 +32,7 @@ import
org.apache.streampipes.sinks.brokers.jvm.config.BrokersJvmConfig;
import org.apache.streampipes.sinks.brokers.jvm.jms.JmsController;
import org.apache.streampipes.sinks.brokers.jvm.kafka.KafkaController;
import org.apache.streampipes.sinks.brokers.jvm.mqtt.MqttPublisherSink;
+import org.apache.streampipes.sinks.brokers.jvm.nats.NatsController;
import org.apache.streampipes.sinks.brokers.jvm.pulsar.PulsarController;
import org.apache.streampipes.sinks.brokers.jvm.rabbitmq.RabbitMqController;
import org.apache.streampipes.sinks.brokers.jvm.rest.RestController;
@@ -48,7 +49,8 @@ public class BrokersJvmInit extends StandaloneModelSubmitter {
.add(new RabbitMqController())
.add(new MqttPublisherSink())
.add(new WebsocketServerSink())
- .add(new PulsarController());
+ .add(new PulsarController())
+ .add(new NatsController());
DeclarersSingleton.getInstance().registerDataFormats(
new JsonDataFormatFactory(),
diff --git
a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/nats/NatsController.java
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/nats/NatsController.java
new file mode 100644
index 0000000..2e593ac
--- /dev/null
+++
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/nats/NatsController.java
@@ -0,0 +1,122 @@
+/*
+ * 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.brokers.jvm.nats;
+
+import org.apache.streampipes.model.graph.DataSinkDescription;
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.model.staticproperty.StaticPropertyAlternative;
+import org.apache.streampipes.sdk.StaticProperties;
+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.Alternatives;
+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 NatsController extends
StandaloneEventSinkDeclarer<NatsParameters> {
+
+ private static final String SUBJECT_KEY = "subject";
+ private static final String URLS_KEY = "natsUrls";
+
+ private static final String ACCESS_MODE = "access-mode";
+ private static final String ANONYMOUS_ACCESS = "anonymous-alternative";
+ private static final String USERNAME_ACCESS = "username-alternative";
+ private static final String USERNAME_GROUP = "username-group";
+ private static final String USERNAME_KEY = "username";
+ private static final String PASSWORD_KEY = "password";
+
+ private static final String CONNECTION_PROPERTIES =
"connection-properties";
+ private static final String NONE_PROPERTIES =
"none-properties-alternative";
+ private static final String CUSTOM_PROPERTIES =
"custom-properties-alternative";
+ private static final String CONNECTION_PROPERTIES_GROUP =
"connection-group";
+ private static final String PROPERTIES_KEY = "properties";
+
+ @Override
+ public DataSinkDescription declareModel() {
+ return
DataSinkBuilder.create("org.apache.streampipes.sinks.brokers.jvm.nats")
+ .withLocales(Locales.EN)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .requiredStream(StreamRequirementsBuilder
+ .create()
+ .requiredProperty(EpRequirements.anyProperty())
+ .build())
+ .requiredTextParameter(Labels.withId(SUBJECT_KEY), false,
false)
+ .requiredTextParameter(Labels.withId(URLS_KEY), false, false)
+ .requiredAlternatives(Labels.withId(ACCESS_MODE),
getAccessModeAlternativesOne(),
+ getAccessModeAlternativesTwo())
+ .requiredAlternatives(Labels.withId(CONNECTION_PROPERTIES),
getConnectionPropertiesAlternativesOne(),
+ getConnectionPropertiesAlternativesTwo())
+ .build();
+ }
+
+ @Override
+ public ConfiguredEventSink<NatsParameters> onInvocation(DataSinkInvocation
graph,
+
DataSinkParameterExtractor extractor) {
+
+ String subject = extractor.singleValueParameter(SUBJECT_KEY,
String.class);
+ String natsUrls = extractor.singleValueParameter(URLS_KEY,
String.class);
+ String authentication =
extractor.selectedAlternativeInternalId(ACCESS_MODE);
+ String connectionProperties =
extractor.selectedAlternativeInternalId(CONNECTION_PROPERTIES);
+ String username = null;
+ String password = null;
+ String properties = null;
+
+ if (authentication.equals(USERNAME_ACCESS)) {
+ username = extractor.singleValueParameter(USERNAME_KEY,
String.class);
+ password = extractor.secretValue(PASSWORD_KEY);
+ }
+
+ if (connectionProperties.equals(CONNECTION_PROPERTIES)) {
+ properties = extractor.singleValueParameter(PROPERTIES_KEY,
String.class);
+ }
+
+ NatsParameters params = new NatsParameters(graph, natsUrls, subject,
username, password, properties);
+
+ return new ConfiguredEventSink<>(params, NatsPublisher::new);
+ }
+
+ public static StaticPropertyAlternative getAccessModeAlternativesOne() {
+ return Alternatives.from(Labels.withId(ANONYMOUS_ACCESS));
+
+ }
+
+ public static StaticPropertyAlternative getAccessModeAlternativesTwo() {
+ return Alternatives.from(Labels.withId(USERNAME_ACCESS),
+ StaticProperties.group(Labels.withId(USERNAME_GROUP),
+
StaticProperties.stringFreeTextProperty(Labels.withId(USERNAME_KEY)),
+
StaticProperties.secretValue(Labels.withId(PASSWORD_KEY))));
+
+ }
+
+ public static StaticPropertyAlternative
getConnectionPropertiesAlternativesOne() {
+ return Alternatives.from(Labels.withId(NONE_PROPERTIES));
+
+ }
+
+ public static StaticPropertyAlternative
getConnectionPropertiesAlternativesTwo() {
+ return Alternatives.from(Labels.withId(CUSTOM_PROPERTIES),
+
StaticProperties.group(Labels.withId(CONNECTION_PROPERTIES_GROUP),
+
StaticProperties.stringFreeTextProperty(Labels.withId(PROPERTIES_KEY))));
+
+ }
+}
diff --git
a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/nats/NatsParameters.java
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/nats/NatsParameters.java
new file mode 100644
index 0000000..50f03fc
--- /dev/null
+++
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/nats/NatsParameters.java
@@ -0,0 +1,61 @@
+/*
+ * 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.brokers.jvm.nats;
+
+import org.apache.streampipes.model.graph.DataSinkInvocation;
+import org.apache.streampipes.wrapper.params.binding.EventSinkBindingParams;
+
+public class NatsParameters extends EventSinkBindingParams {
+
+ private String natsUrls;
+ private String subject;
+ private String username;
+ private String password;
+ private String properties;
+
+ public NatsParameters(DataSinkInvocation graph, String natsUrls, String
subject, String username, String password,
+ String properties) {
+ super(graph);
+ this.natsUrls = natsUrls;
+ this.subject = subject;
+ this.username = username;
+ this.password = password;
+ this.properties = properties;
+ }
+
+ public String getNatsUrls() {
+ return natsUrls;
+ }
+
+ public String getSubject() {
+ return subject;
+ }
+
+ public String getUsername() {
+ return username;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ public String getProperties() {
+ return properties;
+ }
+}
diff --git
a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/nats/NatsPublisher.java
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/nats/NatsPublisher.java
new file mode 100644
index 0000000..104e313
--- /dev/null
+++
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/nats/NatsPublisher.java
@@ -0,0 +1,119 @@
+/*
+ * 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.brokers.jvm.nats;
+
+import io.nats.client.Connection;
+import io.nats.client.Nats;
+import io.nats.client.Options;
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
+import org.apache.streampipes.logging.api.Logger;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
+import org.apache.streampipes.wrapper.runtime.EventSink;
+
+import java.time.Duration;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.TimeoutException;
+
+public class NatsPublisher implements EventSink<NatsParameters> {
+
+ private String subject;
+ private Connection natsConnection;
+ private JsonDataFormatDefinition dataFormatDefinition;
+ private static Logger LOG;
+
+ public NatsPublisher() {
+ this.dataFormatDefinition = new JsonDataFormatDefinition();
+ }
+
+ @Override
+ public void onInvocation(NatsParameters parameters,
EventSinkRuntimeContext runtimeContext)
+ throws SpRuntimeException {
+
+ LOG = parameters.getGraph().getLogger(NatsPublisher.class);
+ this.subject = parameters.getSubject();
+ String natsUrls = parameters.getNatsUrls();
+ String propertiesAsString = parameters.getProperties();
+ String username = parameters.getUsername();
+ String password = parameters.getPassword();
+
+ Properties props = new Properties();
+
+ if (username != null) {
+ props.setProperty(Options.PROP_USERNAME, username);
+ props.setProperty(Options.PROP_PASSWORD, password);
+ }
+
+ if (propertiesAsString != null && !propertiesAsString.isEmpty()) {
+ splitNatsProperties(propertiesAsString, props);
+ }
+
+ String[] natsServerUrls = natsUrls.split(",");
+ Options options;
+ if (natsServerUrls.length > 1) {
+ options = new
Options.Builder(props).servers(natsServerUrls).build();
+ } else {
+ options = new Options.Builder(props).server(natsUrls).build();
+ }
+
+ try {
+ this.natsConnection = Nats.connect(options);
+ } catch (Exception e) {
+ LOG.error("Error when connecting to the Nats broker on " +
natsUrls + " . " + e.toString());
+ }
+ }
+
+ @Override
+ public void onEvent(Event inputEvent) {
+ try {
+ Map<String, Object> event = inputEvent.getRaw();
+ natsConnection.publish(subject,
dataFormatDefinition.fromMap(event));
+ } catch (SpRuntimeException e) {
+ LOG.error("Could not publish events to Nats broker. " +
e.toString());
+ }
+ }
+
+ @Override
+ public void onDetach() throws SpRuntimeException {
+ try {
+ natsConnection.flush(Duration.ofMillis(50));
+ natsConnection.close();
+ } catch (TimeoutException | InterruptedException e) {
+ LOG.error("Error when disconnecting with Nats broker. " +
e.toString());
+ }
+ }
+
+ private void splitNatsProperties(String propertiesAsString, Properties
properties) {
+
+ String[] optionalProperties = propertiesAsString.split(",");
+ if (optionalProperties.length > 0) {
+ for (String header : optionalProperties) {
+ try {
+ String[] configPropertyWithValue = header.split(":", 2);
+ properties.setProperty(configPropertyWithValue[0].trim(),
configPropertyWithValue[1].trim());
+ } catch (Exception e) {
+ LOG.warn("Optional property '" + header + "' is not
defined in the correct format.");
+ }
+ }
+ }
+ }
+
+}
diff --git
a/streampipes-sinks-brokers-jvm/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.nats/documentation.md
b/streampipes-sinks-brokers-jvm/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.nats/documentation.md
new file mode 100644
index 0000000..e6652b7
--- /dev/null
+++
b/streampipes-sinks-brokers-jvm/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.nats/documentation.md
@@ -0,0 +1,72 @@
+<!--
+ ~ 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.
+ ~
+ -->
+
+## Nats Publisher
+
+<p align="center">
+ <img src="icon.png" width="150px;" class="pe-image-documentation"/>
+</p>
+
+***
+
+## Description
+
+Publishes events to NATS broker.
+
+***
+
+## Required input
+
+This sink does not have any requirements and works with any incoming event
type.
+
+***
+
+## Configuration
+
+### NATS Subject
+
+The subject (topic) where events should be sent to.
+
+### NATS Broker URL
+
+The URL to connect to the NATS broker. It can be provided multiple urls
separated by commas(,).
+ (e.g., nats://localhost:4222,nats://localhost:4223)
+
+### Username
+
+The username to authenticate the client with NATS broker.
+
+It is an optional configuration.
+
+### NATS Broker URL
+
+The password to authenticate the client with NATS broker.
+
+It is an optional configuration.
+
+### NATS Connection Properties
+
+All other possible connection configurations that the nats client can be
created with.
+It can be provided as key value pairs separated by colons(:) and commas(,).
+ (e.g., io.nats.client.reconnect.max:1, io.nats.client.timeout:1000)
+
+It is an optional configuration.
+
+## Output
+
+(not applicable for data sinks)
\ No newline at end of file
diff --git
a/streampipes-sinks-brokers-jvm/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.nats/icon.png
b/streampipes-sinks-brokers-jvm/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.nats/icon.png
new file mode 100644
index 0000000..a016ca8
Binary files /dev/null and
b/streampipes-sinks-brokers-jvm/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.nats/icon.png
differ
diff --git
a/streampipes-sinks-brokers-jvm/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.nats/strings.en
b/streampipes-sinks-brokers-jvm/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.nats/strings.en
new file mode 100644
index 0000000..89767d7
--- /dev/null
+++
b/streampipes-sinks-brokers-jvm/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.nats/strings.en
@@ -0,0 +1,39 @@
+org.apache.streampipes.sinks.brokers.jvm.nats.title=NATS Publisher
+org.apache.streampipes.sinks.brokers.jvm.nats.description=Publish events to
NATS broker
+
+subject.title=NATS Subject
+subject.description=Select a NATS subject to publish events
+
+natsUrls.title=NATS broker URLs
+natsUrls.description=Provide NATS server URLs (can be provided multiple urls
separated by commas).
+
+username.title=Username
+username.description=The username to authenticate with the broker
+
+password.title=Password
+password.description=The password to authenticate with the broker
+
+access-mode.title=Access Mode
+access-mode.description=Unauthenticated or SASL/PLAIN
+
+anonymous-alternative.title=Unauthenticated
+anonymous-alternative.description=No authentication and plaintext
+
+username-alternative.title=SASL/PLAIN
+username-alternative.description=Username and password, no encryption
+
+username-group.title=Username and password
+
+properties.title=NATS connection properties
+properties.description=Other possible connection configuration (key-value
pairs separated by colons and commas)
+
+connection-properties.title=Client Connection Properties
+connection-properties.description=Optional client connection configuration
+
+none-properties-alternative.title=None
+none-properties-alternative.description=No optional client properties
+
+custom-properties-alternative.title=Custom Properties
+custom-properties-alternative.description=Custom NATS connection properties
+
+connection-group.title=NATS connection optional properties
