This is an automated email from the ASF dual-hosted git repository.
wiener 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 43df359 [STREAMPIPES-299] add advanced configuration options for mqtt
publisher
new da1afa4 Merge branch 'dev' of
https://github.com/apache/incubator-streampipes-extensions into dev
43df359 is described below
commit 43df359cccd1cc436fb669e2d374b2aa66053743
Author: Patrick Wiener <[email protected]>
AuthorDate: Sun Feb 14 12:41:22 2021 +0100
[STREAMPIPES-299] add advanced configuration options for mqtt publisher
---
.../extensions/all/jvm/AllExtensionsInit.java | 4 +-
.../pe/jvm/AllPipelineElementsInit.java | 4 +-
.../sinks/brokers/jvm/BrokersJvmInit.java | 4 +-
.../sinks/brokers/jvm/mqtt/MqttClient.java | 163 +++++++++++++++++++++
.../sinks/brokers/jvm/mqtt/MqttController.java | 78 ----------
.../sinks/brokers/jvm/mqtt/MqttParameters.java | 48 ------
.../sinks/brokers/jvm/mqtt/MqttPublisher.java | 86 -----------
.../sinks/brokers/jvm/mqtt/MqttPublisherSink.java | 149 +++++++++++++++++++
.../sinks/brokers/jvm/mqtt/common/MqttOptions.java | 162 ++++++++++++++++++++
.../sinks/brokers/jvm/mqtt/common/MqttUtils.java | 68 +++++++++
.../documentation.md | 2 +-
.../strings.en | 77 ++++++++--
12 files changed, 616 insertions(+), 229 deletions(-)
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 9b075cf..dd67855 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
@@ -106,8 +106,8 @@ import
org.apache.streampipes.processors.transformation.jvm.processor.value.dura
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.websocket.WebsocketServerSink;
-import org.apache.streampipes.sinks.brokers.jvm.mqtt.MqttController;
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;
@@ -201,7 +201,7 @@ public class AllExtensionsInit extends
ExtensionsModelSubmitter {
.add(new BufferRestController())
.add(new RabbitMqController())
.add(new PulsarController())
- .add(new MqttController())
+ .add(new MqttPublisherSink())
// 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 0c104dc..efa6f8f 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
@@ -86,7 +86,7 @@ import
org.apache.streampipes.processors.transformation.jvm.processor.value.dura
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.MqttController;
+import org.apache.streampipes.sinks.brokers.jvm.mqtt.MqttPublisherSink;
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;
@@ -183,7 +183,7 @@ public class AllPipelineElementsInit extends
StandaloneModelSubmitter {
.add(new BufferRestController())
.add(new RabbitMqController())
.add(new PulsarController())
- .add(new MqttController())
+ .add(new MqttPublisherSink())
.add(new WebsocketServerSink())
// streampipes-sinks-databases-jvm
.add(new CouchDbController())
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 439b30a..17593d1 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
@@ -31,7 +31,7 @@ import
org.apache.streampipes.sinks.brokers.jvm.bufferrest.BufferRestController;
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.MqttController;
+import org.apache.streampipes.sinks.brokers.jvm.mqtt.MqttPublisherSink;
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;
@@ -46,7 +46,7 @@ public class BrokersJvmInit extends StandaloneModelSubmitter {
.add(new RestController())
.add(new BufferRestController())
.add(new RabbitMqController())
- .add(new MqttController())
+ .add(new MqttPublisherSink())
.add(new WebsocketServerSink())
.add(new PulsarController());
diff --git
a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/MqttClient.java
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/MqttClient.java
new file mode 100644
index 0000000..c5bb124
--- /dev/null
+++
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/MqttClient.java
@@ -0,0 +1,163 @@
+/*
+ * 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.mqtt;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.sinks.brokers.jvm.mqtt.common.MqttOptions;
+import org.apache.streampipes.sinks.brokers.jvm.mqtt.common.MqttUtils;
+import org.apache.streampipes.wrapper.standalone.SinkParams;
+import org.fusesource.mqtt.client.BlockingConnection;
+import org.fusesource.mqtt.client.MQTT;
+
+import java.net.URI;
+
+public class MqttClient {
+
+ private final MqttOptions options;
+ private URI uri;
+ private MQTT mqtt;
+ private BlockingConnection conn;
+
+ public MqttClient(SinkParams params) {
+ this.options = new MqttOptions(params);
+ this.createMqttClient();
+ }
+
+ /**
+ * Create new MQTT client
+ */
+ public void createMqttClient() {
+ this.mqtt = new MQTT();
+ this.uri = MqttUtils.makeMqttServerUri(options.getProtocol(),
options.getHost(), options.getPort());
+ try {
+ /**
+ * Sets the url for connecting to the MQTT broker, e.g. {@code:
tcp://localhost:1883}.
+ */
+ mqtt.setHost(uri);
+
+ // authentication
+ if (options.isBasicAuth()) {
+ /**
+ * The username for authenticated sessions.
+ */
+ mqtt.setUserName(options.getUsername());
+ /**
+ * The password for authenticated sessions.
+ */
+ mqtt.setPassword(options.getPassword());
+ }
+
+ /**
+ * The client id used when connecting to the MQTT broker.
+ */
+ mqtt.setClientId(options.getClientId());
+
+ /**
+ * Set to false if you want the MQTT server to persist topic
subscriptions and ack positions across
+ * client sessions. Defaults to true.
+ */
+ mqtt.setCleanSession(options.isCleanSession());
+
+ /**
+ * The maximum amount of time in ms to wait between reconnect
attempts. Defaults to 30,000.
+ */
+ mqtt.setReconnectDelayMax(options.getReconnectDelayMaxInMs());
+
+ /**
+ * Configures the Keep Alive timer in seconds. Defines the maximum
time interval between messages
+ * received from a client. It enables the server to detect that
the network connection to a client has
+ * dropped, without having to wait for the long TCP/IP timeout.
+ */
+ mqtt.setKeepAlive(options.getKeepAliveInSec());
+
+ /**
+ * Set to "3.1.1" to use MQTT version 3.1.1. Otherwise defaults to
the 3.1 protocol version.
+ */
+ mqtt.setVersion(options.getMqttProtocolVersion());
+
+ // last will and testament options
+ if (options.isLastWill()) {
+ /**
+ * If set the server will publish the client's Will message to
the specified topics if the client has
+ * an unexpected disconnection.
+ */
+ mqtt.setWillTopic(options.getWillTopic());
+
+ /**
+ * Sets the quality of service to use for the Will message.
Defaults to QoS.AT_MOST_ONCE.
+ */
+ mqtt.setWillQos(options.getWillQoS());
+
+ /**
+ * The Will message to send. Defaults to a zero length message.
+ */
+ mqtt.setWillMessage(options.getWillMessage());
+
+ /**
+ * Set to true if you want the Will to be published with the
retain option.
+ */
+ mqtt.setWillRetain(options.getWillRetain());
+ }
+ } catch (Exception e) {
+ throw new SpRuntimeException("Failed to initialize MQTT Client: "
+ e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Start blocking connection to MQTT broker.
+ */
+ public void connect() {
+ try {
+ this.conn = mqtt.blockingConnection();
+ this.conn.connect();
+ } catch (Exception e) {
+ throw new SpRuntimeException("Could not connect to MQTT broker: "
+ + uri.toString() + ", " + e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Publish received event to MQTT broker.
+ * @param event event to be published
+ */
+ public void publish(Event event) {
+ JsonDataFormatDefinition dataFormatDefinition = new
JsonDataFormatDefinition();
+ byte[] payload = new
String(dataFormatDefinition.fromMap(event.getRaw())).getBytes();
+ try {
+ this.conn.publish(options.getTopic(), payload, options.getQos(),
options.isRetain());
+ } catch (Exception e) {
+ throw new SpRuntimeException("Could not publish to MQTT broker: "
+ + uri.toString() + ", " + e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Disconnect from MQTT broker.
+ */
+ public void disconnect() {
+ try {
+ this.conn.disconnect();
+ } catch (Exception e) {
+ throw new SpRuntimeException("Could not disconnect from MQTT
broker: "
+ + uri.toString() + ", " + e.getMessage(), e);
+ }
+ }
+
+}
diff --git
a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/MqttController.java
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/MqttController.java
deleted file mode 100644
index 9976572..0000000
---
a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/MqttController.java
+++ /dev/null
@@ -1,78 +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.brokers.jvm.mqtt;
-
-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.OntologyProperties;
-import org.apache.streampipes.sdk.utils.Assets;
-import org.apache.streampipes.wrapper.standalone.ConfiguredEventSink;
-import
org.apache.streampipes.wrapper.standalone.declarer.StandaloneEventSinkDeclarer;
-
-public class MqttController extends
StandaloneEventSinkDeclarer<MqttParameters> {
-
-// private static final String MQTT_BROKER_SETTINGS_KEY = "broker-settings";
- private static final String TOPIC_KEY = "topic";
- private static final String HOST_KEY = "host";
- private static final String PORT_KEY = "port";
-
-// private static final String MQTT_HOST_URI = "http://schema.org/mqttHost";;
-// private static final String MQTT_PORT_URI = "http://schema.org/mqttPort";;
-
- @Override
- public DataSinkDescription declareModel() {
- return
DataSinkBuilder.create("org.apache.streampipes.sinks.brokers.jvm.mqtt")
- .withLocales(Locales.EN)
- .withAssets(Assets.DOCUMENTATION, Assets.ICON)
- .requiredStream(StreamRequirementsBuilder
- .create()
- .requiredProperty(EpRequirements.anyProperty())
- .build())
- .requiredTextParameter(Labels.withId(TOPIC_KEY), false, false)
- .requiredTextParameter(Labels.withId(HOST_KEY), false, false)
- .requiredIntegerParameter(Labels.withId(PORT_KEY), 1883)
-//
.requiredOntologyConcept(Labels.withId(MQTT_BROKER_SETTINGS_KEY),,
-// OntologyProperties.mandatory(MQTT_HOST_URI),
-// OntologyProperties.mandatory(MQTT_PORT_URI))
- .build();
- }
-
- @Override
- public ConfiguredEventSink<MqttParameters> onInvocation(DataSinkInvocation
graph,
-
DataSinkParameterExtractor extractor) {
-
- String topic = extractor.singleValueParameter(TOPIC_KEY, String.class);
- String mqttHost = extractor.singleValueParameter(HOST_KEY,
String.class);
- Integer mqttPort = extractor.singleValueParameter(PORT_KEY,
Integer.class);
-// String mqttHost =
extractor.supportedOntologyPropertyValue(MQTT_BROKER_SETTINGS_KEY,
MQTT_HOST_URI,
-// String.class);
-// Integer mqttPort =
extractor.supportedOntologyPropertyValue(MQTT_BROKER_SETTINGS_KEY,
MQTT_PORT_URI,
-// Integer.class);
-
- MqttParameters params = new MqttParameters(graph, mqttHost, mqttPort,
topic);
- return new ConfiguredEventSink<>(params, MqttPublisher::new);
- }
-
-}
diff --git
a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/MqttParameters.java
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/MqttParameters.java
deleted file mode 100644
index 216980b..0000000
---
a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/MqttParameters.java
+++ /dev/null
@@ -1,48 +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.brokers.jvm.mqtt;
-
-import org.apache.streampipes.model.graph.DataSinkInvocation;
-import org.apache.streampipes.wrapper.params.binding.EventSinkBindingParams;
-
-public class MqttParameters extends EventSinkBindingParams {
-
- private String mqttHost;
- private Integer mqttPort;
- private String topic;
-
- public MqttParameters(DataSinkInvocation graph, String mqttHost, Integer
mqttPort, String topic) {
- super(graph);
- this.mqttHost = mqttHost;
- this.mqttPort = mqttPort;
- this.topic = topic;
- }
-
- public String getMqttHost() {
- return mqttHost;
- }
-
- public Integer getMqttPort() {
- return mqttPort;
- }
-
- public String getTopic() {
- return topic;
- }
-}
\ No newline at end of file
diff --git
a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/MqttPublisher.java
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/MqttPublisher.java
deleted file mode 100644
index 02861d5..0000000
---
a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/MqttPublisher.java
+++ /dev/null
@@ -1,86 +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.brokers.jvm.mqtt;
-
-import org.apache.streampipes.commons.exceptions.SpRuntimeException;
-import org.apache.streampipes.dataformat.json.JsonDataFormatDefinition;
-import org.apache.streampipes.model.runtime.Event;
-import org.apache.streampipes.wrapper.context.EventSinkRuntimeContext;
-import org.apache.streampipes.wrapper.runtime.EventSink;
-import org.fusesource.mqtt.client.*;
-
-import java.net.URISyntaxException;
-import java.util.Map;
-
-public class MqttPublisher implements EventSink<MqttParameters> {
-
- private MQTT client;
- private String topic;
- private BlockingConnection connection;
- private JsonDataFormatDefinition dataFormatDefinition;
- private MqttParameters params;
-
- public MqttPublisher() {
- this.dataFormatDefinition = new JsonDataFormatDefinition();
- }
-
- @Override
- public void onInvocation(MqttParameters params, EventSinkRuntimeContext
runtimeContext) throws SpRuntimeException {
- this.client = new MQTT();
- this.topic = params.getTopic();
- this.params = params;
-
- try {
- client.setHost(params.getMqttHost() + ":" + params.getMqttPort());
- this.connection = client.blockingConnection();
- this.connection.connect();
- } catch (URISyntaxException e) {
- e.printStackTrace();
- } catch (Exception e) {
- e.printStackTrace();
- }
-
- if (!this.connection.isConnected()) {
- throw new SpRuntimeException("Could not connect to MQTT server: "
+ params.getMqttHost() + " on Port: " + params.getMqttPort() + " to topic: " +
params.getTopic());
- }
- }
-
- @Override
- public void onEvent(Event inputEvent) {
-
- Map<String, Object> event = inputEvent.getRaw();
-
- try {
- this.connection.publish(this.topic, new
String(dataFormatDefinition.fromMap(event)).getBytes(), QoS.AT_LEAST_ONCE,
false);
- } catch (SpRuntimeException e) {
- e.printStackTrace();
- } catch (Exception e) {
- e.printStackTrace();
- }
- }
-
- @Override
- public void onDetach() throws SpRuntimeException {
- try {
- this.connection.disconnect();
- } catch (Exception e) {
- throw new SpRuntimeException("Could not disconnect from MQTT
server: " + params.getMqttHost() + " on Port: " + params.getMqttPort() + " to
topic: " + params.getTopic(), e);
- }
- }
-}
\ No newline at end of file
diff --git
a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/MqttPublisherSink.java
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/MqttPublisherSink.java
new file mode 100644
index 0000000..e817dfb
--- /dev/null
+++
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/MqttPublisherSink.java
@@ -0,0 +1,149 @@
+/*
+ * 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.mqtt;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.apache.streampipes.model.graph.DataSinkDescription;
+import org.apache.streampipes.model.runtime.Event;
+import org.apache.streampipes.model.staticproperty.Option;
+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.helpers.*;
+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 java.util.Arrays;
+
+public class MqttPublisherSink extends StreamPipesDataSink {
+
+ private static final int DEFAULT_MQTT_PORT = 1883;
+ private static final int DEFAULT_RECONNECT_PERIOD = 30;
+ private static final int DEFAULT_KEEP_ALIVE = 30;
+
+ public static final String TOPIC = "topic";
+ public static final String HOST = "host";
+ public static final String PORT = "port";
+ public static final String AUTH_MODE = "auth-mode";
+ public static final String NO_AUTH_ALTERNATIVE = "no-auth-alternative";
+ public static final String AUTH_ALTERNATIVE = "basic-auth-alternative";
+ public static final String USERNAME_GROUP = "username-group";
+ public static final String USERNAME = "username";
+ public static final String PASSWORD = "password";
+ public static final String QOS_LEVEL_KEY = "qos-level";
+ public static final String CLEAN_SESSION_KEY = "clean-session";
+ public static final String WILL_RETAIN = "will-retain";
+ public static final String ENCRYPTION_MODE = "encryption-mode";
+ public static final String RECONNECT_PERIOD_IN_SEC = "reconnect-period";
+ public static final String WILL_MODE = "lwt-mode";
+ public static final String NO_WILL_ALTERNATIVE = "no-lwt-alternative";
+ public static final String WILL_ALTERNATIVE = "lwt-alternative";
+ public static final String WILL_GROUP = "lwt-group";
+ public static final String WILL_TOPIC = "lwt-topic";
+ public static final String WILL_MESSAGE = "lwt-message";
+ public static final String WILL_QOS = "lwt-qos-level";
+ public static final String RETAIN = "retain";
+ public static final String KEEP_ALIVE_IN_SEC = "keep-alive";
+ public static final String MQTT_COMPLIANT = "mqtt-version-compliant";
+
+ private MqttClient mqttClient;
+
+ @Override
+ public DataSinkDescription declareModel() {
+ return
DataSinkBuilder.create("org.apache.streampipes.sinks.brokers.jvm.mqtt")
+ .withLocales(Locales.EN)
+ .withAssets(Assets.DOCUMENTATION, Assets.ICON)
+ .requiredStream(StreamRequirementsBuilder.any())
+ .requiredTextParameter(Labels.withId(TOPIC))
+ .requiredTextParameter(Labels.withId(HOST))
+ .requiredIntegerParameter(Labels.withId(PORT),
DEFAULT_MQTT_PORT)
+ .requiredAlternatives(
+ Labels.withId(AUTH_MODE),
+ Alternatives.from(Labels.withId(NO_AUTH_ALTERNATIVE),
true),
+ Alternatives.from(Labels.withId(AUTH_ALTERNATIVE),
+
StaticProperties.group(Labels.withId(USERNAME_GROUP),
+
StaticProperties.stringFreeTextProperty(Labels.withId(USERNAME)),
+
StaticProperties.secretValue(Labels.withId(PASSWORD)))))
+ .requiredSingleValueSelection(
+ Labels.withId(ENCRYPTION_MODE),
+ Arrays.asList(
+ new Option("TCP", true),
+ // SSL not yet supported
+ new Option("SSL/TLS", false)))
+ .requiredSingleValueSelection(
+ Labels.withId(QOS_LEVEL_KEY),
+ Arrays.asList(
+ new Option("0 - at-most-once", false),
+ new Option("1 - at-least-once", true),
+ new Option("2 - exactly-once", false)))
+ .requiredSingleValueSelection(
+ Labels.withId(RETAIN),
+ Arrays.asList(
+ new Option("Yes", false),
+ new Option("No", true)))
+ .requiredSingleValueSelection(
+ Labels.withId(CLEAN_SESSION_KEY),
+ Arrays.asList(
+ new Option("Yes", true),
+ new Option("No", false)))
+
.requiredIntegerParameter(Labels.withId(RECONNECT_PERIOD_IN_SEC),
DEFAULT_RECONNECT_PERIOD)
+ .requiredIntegerParameter(Labels.withId(KEEP_ALIVE_IN_SEC),
DEFAULT_KEEP_ALIVE)
+ .requiredSingleValueSelection(
+ Labels.withId(MQTT_COMPLIANT),
+ Arrays.asList(
+ new Option("Yes", true),
+ new Option("No", false)))
+ .requiredAlternatives(
+ Labels.withId(WILL_MODE),
+ Alternatives.from(Labels.withId(NO_WILL_ALTERNATIVE),
true),
+ Alternatives.from(Labels.withId(WILL_ALTERNATIVE),
+
StaticProperties.group(Labels.withId(WILL_GROUP),
+
StaticProperties.stringFreeTextProperty(Labels.withId(WILL_TOPIC)),
+
StaticProperties.stringFreeTextProperty(Labels.withId(WILL_MESSAGE)),
+
StaticProperties.singleValueSelection(Labels.withId(WILL_RETAIN),
+ Arrays.asList(
+ new Option("Yes",
false),
+ new Option("No",
true))),
+ StaticProperties.singleValueSelection(
+ Labels.withId(WILL_QOS),
+ Arrays.asList(
+ new Option("0 -
at-most-once", true),
+ new Option("1 -
at-least-once", false),
+ new Option("2 -
exactly-once", false))))))
+ .build();
+ }
+
+
+ @Override
+ public void onInvocation(SinkParams params, EventSinkRuntimeContext
runtimeContext) throws SpRuntimeException {
+ this.mqttClient = new MqttClient(params);
+ this.mqttClient.connect();
+ }
+
+ @Override
+ public void onEvent(Event event) throws SpRuntimeException {
+ this.mqttClient.publish(event);
+ }
+
+ @Override
+ public void onDetach() throws SpRuntimeException {
+ this.mqttClient.disconnect();
+ }
+}
diff --git
a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/common/MqttOptions.java
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/common/MqttOptions.java
new file mode 100644
index 0000000..d7e8d32
--- /dev/null
+++
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/common/MqttOptions.java
@@ -0,0 +1,162 @@
+/*
+ * 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.mqtt.common;
+
+import org.apache.streampipes.sdk.extractor.DataSinkParameterExtractor;
+import org.apache.streampipes.wrapper.standalone.SinkParams;
+import org.fusesource.mqtt.client.QoS;
+
+import static
org.apache.streampipes.sinks.brokers.jvm.mqtt.MqttPublisherSink.*;
+
+public class MqttOptions {
+
+ private final String clientId;
+ private final String host;
+ private final int port;
+ private final String topic;
+ private final String protocol;
+ private final QoS qos;
+ private final long reconnectDelayMaxInMs;
+ private final boolean cleanSession;
+ private final boolean retain;
+ private final short keepAliveInSec;
+
+ private String username = "";
+ private String password = "";
+ private boolean isBasicAuth = false;
+ private boolean isLastWill = false;
+ private QoS willQoS = QoS.AT_MOST_ONCE;
+ private Boolean willRetain = false;
+ private String willTopic = "";
+ private String willMessage = "";
+ private String mqttProtocolVersion = "3.1";
+
+ public MqttOptions(SinkParams params) {
+ DataSinkParameterExtractor extract = params.extractor();
+
+ this.clientId =
MqttUtils.runningInstanceId(params.getGraph().getElementId());
+ this.topic = extract.singleValueParameter(TOPIC, String.class);
+ this.host = extract.singleValueParameter(HOST, String.class);
+ this.port = extract.singleValueParameter(PORT, Integer.class);
+ this.protocol = extract.selectedSingleValue(ENCRYPTION_MODE,
String.class);
+
+ this.qos =
MqttUtils.extractQoSFromString(extract.selectedSingleValue(QOS_LEVEL_KEY,
String.class));
+ this.reconnectDelayMaxInMs =
+
MqttUtils.fromSecToMs(extract.singleValueParameter(RECONNECT_PERIOD_IN_SEC,
Long.class));
+ this.keepAliveInSec = extract.singleValueParameter(KEEP_ALIVE_IN_SEC,
Short.class);
+ this.cleanSession =
MqttUtils.extractBoolean(extract.selectedSingleValue(CLEAN_SESSION_KEY,
String.class));
+ this.retain =
MqttUtils.extractBoolean(extract.selectedSingleValue(RETAIN, String.class));
+
+ boolean isCompliant =
MqttUtils.extractBoolean(extract.selectedSingleValue(MQTT_COMPLIANT,
String.class));
+ if (isCompliant) {
+ this.mqttProtocolVersion = "3.1.1";
+ }
+
+ String accessMode = extract.selectedAlternativeInternalId(AUTH_MODE);
+ if (accessMode.equals(AUTH_ALTERNATIVE)) {
+ this.isBasicAuth = true;
+ this.username = extract.singleValueParameter(USERNAME,
String.class);
+ this.password = extract.secretValue(PASSWORD);
+ }
+
+ String willMode = extract.selectedAlternativeInternalId(WILL_MODE);
+ if (willMode.equals(WILL_ALTERNATIVE)) {
+ this.isLastWill = true;
+ this.willTopic = extract.singleValueParameter(WILL_TOPIC,
String.class);
+ this.willMessage = extract.singleValueParameter(WILL_MESSAGE,
String.class);
+ this.willQoS =
MqttUtils.extractQoSFromString(extract.selectedSingleValue(WILL_QOS,
String.class));
+ this.willRetain =
MqttUtils.extractBoolean(extract.selectedSingleValue(WILL_RETAIN,
String.class));
+ }
+ }
+
+ public String getClientId() {
+ return clientId;
+ }
+
+ public String getHost() {
+ return host;
+ }
+
+ public int getPort() {
+ return port;
+ }
+
+ public String getTopic() {
+ return topic;
+ }
+
+ public String getProtocol() {
+ return protocol;
+ }
+
+ public QoS getQos() {
+ return qos;
+ }
+
+ public long getReconnectDelayMaxInMs() {
+ return reconnectDelayMaxInMs;
+ }
+
+ public boolean isCleanSession() {
+ return cleanSession;
+ }
+
+ public boolean isRetain() {
+ return retain;
+ }
+
+ public short getKeepAliveInSec() {
+ return keepAliveInSec;
+ }
+
+ public String getUsername() {
+ return username;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ public boolean isBasicAuth() {
+ return isBasicAuth;
+ }
+
+ public boolean isLastWill() {
+ return isLastWill;
+ }
+
+ public QoS getWillQoS() {
+ return willQoS;
+ }
+
+ public Boolean getWillRetain() {
+ return willRetain;
+ }
+
+ public String getWillTopic() {
+ return willTopic;
+ }
+
+ public String getWillMessage() {
+ return willMessage;
+ }
+
+ public String getMqttProtocolVersion() {
+ return mqttProtocolVersion;
+ }
+}
diff --git
a/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/common/MqttUtils.java
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/common/MqttUtils.java
new file mode 100644
index 0000000..2d31dd4
--- /dev/null
+++
b/streampipes-sinks-brokers-jvm/src/main/java/org/apache/streampipes/sinks/brokers/jvm/mqtt/common/MqttUtils.java
@@ -0,0 +1,68 @@
+/*
+ * 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.mqtt.common;
+
+import org.apache.streampipes.commons.exceptions.SpRuntimeException;
+import org.fusesource.mqtt.client.QoS;
+
+import java.net.URI;
+
+public class MqttUtils {
+
+ private static final String TCP = "TCP";
+ private static final String SSL_TLS = "SSL/TLS";
+ private static final String TCP_PROTOCOL = "tcp://";
+ private static final String SSL_PROTOCOL = "ssl://";
+ private static final String COLON = ":";
+
+ public MqttUtils() {
+ }
+
+ // remove non-digits
+ public static QoS extractQoSFromString(String s) {
+ int qos = Integer.parseInt(s.replaceAll("\\D+",""));
+ switch(qos) {
+ case 0: return QoS.AT_MOST_ONCE;
+ case 1: return QoS.AT_LEAST_ONCE;
+ case 2: return QoS.EXACTLY_ONCE;
+ } throw new SpRuntimeException("Could not retrieve QoS level: QoS " +
qos);
+ }
+
+ public static String runningInstanceId(String elementId) {
+ return elementId.substring(elementId.lastIndexOf(".") + 1);
+ }
+
+ public static URI makeMqttServerUri(String protocol, String host, int
port) {
+ if (protocol.contains(TCP)) {
+ return URI.create(TCP_PROTOCOL + host + COLON + port);
+ } else if (protocol.contains(SSL_TLS)) {
+ return URI.create(SSL_PROTOCOL + host + COLON + port);
+ } throw new SpRuntimeException("Connection protocol not supported! Use
tcp:// or ssl://");
+ }
+
+ public static boolean extractBoolean(String s) {
+ switch(s) {
+ case "Yes": return true;
+ case "No": return false;
+ } throw new SpRuntimeException("Could not map string value to boolean:
" + s);
+ }
+
+ public static long fromSecToMs(Long value) {
+ return value * 1000;
+ }
+}
diff --git
a/streampipes-sinks-brokers-jvm/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.mqtt/documentation.md
b/streampipes-sinks-brokers-jvm/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.mqtt/documentation.md
index adb8e31..cadccdb 100644
---
a/streampipes-sinks-brokers-jvm/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.mqtt/documentation.md
+++
b/streampipes-sinks-brokers-jvm/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.mqtt/documentation.md
@@ -42,7 +42,7 @@ This sink does not have any requirements and works with any
incoming event type.
The basic settings to connect to the broker.
The MQTT broker URL indicates the URL of the broker (e.g., localhost), the
port indicates the port of the broker
-(e.g., 9092)
+(e.g., 1883)
### MQTT Topic
diff --git
a/streampipes-sinks-brokers-jvm/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.mqtt/strings.en
b/streampipes-sinks-brokers-jvm/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.mqtt/strings.en
index e740419..d49f9f1 100644
---
a/streampipes-sinks-brokers-jvm/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.mqtt/strings.en
+++
b/streampipes-sinks-brokers-jvm/src/main/resources/org.apache.streampipes.sinks.brokers.jvm.mqtt/strings.en
@@ -1,15 +1,72 @@
-org.apache.streampipes.sinks.brokers.jvm.mqtt.title = MQTT Publisher
-org.apache.streampipes.sinks.brokers.jvm.mqtt.description = Publishes events
to a MQTT topic
+org.apache.streampipes.sinks.brokers.jvm.mqtt.title=MQTT Publisher
+org.apache.streampipes.sinks.brokers.jvm.mqtt.description=Publishes events to
a MQTT topic
-topic.title= MQTT Topic
-topic.description=Select a MQTT topic
+host.title=Host
+host.description=IP address or DNS of MQTT broker
-broker-settings.title=MQTT broker settings (use prefix tcp://)
-broker-settings.description=Provide settings of the MQTT broker to connect
with.
+port.title=Port
+port.description=Port of MQTT broker (default 1883)
-host.title=MQTT Broker
-host.description=IP address or hostname of MQTT broker. (use prefix tcp://)
+topic.title=Topic
+topic.description=Enter MQTT topic
-port.title=Port
-port.description=Port of the MQTT broker. Default port 1883
+username.title=Username
+username.description=The username to authenticate with the broker
+
+password.title=Password
+password.description=The password to authenticate with the broker
+
+auth-mode.title=Authentication
+auth-mode.description=Unauthenticated or basic auth
+
+no-auth-alternative.title=Unauthenticated
+no-auth-alternative.description=No authentication
+
+basic-auth-alternative.title=Basic auth
+basic-auth-alternative.description=Username and password
+
+username-group.title=Username and password
+
+qos-level.title=QoS
+qos-level.description=Quality of Service level (default: at-least-once)
+
+clean-session.title=Clean session?
+clean-session.description=Deselect to persist topic subscriptions & ack
positions across client sessions (default: Yes)
+
+will-retain.title=Retain will message?
+will-retain.description=Select if you want the Will to be published with the
retain option (default: No)
+
+encryption-mode.title=Encryption
+encryption-mode.description=Select protocol. TCP (plaintext), SSL/TLS
(encrypted, not yet supported!)
+
+reconnect-period.title=Reconnect period (seconds)
+reconnect-period.description=Amount of time to wait between reconnect attempts
(default: 30s)
+
+lwt-mode.title=Last will and testament
+lwt-mode.description=Last will notifies other clients in case of ungraceful
disconnects
+
+no-lwt-alternative.title=No last will
+no-lwt-alternative.description=No last will and testament
+
+lwt-alternative.title=Last will
+lwt-alternative.description=Use last will and testament
+
+lwt-group.title=Last will and testament
+
+lwt-topic.title=Last will topic
+lwt-topic.description=Enter last will topic
+
+lwt-message.title=Last will message
+lwt-message.description=Enter last will message
+
+lwt-qos-level.title=Last will QoS
+lwt-qos-level.description=Quality of Service level for last will (default:
at-most-once)
+
+retain.title=Retain messages?
+retain.description=Retain message at broker when no subscribers are present to
avoid event discard (default: No)
+
+keep-alive.title=Keep alive (seconds)
+keep-alive.description=Amount of time that the broker and client may not
communicate (default: 30s)
+mqtt-version-compliant.title=MQTT Broker is MQTT v3.1.1 compliant?
+mqtt-version-compliant.description=Set to "3.1.1" to use MQTT version 3.1.1,
otherwise it uses the 3.1 protocol (default: "3.1.1")
\ No newline at end of file
