fix for potential access to no longer existing Message instances in DataHandler, validate MQTT topic during argument parsing, extended DataHandler registration to allow registering multiple instances at once, only create MqttHandler if active

src/ebusd/datahandler.cpp

@@ -32,10 +32,10 @@ static const struct argp_child g_last_argp_child = {NULL, 0, NULL, 0};
 
 /** the list of @a argp_child structures. */
 static struct argp_child g_argp_children[
-#ifdef HAVE_MQTT
             1
-#endif
+#ifdef HAVE_MQTT
             +1
+#endif
 ];
 
 const struct argp_child* datahandler_getargs() {
@@ -54,10 +54,7 @@ bool datahandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap
     list<DataHandler*>* handlers) {
   bool success = true;
 #ifdef HAVE_MQTT
-  DataHandler* handler = mqtthandler_register(userInfo, busHandler, messages);
-  if (handler) {
-    handlers->push_back(handler);
-  } else {
+  if (!mqtthandler_register(userInfo, busHandler, messages, handlers)) {
     success = false;
   }
 #endif
@@ -66,7 +63,7 @@ bool datahandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap
 
 void DataSink::notifyUpdate(Message* message) {
   if (message && message->hasLevel(m_levels)) {
-    m_updatedMessages[message]++;
+    m_updatedMessages[message->getKey()]++;
   }
 }
 

src/ebusd/datahandler.h

@@ -164,8 +164,8 @@ class DataSink : virtual public DataHandler {
   /** the allowed access levels. */
   string m_levels;
 
-  /** a map of updated @p Message instances. */
-  map<Message*, int> m_updatedMessages;
+  /** a map of updated @p Message keys. */
+  map<uint64_t, int> m_updatedMessages;
 };
 
 

src/ebusd/main.cpp

@@ -867,6 +867,7 @@ void executeInstructions(MessageMap* messages, bool verbose) {
 
 result_t loadConfigFiles(MessageMap* messages, bool verbose, bool denyRecursive) {
   logInfo(lf_main, "loading configuration files from %s", opt.configPath);
+  messages->lock();
   messages->clear();
   s_globalTemplates.clear();
   for (auto& it : s_templatesByPath) {
@@ -886,6 +887,7 @@ result_t loadConfigFiles(MessageMap* messages, bool verbose, bool denyRecursive)
     logError(lf_main, "error reading config files: %s, last error: %s", getResultCode(result),
         errorDescription.c_str());
   }
+  messages->unlock();
   return RESULT_OK;
 }
 

src/ebusd/mqtthandler.cpp

@@ -66,7 +66,10 @@ static const char* g_host = "localhost";  //!< host name of MQTT broker [localho
 static uint16_t g_port = 0;               //!< optional port of MQTT broker, 0 to disable [0]
 static const char* g_username = NULL;     //!< optional user name for MQTT broker (no default)
 static const char* g_password = NULL;     //!< optional password for MQTT broker (no default)
-static const char* g_topic = PACKAGE;     //!< MQTT topic to use (prefix if without wildcards) [ebusd]
+/** the MQTT topic string parts. */
+static vector<string> g_topicStrs;
+/** the MQTT topic field parts. */
+static vector<string> g_topicFields;
 static bool g_retain = false;             //!< whether to retail all topics
 static OutputFormat g_publishFormat = 0;  //!< the OutputFormat for publishing messages
 
@@ -78,10 +81,11 @@ static const char* g_keyfile = NULL;   //!< client key file for TLS
 static const char* g_keypass = NULL;   //!< client key file password for TLS
 #endif
 
+bool parseTopic(const string& topic, vector<string>* strs, vector<string>* fields);
 
 /**
  * The MQTT argument parsing function.
- * @param key the key from @a mqtt_argp_options.
+ * @param key the key from @a g_mqtt_argp_options.
  * @param arg the option argument, or NULL.
  * @param state the parsing state.
  */
@@ -126,7 +130,9 @@ static error_t mqtt_parse_opt(int key, char *arg, struct argp_state *state) {
       argp_error(state, "invalid mqtttopic");
       return EINVAL;
     }
-    g_topic = arg;
+    if (!parseTopic(arg, &g_topicStrs, &g_topicFields)) {
+      argp_error(state, "malformed mqtttopic");
+    }
     break;
 
   case O_RETA:  // --mqttretain
@@ -186,12 +192,17 @@ static error_t mqtt_parse_opt(int key, char *arg, struct argp_state *state) {
 static const struct argp g_mqtt_argp = { g_mqtt_argp_options, mqtt_parse_opt, NULL, NULL, NULL, NULL, NULL };
 static const struct argp_child g_mqtt_argp_child = {&g_mqtt_argp, 0, "", 1};
 
+
 const struct argp_child* mqtthandler_getargs() {
   return &g_mqtt_argp_child;
 }
 
-DataHandler* mqtthandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages) {
-  return new MqttHandler(userInfo, busHandler, messages);
+bool mqtthandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages,
+    list<DataHandler*>* handlers) {
+  if (g_port > 0) {
+    handlers->push_back(new MqttHandler(userInfo, busHandler, messages));
+  }
+  return true;
 }
 
 /** the known topic field names. */
@@ -216,6 +227,8 @@ bool parseTopic(const string& topic, vector<string>* strs, vector<string>* field
   size_t lastpos = 0;
   size_t end = topic.length();
   vector<string> columns;
+  strs->clear();
+  fields->clear();
   for (size_t pos=topic.find('%', lastpos); pos != string::npos; ) {
     size_t idx = knownFieldCount;
     size_t len = 0;
@@ -281,37 +294,29 @@ void on_connect(
 MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages)
   : DataSink(userInfo, "mqtt"), DataSource(busHandler), Thread(), m_messages(messages), m_connected(false),
     m_lastUpdateCheckResult(".") {
-  bool enabled = g_port != 0;
   m_publishByField = false;
   m_mosquitto = NULL;
-  if (enabled && !parseTopic(g_topic, &m_topicStrs, &m_topicFields)) {
-    logOtherError("mqtt", "malformed topic %s", g_topic);
-    return;
-  }
-  if (!enabled) {
-    return;
-  }
   int major = -1;
   mosquitto_lib_version(&major, NULL, NULL);
   if (major != LIBMOSQUITTO_MAJOR) {
     logOtherError("mqtt", "invalid mosquitto version %d instead of %d", major, LIBMOSQUITTO_MAJOR);
     return;
   }
-  if (m_topicFields.empty()) {
-    if (m_topicStrs.empty()) {
-      m_topicStrs.push_back("");
+  if (g_topicFields.empty()) {
+    if (g_topicStrs.empty()) {
+      g_topicStrs.push_back("");
     } else {
-      string str = m_topicStrs[0];
+      string str = g_topicStrs[0];
       if (str.empty() || str[str.length()-1] != '/') {
-        m_topicStrs[0] = str+"/";
+        g_topicStrs[0] = str+"/";
       }
     }
-    m_topicFields.push_back("circuit");
-    m_topicStrs.push_back("/");
-    m_topicFields.push_back("name");
+    g_topicFields.push_back("circuit");
+    g_topicStrs.push_back("/");
+    g_topicFields.push_back("name");
   } else {
-    for (size_t i = 0; i < m_topicFields.size(); i++) {
-      if (m_topicFields[i] == "field") {
+    for (size_t i = 0; i < g_topicFields.size(); i++) {
+      if (g_topicFields[i] == "field") {
         m_publishByField = true;
         break;
       }
@@ -431,16 +436,16 @@ void MqttHandler::notifyTopic(const string& topic, const string& data) {
   size_t last = 0;
   string circuit, name;
   size_t idx;
-  for (idx = 0; idx < m_topicStrs.size()+1; idx++) {
+  for (idx = 0; idx < g_topicStrs.size()+1; idx++) {
     string field;
     string chk;
-    if (idx < m_topicStrs.size()) {
-      chk = m_topicStrs[idx];
+    if (idx < g_topicStrs.size()) {
+      chk = g_topicStrs[idx];
       pos = remain.find(chk, last);
       if (pos == string::npos) {
         return;
       }
-    } else if (idx-1 < m_topicFields.size()) {
+    } else if (idx-1 < g_topicFields.size()) {
       pos = remain.size();
     } else if (last < remain.size()) {
       return;
@@ -457,7 +462,7 @@ void MqttHandler::notifyTopic(const string& topic, const string& data) {
       if (field.empty()) {
         return;
       }
-      string fieldName = m_topicFields[idx-1];
+      string fieldName = g_topicFields[idx-1];
       if (fieldName == "circuit") {
         circuit = field;
       } else if (fieldName == "name") {
@@ -547,15 +552,28 @@ void MqttHandler::run() {
       publishTopic(uptimeTopic, updates.str());
       time(&lastTaskRun);
     }
-    if (m_connected && !m_updatedMessages.empty()) {
-      for (const auto it : m_updatedMessages) {
-        updates.str("");
-        updates.clear();
-        updates << dec;
-        publishMessage(it.first, &updates);  // TODO avoid using the message while reload command is executed
+    if (!m_updatedMessages.empty()) {
+      if (m_connected) {
+        m_messages->lock();
+        for(auto it = m_updatedMessages.begin(); it != m_updatedMessages.end(); ) {
+          const vector<Message*>* messages = m_messages->getByKey(it->first);
+          if (messages) {
+            updates.str("");
+            updates.clear();
+            updates << dec;
+            for (auto message : *messages) {
+              if (message->getLastChangeTime() > 0 && message->isAvailable()) {
+                publishMessage(message, &updates);
+              }
+            }
+          }
+          it = m_updatedMessages.erase(it);
+        }
+        m_messages->unlock();
+      } else {
+        m_updatedMessages.clear();
       }
     }
-    m_updatedMessages.clear();
   }
 }
 
@@ -586,16 +604,16 @@ void MqttHandler::handleTraffic() {
 
 string MqttHandler::getTopic(const Message* message, const string& suffix, const string& fieldName) {
   ostringstream ret;
-  for (size_t i = 0; i < m_topicStrs.size(); i++) {
-    ret << m_topicStrs[i];
+  for (size_t i = 0; i < g_topicStrs.size(); i++) {
+    ret << g_topicStrs[i];
     if (!message) {
       break;
     }
-    if (i < m_topicFields.size()) {
-      if (m_topicFields[i] == "field") {
+    if (i < g_topicFields.size()) {
+      if (g_topicFields[i] == "field") {
         ret << fieldName;
       } else {
-        message->dumpField(m_topicFields[i], false, &ret);
+        message->dumpField(g_topicFields[i], false, &ret);
       }
     }
   }

src/ebusd/mqtthandler.h

@@ -48,9 +48,11 @@ const struct argp_child* mqtthandler_getargs();
  * @param userInfo the @a UserInfo instance.
  * @param busHandler the @a BusHandler instance.
  * @param messages the @a MessageMap instance.
- * @return the create @a DataHandler, or NULL on error.
+ * @param handlers the @a list to which new @a DataHandler instances shall be added.
+ * @return true if registration was successful.
  */
-DataHandler* mqtthandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages);
+bool mqtthandler_register(UserInfo* userInfo, BusHandler* busHandler, MessageMap* messages,
+    list<DataHandler*>* handlers);
 
 /**
  * The main class supporting MQTT data handling.
@@ -121,12 +123,6 @@ class MqttHandler : public DataSink, public DataSource, public Thread {
   /** the @a MessageMap instance. */
   MessageMap* m_messages;
 
-  /** the MQTT topic string parts. */
-  vector<string> m_topicStrs;
-
-  /** the MQTT topic field parts. */
-  vector<string> m_topicFields;
-
   /** the global topic prefix. */
   string m_globalTopic;
 

src/lib/ebus/message.cpp

@@ -2559,8 +2559,10 @@ void MessageMap::invalidateCache(Message* message) {
 
 void MessageMap::addPollMessage(bool toFront, Message* message) {
   if (message != NULL && message->getPollPriority() > 0) {
+    lock();
     message->m_lastPollTime = toFront ? 0 : m_pollMessages.size();
     m_pollMessages.push(message);
+    unlock();
   }
 }
 

src/lib/ebus/message.h

@@ -26,6 +26,7 @@
 #include <map>
 #include <queue>
 #include <functional>
+#include <mutex>
 #include "lib/ebus/data.h"
 #include "lib/ebus/result.h"
 #include "lib/ebus/symbol.h"
@@ -1432,6 +1433,16 @@ class MessageMap : public MappedFileReader {
    */
   bool decodeCircuit(const string& circuit, OutputFormat outputFormat, ostringstream* output) const;
 
+  /**
+   * Lock this instance against simultaneous modifying access.
+   */
+  void lock() { m_mutex.lock(); }
+
+  /**
+   * Unlock this instance against simultaneous modifying access.
+   */
+  void unlock() { m_mutex.unlock(); }
+
   /**
    * Removes all @a Message instances.
    */
@@ -1545,6 +1556,9 @@ class MessageMap : public MappedFileReader {
 
   /** additional attributes by circuit name. */
   map<string, AttributedItem*> m_circuitData;
+
+  /** a @a mutex for locking out changes. */
+  mutex m_mutex;
 };
 
 }  // namespace ebusd