diff --git a/src/ebusd/datahandler.cpp b/src/ebusd/datahandler.cpp index 8da17654..0cae5981 100644 --- a/src/ebusd/datahandler.cpp +++ b/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* 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()]++; } } diff --git a/src/ebusd/datahandler.h b/src/ebusd/datahandler.h index b43ccd8f..1f7965c0 100644 --- a/src/ebusd/datahandler.h +++ b/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 m_updatedMessages; + /** a map of updated @p Message keys. */ + map m_updatedMessages; }; diff --git a/src/ebusd/main.cpp b/src/ebusd/main.cpp index 078ded3c..ffec85da 100644 --- a/src/ebusd/main.cpp +++ b/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; } diff --git a/src/ebusd/mqtthandler.cpp b/src/ebusd/mqtthandler.cpp index db9daa21..03a834eb 100644 --- a/src/ebusd/mqtthandler.cpp +++ b/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 g_topicStrs; +/** the MQTT topic field parts. */ +static vector 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* strs, vector* 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* 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* strs, vector* field size_t lastpos = 0; size_t end = topic.length(); vector 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* 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); } } } diff --git a/src/ebusd/mqtthandler.h b/src/ebusd/mqtthandler.h index cc2c1f34..e9afca2d 100644 --- a/src/ebusd/mqtthandler.h +++ b/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* 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 m_topicStrs; - - /** the MQTT topic field parts. */ - vector m_topicFields; - /** the global topic prefix. */ string m_globalTopic; diff --git a/src/lib/ebus/message.cpp b/src/lib/ebus/message.cpp index 62d8c69f..cdcfa02b 100644 --- a/src/lib/ebus/message.cpp +++ b/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(); } } diff --git a/src/lib/ebus/message.h b/src/lib/ebus/message.h index 7969dc78..7fca7540 100644 --- a/src/lib/ebus/message.h +++ b/src/lib/ebus/message.h @@ -26,6 +26,7 @@ #include #include #include +#include #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 m_circuitData; + + /** a @a mutex for locking out changes. */ + mutex m_mutex; }; } // namespace ebusd