extend MQTT integration support with seen filter and variable compression

src/ebusd/mqtthandler.cpp

@@ -478,6 +478,28 @@ bool MqttReplacer::isReducable(const map<string, string>& values) const {
   return true;
 }
 
+void MqttReplacer::compress(const map<string, string>& 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<string, string>& 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++) {

src/ebusd/mqtthandler.h

@@ -112,6 +112,12 @@ class MqttReplacer {
    */
   bool isReducable(const map<string, string>& 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<string, string>& 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. */

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<string, string>& attributes,

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;