diff --git a/src/ebusd/mqtthandler.cpp b/src/ebusd/mqtthandler.cpp index a7b4f5db..e4ad13bc 100755 --- a/src/ebusd/mqtthandler.cpp +++ b/src/ebusd/mqtthandler.cpp @@ -411,6 +411,17 @@ void MqttReplacer::normalize(string& str) { }); } +const string MqttReplacer::str() const { + ostringstream ret; + for (const auto &it: m_parts) { + if (it.second >= 0) { + ret << '%'; + } + ret << it.first; + } + return ret.str(); +} + void MqttReplacer::ensureDefault() { if (m_parts.empty()) { m_parts.emplace_back(string(PACKAGE) + "/", -1); @@ -451,8 +462,16 @@ string MqttReplacer::get(const map& values, bool untilFirstEmpty if (untilFirstEmpty) { break; } - } else if (untilFirstEmpty && pos->second.empty()) { - break; + if (m_emptyIfMissing) { + return ""; + } + } else if (pos->second.empty()) { + if (untilFirstEmpty) { + break; + } + if (m_emptyIfMissing) { + return ""; + } } else { ret << pos->second; } @@ -603,6 +622,14 @@ MqttReplacer& MqttReplacers::get(const string& key) { return ret; } +MqttReplacer MqttReplacers::get(const string& key) const { + const auto& it = m_replacers.find(key); + if (it != m_replacers.cend()) { + return it->second; + } + return MqttReplacer(); +} + string MqttReplacers::get(const string& key, bool untilFirstEmpty, bool onlyAlphanum, const string& fallbackKey) const { auto itc = m_constants.find(key); if (itc!=m_constants.end()) { @@ -784,11 +811,21 @@ void MqttHandler::parseIntegration(const string& line) { } } +/** + * possible data type names. + */ static const char* typeNames[] = { - "number", "bits", "string", "date", "time", "datetime", + "number", "list", "string", "date", "time", "datetime", }; -const string removeTrailingNonTopicPart(const string& str) { +/** + * possible message direction names by (write*2) + (passive*1). + */ +static const char* directionNames[] = { + "r", "u", "w", "uw", +}; + +string removeTrailingNonTopicPart(const string& str) { size_t pos = str.find_last_not_of("/_"); if (pos==string::npos) { return str; @@ -861,37 +898,44 @@ MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap* m_replacers.set("prefixn", removeTrailingNonTopicPart(line)); } m_replacers.reduce(true); - if (m_replacers.uses("type_switch")) { - for (auto typeName: typeNames) { - string str = m_replacers.get("type_switch-" + string(typeName), false, false, "type_switch"); - if (str.empty()) { - continue; - } - str += '\n'; // add trailing newline to ease the split - size_t from = 0; - do { - size_t pos = str.find('\n', from); - string line = str.substr(from, pos-from); - from = pos+1; - FileReader::trim(&line); - if (!line.empty()) { - pos = line.find('='); - if (pos!=string::npos && pos>0) { - string left = line.substr(0, pos); - FileReader::trim(&left); - if (!left.empty()) { - string right = line.substr(pos+1); + if (!m_replacers["type_switch-names"].empty() || m_replacers.uses("type_switch")) { + ostringstream ostr; + for (int i=-1; i(sizeof(directionNames)/sizeof(char*)); i++) { + for (auto typeName : typeNames) { + ostr.str(""); + if (i>=0) { + ostr << directionNames[i] << '-'; + } + ostr << typeName; + const string suffix = ostr.str(); + string str = m_replacers.get("type_switch-" + suffix, false, false, "type_switch"); + if (str.empty()) { + continue; + } + str += '\n'; // add trailing newline to ease the split + size_t from = 0; + do { + size_t pos = str.find('\n', from); + string line = str.substr(from, pos - from); + from = pos + 1; + FileReader::trim(&line); + if (!line.empty()) { + pos = line.find('='); + if (pos != string::npos && pos > 0) { + string left = line.substr(0, pos); + FileReader::trim(&left); + string right = line.substr(pos + 1); FileReader::trim(&right); FileReader::tolower(&right); - m_typeSwitches[typeName].push_back({left, right}); + m_typeSwitches[suffix].push_back({left, right}); } } - } - } while (from* row) { + std::istringstream istr; + istr.str(str); + unsigned int lineNo = 0; + FileReader::splitFields(&istr, row, &lineNo); + if (row->size() == 1 && (*row)[0].empty()) { + row->clear(); + } +} + void MqttHandler::run() { time_t lastTaskRun, now, start, lastSignal = 0, lastUpdates = 0; bool signal = false; - string signalTopic = m_globalTopic+"signal"; - string uptimeTopic = m_globalTopic+"uptime"; + string signalTopic = m_globalTopic + "signal"; + string uptimeTopic = m_globalTopic + "uptime"; ostringstream updates; - + unsigned int filterPriority = 0; + bool filterSeen = false; + string filterCircuit, filterName, filterLevel, filterField, filterDirection; + vector typeSwitchNames; + if (m_hasDefinitionTopic) { + result_t result = RESULT_OK; + filterPriority = parseInt(m_replacers["filter-priority"].c_str(), 10, 0, 9, &result); + if (result != RESULT_OK) { + filterPriority = 0; + } + filterSeen = parseBool(m_replacers["filter-seen"]); + filterCircuit = m_replacers["filter-circuit"]; + FileReader::tolower(&filterCircuit); + filterName = m_replacers["filter-name"]; + FileReader::tolower(&filterName); + filterLevel = m_replacers["filter-level"]; + FileReader::tolower(&filterLevel); + filterField = m_replacers["filter-field"]; + FileReader::tolower(&filterField); + filterDirection = m_replacers["filter-direction"]; + FileReader::tolower(&filterDirection); + if (!m_typeSwitches.empty()) { + splitFields(m_replacers["type_switch-names"], &typeSwitchNames); + } + } time(&now); start = lastTaskRun = now; bool allowReconnect = false; @@ -1174,23 +1252,6 @@ void MqttHandler::run() { } if (m_connected && m_hasDefinitionTopic) { deque messages; - result_t result = RESULT_OK; - unsigned int filterPriority = parseInt(m_replacers["filter-priority"].c_str(), 10, 0, 9, &result); - if (result != RESULT_OK) { - filterPriority = 0; - } - bool filterSeen = parseBool(m_replacers["filter-seen"]); - string filterCircuit = m_replacers["filter-circuit"]; - FileReader::tolower(&filterCircuit); - string filterName = m_replacers["filter-name"]; - FileReader::tolower(&filterName); - string filterLevel = m_replacers["filter-level"]; - FileReader::tolower(&filterLevel); - string filterField = m_replacers["filter-field"]; - FileReader::tolower(&filterField); - string filterDirection = m_replacers["filter-direction"]; - FileReader::tolower(&filterDirection); - bool usesTypeSwitch = !m_typeSwitches.empty(); m_messages->findAll("", "", "", false, true, true, true, true, true, 0, 0, false, &messages); for (const auto& message : messages) { if (filterSeen) { @@ -1199,7 +1260,7 @@ void MqttHandler::run() { } if (message->getDataHandlerState()==1) { // already seen in the past, check for poll prio update - if (message->getCreateTime() <= m_definitionsSince) { + if (m_definitionsSince>1 && message->getCreateTime() <= m_definitionsSince) { continue; } } @@ -1213,7 +1274,7 @@ void MqttHandler::run() { || !FileReader::matches(message->getLevel(), filterLevel, true, true)) { continue; } - string direction = message->isWrite() ? (message->isPassive() ? "uw" : "w") : (message->isPassive() ? "r" : "u"); + const string direction = directionNames[(message->isWrite() ? 2 : 0) + (message->isPassive() ? 1 : 0)]; if (!FileReader::matches(direction, filterDirection, true, true)) { continue; } @@ -1294,20 +1355,40 @@ void MqttHandler::run() { values.set("max", ostr.str()); }; } - if (usesTypeSwitch) { - values.reduce(); + if (!m_typeSwitches.empty()) { + values.reduce(true); str = values.get("type_switch-by", false, false); string typeSwitch; - for (auto& check : m_typeSwitches[typeStr]) { - if (FileReader::matches(str, check.second, true, true)) { - typeSwitch = check.first; - break; + for (int i=0; i<2; i++) { + ostr.str(""); + if (i==0) { + ostr << direction << '-'; + } + ostr << typeStr; + const string key = ostr.str(); + for (auto const &check: m_typeSwitches[key]) { + if (FileReader::matches(str, check.second, true, true)) { + typeSwitch = check.first; + i = 2; // early exit + break; + } } } values.set("type_switch", typeSwitch); + if (!typeSwitchNames.empty()) { + vector strs; + splitFields(typeSwitch, &strs); + for (size_t pos = 0; posgetLastUpdateTime()>message->getCreateTime()) { + // ensure data is published as well + m_updatedMessages[message->getKey()]++; + } } time(&m_definitionsSince); needsWait = true; @@ -1413,6 +1498,21 @@ void MqttHandler::publishDefinition(MqttReplacers values, const string& prefix, publishTopic(defTopic, payload, retain); } +void MqttHandler::publishDefinition(const MqttReplacers& values) { + string defTopic = values.get("definition-topic", false); + if (defTopic.empty()) { + if (needsLog(lf_other, ll_debug)) { + const string str = values.get("definition-topic").str(); + logOtherDebug("mqtt", "cannot publish incomplete definition topic %s", str.c_str()); + } + return; + } + string payload = values.get("definition-payload", false); + string retainStr = values.get("definition-retain", false); + bool retain = parseBool(retainStr); + publishTopic(defTopic, payload, retain); +} + bool MqttHandler::handleTraffic(bool allowReconnect) { if (!m_mosquitto) { return false; diff --git a/src/ebusd/mqtthandler.h b/src/ebusd/mqtthandler.h index 6c2647bc..bedbb076 100644 --- a/src/ebusd/mqtthandler.h +++ b/src/ebusd/mqtthandler.h @@ -67,6 +67,12 @@ class MqttReplacer { */ static void normalize(string& str); + /** + * Get the template string. + * @return the template string (might already be partially reduced). + */ + const string str() const; + /** * Parse the template string. * @param templateStr the template string. @@ -178,6 +184,13 @@ class MqttReplacers { */ MqttReplacer& get(const string& key); + /** + * Get the variable value of the specified key. + * @param key the key for which to get the value. + * @return the value @a MqttReplacer. + */ + MqttReplacer get(const string& key) const; + /** * Get the variable or constant value of the specified key. * @param key the key for which to get the value. @@ -279,8 +292,14 @@ class MqttHandler : public DataSink, public DataSource, public WaitThread { * @param name optional name to set before building the topic/payload, or empty. * @param fallbackPrefix optional fallback prefix to use when topic/payload/retain with prefix above is not defined. */ - void publishDefinition(MqttReplacers values, const string& prefix = "definition-", const string& topic = "", - const string& circuit = "", const string& name = "", const string& fallbackPrefix = ""); + void publishDefinition(MqttReplacers values, const string& prefix, const string& topic, + const string& circuit, const string& name, const string& fallbackPrefix); + + /** + * Publish a definition topic as specified in the given values. + * @param values the values with the message specification. + */ + void publishDefinition(const MqttReplacers& values); /** * Called regularly to handle MQTT traffic.