Add cleanMqttRetain

Author: orianar

src/lib/common/string.cpp

@@ -536,6 +536,7 @@ bool  parseKafkaBrokerList(const std::string& url, std::string* cleanListOut, st
       /* Minimum path is always "/" */
       *path = "/";
     }
+
     size_t slashPos = rawBroker.find('/');
     if (slashPos != std::string::npos)
     {

src/lib/mongoBackend/MongoCommonUpdate.cpp

@@ -2121,11 +2121,17 @@ static unsigned int processSubscriptions
     notification.httpInfo.fill(tSubP->httpInfo);
     notification.mqttInfo.fill(tSubP->mqttInfo);
     notification.kafkaInfo.fill(tSubP->kafkaInfo);
-    notification.type = (
-      notification.mqttInfo.topic.empty() ?
-      (notification.kafkaInfo.topic.empty() ?  ngsiv2::HttpNotification : ngsiv2::KafkaNotification)
-      : ngsiv2::MqttNotification
-      );
+
+    if (!notification.mqttInfo.topic.empty())
+    {
+      notification.type = ngsiv2::MqttNotification;
+    } else if (!notification.kafkaInfo.topic.empty())
+    {
+      notification.type = ngsiv2::KafkaNotification;
+    } else
+    {
+      notification.type = ngsiv2::HttpNotification;
+    }
 
     notificationSent = processNotification(notifyCerP,
                                            tSubP->attrL,

src/lib/mongoBackend/dbConstants.h

@@ -130,7 +130,7 @@
 #define CSUB_MQTTQOS                 "qos"
 #define CSUB_MQTTRETAIN              "retain"
 
-#define CSUB_KAFKATOPIC              "kafka_topic"
+#define CSUB_KAFKATOPIC              "kafkaTopic"
 
 #define CSUB_USER                    "user"
 #define CSUB_PASSWD                  "passwd"

src/lib/mongoBackend/mongoSubCache.cpp

@@ -140,8 +140,9 @@ int mongoSubCacheItemInsert(const char* tenant, const orion::BSONObj& sub)
   if (sub.hasField(CSUB_MQTTTOPIC))
   {
     cSubP->mqttInfo.fill(sub);
-  } else
-  if (sub.hasField(CSUB_KAFKATOPIC)){
+  }
+  else if (sub.hasField(CSUB_KAFKATOPIC))
+  {
     cSubP->kafkaInfo.fill(sub);
   }
   else

test/functionalTest/cases/4388_mqtt_retain/mqttCustom_retain_true.test

@@ -178,7 +178,7 @@ MQTT message at topic orion/retain/custom:
 
 
 --TEARDOWN--
-mosquitto_pub -h mosquitto -p 1883 -t 'orion/retain/custom' -n -r
+cleanMqttRetain mosquitto 'orion/retain/custom' 1883
 brokerStop CB
 dbDrop CB
 accumulatorStop

test/functionalTest/cases/4388_mqtt_retain/mqtt_retain_true.test

@@ -178,7 +178,7 @@ MQTT message at topic orion/retain:
 
 
 --TEARDOWN--
-mosquitto_pub -h mosquitto -p 1883 -t 'orion/retain' -n -r
+cleanMqttRetain mosquitto 'orion/retain' 1883
 brokerStop CB
 dbDrop CB
 accumulatorStop

test/functionalTest/harnessFunctions.sh

@@ -1221,6 +1221,27 @@ kafkaDestroyTopics() {
   done
 }
 
+
+
+# ------------------------------------------------------------------------------
+#
+# cleanMqttRetain -
+#
+# Delete retained topics by posting an empty message in the topic.
+cleanMqttRetain() {
+  local host="$1"
+  local topic="$2"
+  local port="${3:-1883}"
+
+  if [[ -z "$host" || -z "$topic" ]]; then
+    echo "Uso: mqtt_retain_empty <host> <topic> [puerto]" >&2
+    return 1
+  fi
+
+  mosquitto_pub -h "$host" -p "$port" -t "$topic" -n -r
+}
+
+
 # ------------------------------------------------------------------------------
 #
 # orionCurl
@@ -1492,3 +1513,4 @@ export -f brokerStartAwait
 export -f brokerStopAwait
 export -f kafkaCreateTopics
 export -f kafkaDestroyTopics
+export -f cleanMqttRetain