From a42ece3a8eb7b54eadbbfcf454e7338c374adfe9 Mon Sep 17 00:00:00 2001
From: Sascha Vis <sascha.vis@twtg.io>
Date: Fri, 7 Oct 2022 14:47:47 +0200
Subject: [PATCH] Add retained message flag to the MQTT configuration

---
 .../main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java  | 2 +-
 .../thingsboard/rule/engine/mqtt/TbMqttNodeConfiguration.java   | 2 ++
 2 files changed, 3 insertions(+), 1 deletion(-)

diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java
index 6dd632aa97..77b3eb6ff1 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNode.java
@@ -76,7 +76,7 @@ public class TbMqttNode implements TbNode {
     @Override
     public void onMsg(TbContext ctx, TbMsg msg) {
         String topic = TbNodeUtils.processPattern(this.mqttNodeConfiguration.getTopicPattern(), msg);
-        this.mqttClient.publish(topic, Unpooled.wrappedBuffer(msg.getData().getBytes(UTF8)), MqttQoS.AT_LEAST_ONCE)
+        this.mqttClient.publish(topic, Unpooled.wrappedBuffer(msg.getData().getBytes(UTF8)), MqttQoS.AT_LEAST_ONCE, mqttNodeConfiguration.isRetainedMessage())
                 .addListener(future -> {
                             if (future.isSuccess()) {
                                 ctx.tellSuccess(msg);
diff --git a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNodeConfiguration.java b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNodeConfiguration.java
index 8ae95134ad..e4441a85c1 100644
--- a/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNodeConfiguration.java
+++ b/rule-engine/rule-engine-components/src/main/java/org/thingsboard/rule/engine/mqtt/TbMqttNodeConfiguration.java
@@ -29,6 +29,7 @@ public class TbMqttNodeConfiguration implements NodeConfiguration<TbMqttNodeConf
     private int connectTimeoutSec;
     private String clientId;
     private boolean appendClientIdSuffix;
+    private boolean retainedMessage;
 
     private boolean cleanSession;
     private boolean ssl;
@@ -42,6 +43,7 @@ public class TbMqttNodeConfiguration implements NodeConfiguration<TbMqttNodeConf
         configuration.setConnectTimeoutSec(10);
         configuration.setCleanSession(true);
         configuration.setSsl(false);
+        configuration.setRetainedMessage(false);
         configuration.setCredentials(new AnonymousCredentials());
         return configuration;
     }