Fix parseKafkaBrokerList

Author: orianar

doc/manuals/admin/build_source.md

@@ -151,4 +151,4 @@ You can generate coverage reports for the Orion Context Broker using the followi
 
         make coverage INSTALL_DIR=~
 
-*NOTE*: Functional tests relying in debug traces are expected to fail under coverage execution (e.g. notification_different_sizes or not_posix_regex_idpattern.test). This is due to the LM_T macros used by the debug traces are disabled in the coverage code build, as they add "noise" in condition coverage. This way coverage reports are more useful.
+*NOTE*: Functional tests relying in debug traces are expected to fail under coverage execution (e.g. notification_different_sizes or not_posix_regex_idpattern.test). This is due to the LM_T macros used by the debug traces are disabled in the coverage code build, as they add "noise" in condition coverage. This way coverage reports are more useful.

src/lib/common/string.cpp

@@ -489,11 +489,16 @@ bool parseUrl(const std::string& url, std::string& host, int& port, std::string&
 *
 * parseKafkaBrokerList
 */
-bool parseKafkaBrokerList(const std::string& url, std::string& cleanListOut, std::string& protocol, std::string& path)
+bool  parseKafkaBrokerList(const std::string& url, std::string* cleanListOut, std::string* protocol, std::string* path)
 {
   if (url.empty())
     return false;
 
+  // Validate output pointers
+  if (!cleanListOut || !protocol || !path) {
+    return false;
+  }
+
   std::stringstream        ss(url);
   std::string              rawBroker;
   std::vector<std::string> cleaned;
@@ -507,8 +512,7 @@ bool parseKafkaBrokerList(const std::string& url, std::string& cleanListOut, std
   //
   // // Accept only kafka://
   //
-  protocol = urlTokens[0];
-
+  *protocol = urlTokens[0];
 
   while (std::getline(ss, rawBroker, ','))
   {
@@ -519,27 +523,26 @@ bool parseKafkaBrokerList(const std::string& url, std::string& cleanListOut, std
       continue;
 
     // Prefix kafka://
-
-    const std::string prefix = protocol + "//";
+    const std::string prefix = *protocol + "//";
 
     if (rawBroker.rfind(prefix, 0) == 0)
     {
       rawBroker.erase(0, prefix.size());
     }
-    // reject if it contains ‘/’ (there should be no path)
-    if (path.empty())
+
+    // reject if it contains '/' (there should be no path)
+    if (path->empty())
     {
       /* Minimum path is always "/" */
-      path = "/";
+      *path = "/";
     }
     size_t slashPos = rawBroker.find('/');
     if (slashPos != std::string::npos)
     {
-      path = rawBroker.substr(slashPos);
+      *path = rawBroker.substr(slashPos);
     }
 
-    //Validations
-
+    // Validations
     std::string host, portStr;
     int         port = 0;
 
@@ -552,8 +555,8 @@ bool parseKafkaBrokerList(const std::string& url, std::string& cleanListOut, std
     }
     else
     {
-      // Pv4 addr:port
-      if (!getIPv4HostPort(rawBroker, host, port, protocol))
+      // IPv4 addr:port
+      if (!getIPv4HostPort(rawBroker, host, port, *protocol))
         return false;
     }
 
@@ -582,18 +585,17 @@ bool parseKafkaBrokerList(const std::string& url, std::string& cleanListOut, std
     return false;
 
   // Rebuild final list
-  cleanListOut.clear();
+  cleanListOut->clear();
   for (size_t i = 0; i < cleaned.size(); ++i)
   {
-    if (i) cleanListOut += ',';
-    cleanListOut += cleaned[i];
+    if (i) *cleanListOut += ',';
+    *cleanListOut += cleaned[i];
   }
 
   return true;
 }
 
 
-
 /* ****************************************************************************
 *
 * parsedUptime

src/lib/common/string.h

@@ -82,9 +82,9 @@ extern bool parseUrl
 extern bool parseKafkaBrokerList
 (
   const std::string&  url,
-  std::string&        cleanListOut,
-  std::string&        protocol,
-  std::string&        path
+  std::string*        cleanListOut,
+  std::string*        protocol,
+  std::string*        path
 );
 
 /* ****************************************************************************

src/lib/jsonParseV2/parseSubscription.cpp

@@ -655,7 +655,7 @@ static std::string parseKafkaUrl(ConnectionInfo* ciP, SubscriptionUpdate* subsP,
   std::string protocol;
   std::string  path;
 
-  if (!parseKafkaBrokerList(urlOpt.value, cleanBrokers, protocol, path))
+  if (!parseKafkaBrokerList(urlOpt.value, &cleanBrokers, &protocol, &path))
     return badInput(ciP, "invalid kafka /url/");
 
   if (protocol != "kafka:")

src/lib/ngsiNotify/Notifier.cpp

@@ -566,7 +566,7 @@ static SenderThreadParams* buildSenderParamsCustom
     //
     // Kafka notification => list of brokers validated with parseKafkaBrokerList()
     //
-    if (!parseKafkaBrokerList(url, cleanBrokers, protocol, uriPath))
+    if (!parseKafkaBrokerList(url, &cleanBrokers, &protocol, &uriPath))
     {
       LM_E(("Runtime Error (not sending notification: malformed Kafka broker "
              "list: '%s')", url.c_str()));
@@ -801,7 +801,7 @@ SenderThreadParams* Notifier::buildSenderParams
     //
     // Kafka notification => list of brokers validated with parseKafkaBrokerList()
     //
-    if (!parseKafkaBrokerList(url, cleanBrokers, protocol, uriPath))
+    if (!parseKafkaBrokerList(url, &cleanBrokers, &protocol, &uriPath))
     {
       LM_E(("Runtime Error (not sending notification: malformed Kafka broker "
              "list: '%s')", url.c_str()));