Merge branch 'master' into enhanced_device

src/ebusd/bushandler.cpp

@@ -1238,7 +1238,7 @@ result_t BusHandler::prepareScan(symbol_t slave, bool full, const string& levels
     return RESULT_OK;
   }
   deque<Message*> messages;
-  m_messages->findAll("scan", "", levels, true, true, false, false, true, true, 0, 0, &messages);
+  m_messages->findAll("scan", "", levels, true, true, false, false, true, true, 0, 0, false, &messages);
   auto it = messages.begin();
   while (it != messages.end()) {
     Message* message = *it;

src/ebusd/datahandler.h

@@ -149,7 +149,7 @@ class DataSink : virtual public DataHandler {
   bool isDataSink() const override { return true; }
 
   /**
-   * Notify the sink of an updated @a Message.
+   * Notify the sink of an updated @a Message (not necessarily changed though).
    * @param message the updated @a Message.
    */
   virtual void notifyUpdate(Message* message);

src/ebusd/mainloop.cpp

@@ -359,7 +359,7 @@ void MainLoop::run() {
     time(&now);
     if (!dataSinks.empty()) {
       messages.clear();
-      m_messages->findAll("", "", "*", false, true, true, true, true, true, sinkSince, now, &messages);
+      m_messages->findAll("", "", "*", false, true, true, true, true, true, sinkSince, now, false, &messages);
       for (const auto message : messages) {
         for (const auto dataSink : dataSinks) {
           dataSink->notifyUpdate(message);
@@ -403,7 +403,7 @@ void MainLoop::run() {
     if (listening) {
       string levels = getUserLevels(user);
       messages.clear();
-      m_messages->findAll("", "", levels, false, true, true, true, true, true, since, now, &messages);
+      m_messages->findAll("", "", levels, false, true, true, true, true, true, since, now, true, &messages);
       for (const auto message : messages) {
         ostream << message->getCircuit() << " " << message->getName() << " = " << dec;
         message->decodeLastData(false, nullptr, -1, 0, &ostream);
@@ -461,26 +461,26 @@ result_t MainLoop::decodeMessage(const string &data, bool isHttp, bool* connecte
   string token, previous;
   istringstream stream(data);
   vector<string> args;
-  bool escaped = false;
+  char escaped = 0;
 
   char delim = ' ';
   while (getline(stream, token, delim)) {
     if (!isHttp) {
       if (escaped) {
         args.pop_back();
-        if (token.length() > 0 && token[token.length()-1] == '"') {
+        if (token.length() > 0 && token[token.length()-1] == escaped) {
           token.erase(token.length() - 1, 1);
-          escaped = false;
+          escaped = 0;
         }
         token = previous + " " + token;
       } else if (token.length() == 0) {  // allow multiple space chars for a single delimiter
         continue;
-      } else if (token[0] == '"') {
+      } else if (token[0] == '"' || token[0] == '\'') {
         token.erase(0, 1);
-        if (token.length() > 0 && token[token.length()-1] == '"') {
+        if (token.length() > 0 && token[token.length()-1] == token[0]) {
           token.erase(token.length() - 1, 1);
         } else {
-          escaped = true;
+          escaped = token[0];
         }
       }
     }
@@ -867,7 +867,7 @@ result_t MainLoop::executeRead(const vector<string>& args, const string& levels,
       return RESULT_OK;
     }
     deque<Message*> messages;
-    m_newlyDefinedMessages->findAll("", "", levels, false, true, false, false, true, false, 0, 0, &messages);
+    m_newlyDefinedMessages->findAll("", "", levels, false, true, false, false, true, false, 0, 0, false, &messages);
     if (messages.empty()) {
       *ostream << "ERR: bad definition: no read message";
       return RESULT_OK;
@@ -1082,7 +1082,7 @@ result_t MainLoop::executeWrite(const vector<string>& args, const string levels,
       return RESULT_OK;
     }
     deque<Message*> messages;
-    m_newlyDefinedMessages->findAll("", "", levels, false, false, true, false, true, false, 0, 0, &messages);
+    m_newlyDefinedMessages->findAll("", "", levels, false, false, true, false, true, false, 0, 0, false, &messages);
     if (messages.empty()) {
       *ostream << "ERR: bad definition: no write message";
       return RESULT_OK;
@@ -1325,7 +1325,7 @@ result_t MainLoop::executeFind(const vector<string>& args, const string& levels,
   }
   deque<Message*> messages;
   m_messages->findAll(circuit, args.size() == argPos ? "" : args[argPos], useLevels,
-      exact, withRead, withWrite, withPassive, userLevel, !withConditions, 0, 0, &messages);
+      exact, withRead, withWrite, withPassive, userLevel, !withConditions, 0, 0, false, &messages);
 
   bool found = false;
   char str[32];
@@ -1910,7 +1910,7 @@ result_t MainLoop::executeGet(const vector<string>& args, bool* connected, ostri
       verbosity |= (valueName ? OF_VALUENAME : numeric ? OF_NUMERIC : 0) | OF_JSON | (full ? OF_ALL_ATTRS : 0)
                    | (withDefinition ? OF_DEFINTION : 0);
       deque<Message*> messages;
-      m_messages->findAll(circuit, name, getUserLevels(user), exact, true, withWrite, true, true, true, 0, 0,
+      m_messages->findAll(circuit, name, getUserLevels(user), exact, true, withWrite, true, true, true, 0, 0, false,
                           &messages);
       string lastName;
       for (deque<Message*>::iterator it = messages.begin(); it != messages.end(); it++) {

src/ebusd/mqtthandler.cpp

@@ -30,13 +30,15 @@ using std::dec;
 
 #define O_HOST 1
 #define O_PORT (O_HOST+1)
-#define O_USER (O_PORT+1)
+#define O_CLID (O_PORT+1)
+#define O_USER (O_CLID+1)
 #define O_PASS (O_USER+1)
 #define O_TOPI (O_PASS+1)
 #define O_RETA (O_TOPI+1)
 #define O_JSON (O_RETA+1)
 #define O_IGIN (O_JSON+1)
-#define O_CAFI (O_IGIN+1)
+#define O_CHGS (O_IGIN+1)
+#define O_CAFI (O_CHGS+1)
 #define O_CERT (O_CAFI+1)
 #define O_KEYF (O_CERT+1)
 #define O_KEPA (O_KEYF+1)
@@ -46,6 +48,8 @@ static const struct argp_option g_mqtt_argp_options[] = {
   {nullptr,       0,      nullptr,       0, "MQTT options:", 1 },
   {"mqtthost",    O_HOST, "HOST",        0, "Connect to MQTT broker on HOST [localhost]", 0 },
   {"mqttport",    O_PORT, "PORT",        0, "Connect to MQTT broker on PORT (usually 1883), 0 to disable [0]", 0 },
+  {"mqttclientid",O_CLID, "ID",          0, "Set client ID for connection to MQTT broker [" PACKAGE_NAME "_"
+   PACKAGE_VERSION "_<pid>]", 0 },
   {"mqttuser",    O_USER, "USER",        0, "Connect as USER to MQTT broker (no default)", 0 },
   {"mqttpass",    O_PASS, "PASSWORD",    0, "Use PASSWORD when connecting to MQTT broker (no default)", 0 },
   {"mqtttopic",   O_TOPI, "TOPIC",       0,
@@ -54,6 +58,7 @@ static const struct argp_option g_mqtt_argp_options[] = {
   {"mqttjson",    O_JSON, nullptr,       0, "Publish in JSON format instead of strings", 0 },
   {"mqttignoreinvalid", O_IGIN, nullptr, 0,
    "Ignore invalid parameters during init (e.g. for DNS not resolvable yet)", 0 },
+  {"mqttchanges", O_CHGS, nullptr,       0, "Whether to only publish changed messages instead of all received", 0 },
 
 #if (LIBMOSQUITTO_MAJOR >= 1)
   {"mqttca",      O_CAFI, "CA",          0, "Use CA file or dir (ending with '/') for MQTT TLS (no default)", 0 },
@@ -67,6 +72,7 @@ static const struct argp_option g_mqtt_argp_options[] = {
 
 static const char* g_host = "localhost";  //!< host name of MQTT broker [localhost]
 static uint16_t g_port = 0;               //!< optional port of MQTT broker, 0 to disable [0]
+static const char* g_clientId = nullptr;  //!< optional clientid override for MQTT broker
 static const char* g_username = nullptr;  //!< optional user name for MQTT broker (no default)
 static const char* g_password = nullptr;  //!< optional password for MQTT broker (no default)
 /** the MQTT topic string parts. */
@@ -76,6 +82,7 @@ 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
 static bool g_ignoreInvalidParams = false;  //!< ignore invalid parameters during init
+static bool g_onlyChanges = true;         //!< whether to only publish changed messages instead of all received
 
 #if (LIBMOSQUITTO_MAJOR >= 1)
 static const char* g_cafile = nullptr;    //!< CA file for TLS
@@ -113,6 +120,14 @@ static error_t mqtt_parse_opt(int key, char *arg, struct argp_state *state) {
     }
     break;
 
+  case O_CLID:  // --mqttclientid=clientid
+    if (arg == nullptr || arg[0] == 0) {
+      argp_error(state, "invalid mqttclientid");
+      return EINVAL;
+    }
+    g_clientId = arg;
+    break;
+
   case O_USER:  // --mqttuser=username
     if (arg == nullptr) {
       argp_error(state, "invalid mqttuser");
@@ -151,6 +166,10 @@ static error_t mqtt_parse_opt(int key, char *arg, struct argp_state *state) {
     g_ignoreInvalidParams = true;
     break;
 
+  case O_CHGS:
+    g_onlyChanges = true;
+    break;
+
 #if (LIBMOSQUITTO_MAJOR >= 1)
     case O_CAFI:  // --mqttca=file or --mqttca=dir/
       if (arg == nullptr || arg[0] == 0) {
@@ -346,7 +365,11 @@ MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap*
   } else {
     signal(SIGPIPE, SIG_IGN);  // needed before libmosquitto v. 1.1.3
     ostringstream clientId;
+    if (g_clientId) {
+      clientId << g_clientId;
+    } else {
       clientId << PACKAGE_NAME << '_' << PACKAGE_VERSION << '_' << static_cast<unsigned>(getpid());
+    }
 #if (LIBMOSQUITTO_MAJOR >= 1)
     m_mosquitto = mosquitto_new(clientId.str().c_str(), true, this);
 #else
@@ -430,12 +453,6 @@ void MqttHandler::start() {
   }
 }
 
-void MqttHandler::notifyConnected() {
-  if (m_mosquitto && isRunning()) {
-    mosquitto_subscribe(m_mosquitto, nullptr, m_subscribeTopic.c_str(), 0);
-  }
-}
-
 void on_message(
 #if (LIBMOSQUITTO_MAJOR >= 1)
   struct mosquitto *mosq,
@@ -450,6 +467,16 @@ void on_message(
   handler->notifyTopic(topic, data);
 }
 
+void MqttHandler::notifyConnected() {
+  if (m_mosquitto && isRunning()) {
+    const string sep = (g_publishFormat & OF_JSON) ? "\"" : "";
+    publishTopic(m_globalTopic+"version", sep + (PACKAGE_STRING "." REVISION) + sep, true);
+    publishTopic(m_globalTopic+"running", "true", true);
+    mosquitto_message_callback_set(m_mosquitto, on_message);
+    mosquitto_subscribe(m_mosquitto, nullptr, m_subscribeTopic.c_str(), 0);
+  }
+}
+
 void MqttHandler::notifyTopic(const string& topic, const string& data) {
   size_t pos = topic.rfind('/');
   if (pos == string::npos) {
@@ -508,7 +535,7 @@ void MqttHandler::notifyTopic(const string& topic, const string& data) {
       }
     }
   }
-  if (circuit.empty() || name.empty()) {
+  if (name.empty()) {
     return;
   }
   logOtherInfo("mqtt", "received topic for %s %s", circuit.c_str(), name.c_str());
@@ -542,7 +569,7 @@ void MqttHandler::notifyUpdateCheckResult(const string& checkResult) {
 }
 
 void MqttHandler::run() {
-  time_t lastTaskRun, now, start, lastSignal = 0;
+  time_t lastTaskRun, now, start, lastSignal = 0, lastUpdates = 0;
   bool signal = false;
   string signalTopic = m_globalTopic+"signal";
   string uptimeTopic = m_globalTopic+"uptime";
@@ -550,16 +577,14 @@ void MqttHandler::run() {
 
   time(&now);
   start = lastTaskRun = now;
-  const string sep = (g_publishFormat & OF_JSON) ? "\"" : "";
-  publishTopic(m_globalTopic+"version", sep + (PACKAGE_STRING "." REVISION) + sep, true);
-  publishTopic(m_globalTopic+"running", "true", true);
-  publishTopic(signalTopic, "false");
-  mosquitto_message_callback_set(m_mosquitto, on_message);
   bool allowReconnect = false;
   while (isRunning()) {
+    bool wasConnected = m_connected;
     handleTraffic(allowReconnect);
+    bool reconnected = !wasConnected && m_connected;
     allowReconnect = false;
     time(&now);
+    bool sendSignal = reconnected;
     if (now < start) {
       // clock skew
       if (now < lastSignal) {
@@ -568,18 +593,7 @@ void MqttHandler::run() {
       lastTaskRun = now;
     } else if (now > lastTaskRun+15) {
       allowReconnect = true;
-      if (m_busHandler->hasSignal()) {
+      sendSignal = true;
-        lastSignal = now;
-        if (!signal) {
-          signal = true;
-          publishTopic(signalTopic, "true");
-        }
-      } else {
-        if (signal) {
-          signal = false;
-          publishTopic(signalTopic, "false");
-        }
-      }
       time_t uptime = now-start;
       updates.str("");
       updates.clear();
@@ -587,6 +601,20 @@ void MqttHandler::run() {
       publishTopic(uptimeTopic, updates.str());
       time(&lastTaskRun);
     }
+    if (sendSignal) {
+      if (m_busHandler->hasSignal()) {
+        lastSignal = now;
+        if (!signal || reconnected) {
+          signal = true;
+          publishTopic(signalTopic, "true");
+        }
+      } else {
+        if (signal || reconnected) {
+          signal = false;
+          publishTopic(signalTopic, "false");
+        }
+      }
+    }
     if (!m_updatedMessages.empty()) {
       if (m_connected) {
         m_messages->lock();
@@ -597,7 +625,8 @@ void MqttHandler::run() {
             updates.clear();
             updates << dec;
             for (auto message : *messages) {
-              if (message->getLastChangeTime() > 0 && message->isAvailable()) {
+              if (message->getLastChangeTime() > 0 && message->isAvailable()
+              && (!g_onlyChanges || message->getLastChangeTime() > lastUpdates)) {
                 publishMessage(message, &updates);
               }
             }
@@ -605,6 +634,7 @@ void MqttHandler::run() {
           it = m_updatedMessages.erase(it);
         }
         m_messages->unlock();
+        time(&lastUpdates);
       } else {
         m_updatedMessages.clear();
       }

src/lib/ebus/datatype.cpp

@@ -696,6 +696,14 @@ result_t NumberDataType::readSymbols(size_t offset, size_t length, const SymbolS
         }
       }
 #endif
+      if (isnan(val)) {
+        if (outputFormat & OF_JSON) {
+          *output << "null";
+        } else {
+          *output << nullptr_VALUE;
+        }
+        return RESULT_OK;
+      }
       if (val != 0.0) {
         if (m_divisor < 0) {
           val *= static_cast<float>(-m_divisor);

src/lib/ebus/message.cpp

@@ -2506,7 +2506,7 @@ Message* MessageMap::find(const string& circuit, const string& name, const strin
 
 void MessageMap::findAll(const string& circuit, const string& name, const string& levels,
     bool completeMatch, bool withRead, bool withWrite, bool withPassive, bool includeEmptyLevel, bool onlyAvailable,
-    time_t since, time_t until, deque<Message*>* messages) const {
+    time_t since, time_t until, bool changedSince, deque<Message*>* messages) const {
   string lcircuit = circuit;
   FileReader::tolower(&lcircuit);
   string lname = name;
@@ -2553,7 +2553,7 @@ void MessageMap::findAll(const string& circuit, const string& name, const string
         if (message->getDstAddress() == SYN) {
           continue;
         }
-        time_t lastchg = message->getLastChangeTime();
+        time_t lastchg = changedSince ? message->getLastChangeTime() : message->getLastUpdateTime();
         if ((since != 0 && lastchg < since)
         || (until != 0 && lastchg >= until)) {
           continue;
@@ -2647,7 +2647,7 @@ void MessageMap::invalidateCache(Message* message) {
   string circuit = message->getCircuit();
   string name = message->getName();
   deque<Message*> messages;
-  findAll(circuit, name, "*", true, true, true, true, true, true, 0, 0, &messages);
+  findAll(circuit, name, "*", true, true, true, true, true, true, 0, 0, false, &messages);
   for (auto checkMessage : messages) {
     if (checkMessage != message) {
       checkMessage->m_lastUpdateTime = 0;

src/lib/ebus/message.h

@@ -1403,11 +1403,12 @@ class MessageMap : public MappedFileReader {
    * address), or 0 to ignore.
    * @param until the end time to which to add updates (exclusive, also removes messages with unset destination
    * address), or 0 to ignore.
+   * @changedSince true to use the last change time for the since/until range, false to use the last seen time.
    * @param messages the @a deque to which to add the found @a Message instances.
    */
   void findAll(const string& circuit, const string& name, const string& levels,
     bool completeMatch, bool withRead, bool withWrite, bool withPassive, bool includeEmptyLevel, bool onlyAvailable,
-    time_t since, time_t until, deque<Message*>* messages) const;
+    time_t since, time_t until, bool changedSince, deque<Message*>* messages) const;
 
   /**
    * Find the @a Message instance for the specified master data.

src/lib/ebus/test/test_data.cpp

@@ -412,6 +412,7 @@ int main() {
     {"x,,exp", "-32.767", "10feffff04681103c2", "00", ""},
     {"x,,exp,1000", "-0.000090000",  "10feffff04ec51b8bd", "00", ""},
     {"x,,exp,-100", "-9",  "10feffff04ec51b8bd", "00", ""},
+    {"x,,exp", "-",      "10feffff040000c07f", "00", "W"},
     {"x,,exr", "-0.09",  "10feffff04bdb851ec", "00", ""},
     {"x,,exr", "0.0",    "10feffff0400000000", "00", ""},
     {"x,,exr", "-0.001", "10feffff04ba83126f", "00", ""},

src/lib/ebus/test/test_message.cpp

@@ -367,7 +367,7 @@ int main() {
         continue;
       }
       deque<Message*> msgs;
-      messages->findAll("", "", "*", false, true, true, true, true, false, 0, 0, &msgs);
+      messages->findAll("", "", "*", false, true, true, true, true, false, 0, 0, false, &msgs);
       if (msgs.empty()) {
         message = nullptr;
         cout << "\"" << check[0] << "\": create error: message not found" << endl;