From aec6dfccbef0ab8d27cfa19e1486851a4317621e Mon Sep 17 00:00:00 2001 From: john Date: Sun, 2 Dec 2018 11:34:23 +0100 Subject: [PATCH 1/7] allow empty circuit if not part of topic format --- src/ebusd/mqtthandler.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) mode change 100644 => 100755 src/ebusd/mqtthandler.cpp diff --git a/src/ebusd/mqtthandler.cpp b/src/ebusd/mqtthandler.cpp old mode 100644 new mode 100755 index f3ae76eb..9c0f9983 --- a/src/ebusd/mqtthandler.cpp +++ b/src/ebusd/mqtthandler.cpp @@ -508,7 +508,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()); From c9eae87bd92aa7457a789bfb86d78fd07abc6762 Mon Sep 17 00:00:00 2001 From: john30 Date: Sun, 2 Dec 2018 11:59:42 +0100 Subject: [PATCH 2/7] send version, running, and current signal after reconnect, re-set message callback on reconnect (fixes #222) --- src/ebusd/mqtthandler.cpp | 51 +++++++++++++++++++++------------------ 1 file changed, 28 insertions(+), 23 deletions(-) diff --git a/src/ebusd/mqtthandler.cpp b/src/ebusd/mqtthandler.cpp index f3ae76eb..0ba3fa96 100644 --- a/src/ebusd/mqtthandler.cpp +++ b/src/ebusd/mqtthandler.cpp @@ -430,12 +430,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 +444,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) { @@ -550,16 +554,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 +570,7 @@ void MqttHandler::run() { lastTaskRun = now; } else if (now > lastTaskRun+15) { allowReconnect = true; - if (m_busHandler->hasSignal()) { - lastSignal = now; - if (!signal) { - signal = true; - publishTopic(signalTopic, "true"); - } - } else { - if (signal) { - signal = false; - publishTopic(signalTopic, "false"); - } - } + sendSignal = true; time_t uptime = now-start; updates.str(""); updates.clear(); @@ -587,6 +578,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(); From c467f2b2451afe41032e5251a7e2c21166e32c3e Mon Sep 17 00:00:00 2001 From: john Date: Sun, 2 Dec 2018 12:22:37 +0100 Subject: [PATCH 3/7] add MessageMap::findAll changedSince argument to distinguish touched/changed for since/until range --- src/ebusd/bushandler.cpp | 2 +- src/ebusd/datahandler.h | 2 +- src/ebusd/mainloop.cpp | 14 +++++++------- src/lib/ebus/message.cpp | 6 +++--- src/lib/ebus/message.h | 3 ++- src/lib/ebus/test/test_message.cpp | 2 +- 6 files changed, 15 insertions(+), 14 deletions(-) diff --git a/src/ebusd/bushandler.cpp b/src/ebusd/bushandler.cpp index 9ca23246..42f06491 100755 --- a/src/ebusd/bushandler.cpp +++ b/src/ebusd/bushandler.cpp @@ -1197,7 +1197,7 @@ result_t BusHandler::prepareScan(symbol_t slave, bool full, const string& levels return RESULT_OK; } deque 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; diff --git a/src/ebusd/datahandler.h b/src/ebusd/datahandler.h index c73fa448..253386c6 100755 --- a/src/ebusd/datahandler.h +++ b/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); diff --git a/src/ebusd/mainloop.cpp b/src/ebusd/mainloop.cpp index 7d459319..ffe00370 100755 --- a/src/ebusd/mainloop.cpp +++ b/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); @@ -867,7 +867,7 @@ result_t MainLoop::executeRead(const vector& args, const string& levels, return RESULT_OK; } deque 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& args, const string levels, return RESULT_OK; } deque 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& args, const string& levels, } deque 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,8 +1910,8 @@ result_t MainLoop::executeGet(const vector& args, bool* connected, ostri verbosity |= (valueName ? OF_VALUENAME : numeric ? OF_NUMERIC : 0) | OF_JSON | (full ? OF_ALL_ATTRS : 0) | (withDefinition ? OF_DEFINTION : 0); deque messages; - m_messages->findAll(circuit, name, getUserLevels(user), exact, true, withWrite, true, true, true, 0, 0, - &messages); + m_messages->findAll(circuit, name, getUserLevels(user), exact, true, withWrite, true, true, true, 0, 0, false, + &messages); string lastName; for (deque::iterator it = messages.begin(); it != messages.end(); it++) { Message* message = *it; diff --git a/src/lib/ebus/message.cpp b/src/lib/ebus/message.cpp index f525b310..5bb97027 100755 --- a/src/lib/ebus/message.cpp +++ b/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* messages) const { + time_t since, time_t until, bool changedSince, deque* 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 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; diff --git a/src/lib/ebus/message.h b/src/lib/ebus/message.h index 2cfe44a2..daf7fd80 100755 --- a/src/lib/ebus/message.h +++ b/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* messages) const; + time_t since, time_t until, bool changedSince, deque* messages) const; /** * Find the @a Message instance for the specified master data. diff --git a/src/lib/ebus/test/test_message.cpp b/src/lib/ebus/test/test_message.cpp index b409ffab..16d54bc0 100755 --- a/src/lib/ebus/test/test_message.cpp +++ b/src/lib/ebus/test/test_message.cpp @@ -367,7 +367,7 @@ int main() { continue; } deque 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; From a84c3cb296730d86e89e8153e5cde5df8b23fe01 Mon Sep 17 00:00:00 2001 From: john30 Date: Sun, 2 Dec 2018 12:28:45 +0100 Subject: [PATCH 4/7] added --mqttchanges --- src/ebusd/mqtthandler.cpp | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/src/ebusd/mqtthandler.cpp b/src/ebusd/mqtthandler.cpp index ae6625b0..1a6b5222 100755 --- a/src/ebusd/mqtthandler.cpp +++ b/src/ebusd/mqtthandler.cpp @@ -36,7 +36,8 @@ using std::dec; #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) @@ -54,6 +55,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 }, @@ -76,6 +78,7 @@ static vector 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 @@ -151,6 +154,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) { @@ -546,7 +553,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"; @@ -602,7 +609,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); } } @@ -610,6 +618,7 @@ void MqttHandler::run() { it = m_updatedMessages.erase(it); } m_messages->unlock(); + time(&lastUpdates); } else { m_updatedMessages.clear(); } From 224c947499e9dbf2c98c890a87d69e942f84ae26 Mon Sep 17 00:00:00 2001 From: john30 Date: Sun, 2 Dec 2018 12:44:53 +0100 Subject: [PATCH 5/7] add --mqttclientid (fixes #226) --- src/ebusd/mqtthandler.cpp | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/src/ebusd/mqtthandler.cpp b/src/ebusd/mqtthandler.cpp index 1a6b5222..84b882d7 100755 --- a/src/ebusd/mqtthandler.cpp +++ b/src/ebusd/mqtthandler.cpp @@ -30,7 +30,8 @@ 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) @@ -47,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 "_]", 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, @@ -69,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. */ @@ -116,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"); @@ -353,7 +365,11 @@ MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap* } else { signal(SIGPIPE, SIG_IGN); // needed before libmosquitto v. 1.1.3 ostringstream clientId; - clientId << PACKAGE_NAME << '_' << PACKAGE_VERSION << '_' << static_cast(getpid()); + if (g_clientId) { + clientId << g_clientId; + } else { + clientId << PACKAGE_NAME << '_' << PACKAGE_VERSION << '_' << static_cast(getpid()); + } #if (LIBMOSQUITTO_MAJOR >= 1) m_mosquitto = mosquitto_new(clientId.str().c_str(), true, this); #else From 8ed1e4f15193657d442f812d4f0f0c8efc0770de Mon Sep 17 00:00:00 2001 From: john30 Date: Sun, 2 Dec 2018 13:13:05 +0100 Subject: [PATCH 6/7] fixed NaN in JSON (fixes #225) --- src/lib/ebus/datatype.cpp | 8 ++++++++ src/lib/ebus/test/test_data.cpp | 1 + 2 files changed, 9 insertions(+) diff --git a/src/lib/ebus/datatype.cpp b/src/lib/ebus/datatype.cpp index 399cc6fd..59b5cab2 100755 --- a/src/lib/ebus/datatype.cpp +++ b/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(-m_divisor); diff --git a/src/lib/ebus/test/test_data.cpp b/src/lib/ebus/test/test_data.cpp index c9373c4c..3ca8fd9a 100755 --- a/src/lib/ebus/test/test_data.cpp +++ b/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", ""}, From 7d10104dbf73ccac4ceaba6f02f56d0723fb3ea6 Mon Sep 17 00:00:00 2001 From: john30 Date: Sun, 2 Dec 2018 13:22:16 +0100 Subject: [PATCH 7/7] allow single quote for escaping as well (fixes #208) --- src/ebusd/mainloop.cpp | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/src/ebusd/mainloop.cpp b/src/ebusd/mainloop.cpp index ffe00370..927e16cc 100755 --- a/src/ebusd/mainloop.cpp +++ b/src/ebusd/mainloop.cpp @@ -461,26 +461,26 @@ result_t MainLoop::decodeMessage(const string &data, bool isHttp, bool* connecte string token, previous; istringstream stream(data); vector 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]; } } }