From 7128a380f3e085ac2bc7b3628f58f4931b10c2a1 Mon Sep 17 00:00:00 2001 From: John Date: Sun, 23 Jan 2022 18:45:35 +0100 Subject: [PATCH] extend MQTT integration support with seen filter and variable compression --- src/ebusd/mqtthandler.cpp | 47 ++++++++++++++++++++++++++++++++++----- src/ebusd/mqtthandler.h | 9 +++++++- src/lib/ebus/message.cpp | 10 ++++++++- src/lib/ebus/message.h | 18 ++++++++++++++- 4 files changed, 76 insertions(+), 8 deletions(-) diff --git a/src/ebusd/mqtthandler.cpp b/src/ebusd/mqtthandler.cpp index 2762e313..a7b4f5db 100755 --- a/src/ebusd/mqtthandler.cpp +++ b/src/ebusd/mqtthandler.cpp @@ -478,6 +478,28 @@ bool MqttReplacer::isReducable(const map& values) const { return true; } +void MqttReplacer::compress(const map& values) { + bool lastConstant = false; + for (auto it = m_parts.begin(); it != m_parts.end(); ) { + bool isConstant = it->second < 0; + if (!isConstant) { + const auto pos = values.find(it->first); + if (pos!=values.cend()) { + it->second = -1; + it->first = pos->second; + isConstant = true; + } + } + if (!lastConstant || !isConstant) { + lastConstant = isConstant; + ++it; + continue; + } + (it-1)->first += it->first; + it = m_parts.erase(it); + } +} + bool MqttReplacer::reduce(const map& values, string& result, bool onlyAlphanum) const { ostringstream ret; for (const auto &it: m_parts) { @@ -631,7 +653,7 @@ void MqttReplacers::set(const string& key, int value) { m_constants[key] = str.str(); } -void MqttReplacers::reduce() { +void MqttReplacers::reduce(bool compress) { // iterate through variables and reduce as many to constants as possible bool reduced = false; do { @@ -640,6 +662,9 @@ void MqttReplacers::reduce() { string str; if (!it->second.isReducable(m_constants) || !it->second.reduce(m_constants, str)) { + if (compress) { + it->second.compress(m_constants); + } ++it; continue; } @@ -835,7 +860,7 @@ MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap* m_replacers.set("prefix", line); m_replacers.set("prefixn", removeTrailingNonTopicPart(line)); } - m_replacers.reduce(); + 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"); @@ -1154,6 +1179,7 @@ void MqttHandler::run() { 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"]; @@ -1167,7 +1193,18 @@ void MqttHandler::run() { bool usesTypeSwitch = !m_typeSwitches.empty(); m_messages->findAll("", "", "", false, true, true, true, true, true, 0, 0, false, &messages); for (const auto& message : messages) { - if (message->getCreateTime() <= m_definitionsSince) { // only newer defined + if (filterSeen) { + if (message->getLastUpdateTime()==0) { + continue; // no data ever + } + if (message->getDataHandlerState()==1) { + // already seen in the past, check for poll prio update + if (message->getCreateTime() <= m_definitionsSince) { + continue; + } + } + message->setDataHandlerState(1); + } else if (message->getCreateTime() <= m_definitionsSince) { // only newer defined continue; } if ((filterPriority>0 && (message->getPollPriority()==0 || message->getPollPriority()>filterPriority)) @@ -1191,10 +1228,10 @@ void MqttHandler::run() { if (!m_publishByField) { msgValues.set("topic", getTopic(message, "", "")); // TODO already present? } - msgValues.reduce(); + msgValues.reduce(true); string str = msgValues.get("direction_map-"+direction, false, false); msgValues.set("direction_map", str); - msgValues.reduce(); + msgValues.reduce(true); ostringstream fields; size_t fieldCount = message->getFieldCount(); for (size_t index = 0; index < fieldCount; index++) { diff --git a/src/ebusd/mqtthandler.h b/src/ebusd/mqtthandler.h index be858201..6c2647bc 100644 --- a/src/ebusd/mqtthandler.h +++ b/src/ebusd/mqtthandler.h @@ -112,6 +112,12 @@ class MqttReplacer { */ bool isReducable(const map& values) const; + /** + * Compress all subsequent constant values to a single constant value if possible. + * @param values the named values for replacement. + */ + void compress(const map& values); + /** * Reduce the fields to a constant value if possible. * @param values the named values for replacement. @@ -201,8 +207,9 @@ class MqttReplacers { /** * Reduce as many variables to constants as possible. + * @param compress true to compress non-reducable replacers if possible. */ - void reduce(); + void reduce(bool compress = false); private: /** constant values from the integration file. */ diff --git a/src/lib/ebus/message.cpp b/src/lib/ebus/message.cpp index e632c9c4..504e2078 100644 --- a/src/lib/ebus/message.cpp +++ b/src/lib/ebus/message.cpp @@ -115,7 +115,7 @@ Message::Message(const string& filename, const string& circuit, const string& le m_data(data), m_deleteData(deleteData), m_pollPriority(pollPriority), m_usedByCondition(false), m_isScanMessage(false), m_condition(condition), - m_lastUpdateTime(0), m_lastChangeTime(0), m_pollOrder(0), m_lastPollTime(0) { + m_dataHandlerState(0), m_lastUpdateTime(0), m_lastChangeTime(0), m_pollOrder(0), m_lastPollTime(0) { if (circuit == "scan") { setScanMessage(); m_pollPriority = 0; @@ -1025,6 +1025,14 @@ void Message::decodeJson(bool leadingSeparator, bool appendDirectionCondition, b *output << "\n }"; } +bool Message::setDataHandlerState(int state) { + if (state == m_dataHandlerState) { + return false; + } + m_dataHandlerState = state; + return true; +} + ChainedMessage::ChainedMessage(const string& filename, const string& circuit, const string& level, const string& name, bool isWrite, const map& attributes, diff --git a/src/lib/ebus/message.h b/src/lib/ebus/message.h index f4a50fd8..1768f5c8 100644 --- a/src/lib/ebus/message.h +++ b/src/lib/ebus/message.h @@ -525,6 +525,19 @@ class Message : public AttributedItem { */ time_t getCreateTime() const { return m_createTime; } + /** + * Get the arbitrary state value for data handlers. + * @return the state value. + */ + int getDataHandlerState() const { return m_dataHandlerState; } + + /** + * Set the arbitrary state value for data handlers. + * @param state the new state value. + * @return true when the state was changed. + */ + bool setDataHandlerState(int state); + /** * Get the time when this message was last seen with reasonable data. * @return the time when this message was last seen, or 0. @@ -660,9 +673,12 @@ class Message : public AttributedItem { /** the last seen @a SlaveSymbolString. */ SlaveSymbolString m_lastSlaveData; - /** the system time when the message was created. */ + /** the system time when the message was created or changed in poll priority. */ time_t m_createTime; + /** an arbitrary state for data handlers. */ + int m_dataHandlerState; + /** the system time when the message was last updated, 0 for never. */ time_t m_lastUpdateTime;