From a3f49994ad96f832dd681a58e34584fd2ca201af Mon Sep 17 00:00:00 2001 From: john30 Date: Sun, 7 Apr 2019 21:00:30 +0200 Subject: [PATCH] added /list topic (fixes #277) --- src/ebusd/mqtthandler.cpp | 65 ++++++++++++++++++++++++++++++++------- src/ebusd/mqtthandler.h | 9 +++++- 2 files changed, 62 insertions(+), 12 deletions(-) diff --git a/src/ebusd/mqtthandler.cpp b/src/ebusd/mqtthandler.cpp index 8d995600..391b330a 100644 --- a/src/ebusd/mqtthandler.cpp +++ b/src/ebusd/mqtthandler.cpp @@ -558,12 +558,12 @@ void MqttHandler::notifyTopic(const string& topic, const string& data) { return; } string direction = topic.substr(pos+1); - bool isWrite = false; if (direction.empty()) { return; } - isWrite = direction == "set"; - if (!isWrite && direction != "get") { + bool isWrite = direction == "set"; + bool isList = !isWrite && direction == "list"; + if (!isWrite && !isList && direction != "get") { return; } @@ -571,20 +571,30 @@ void MqttHandler::notifyTopic(const string& topic, const string& data) { string remain = topic.substr(0, pos); size_t last = 0; string circuit, name; - size_t idx; - for (idx = 0; idx < g_topicStrs.size()+1; idx++) { + bool finalField = false; + for (size_t idx = 0; idx < g_topicStrs.size()+1 && !finalField; idx++) { string field; string chk; if (idx < g_topicStrs.size()) { chk = g_topicStrs[idx]; pos = remain.find(chk, last); if (pos == string::npos) { - return; + if (!isList) { + return; + } + if (idx == 0 && remain+"/" == chk) { // check for only first prefix, e.g. "ebusd/" + break; + } + pos = remain.size(); + finalField = true; } } else if (idx-1 < g_topicFields.size()) { pos = remain.size(); } else if (last < remain.size()) { - return; + if (!isList) { + return; + } + break; } else { break; } @@ -596,7 +606,10 @@ void MqttHandler::notifyTopic(const string& topic, const string& data) { } } else { if (field.empty()) { - return; + if (!isList) { + return; + } + continue; } string fieldName = g_topicFields[idx-1]; if (fieldName == "circuit") { @@ -610,10 +623,25 @@ void MqttHandler::notifyTopic(const string& topic, const string& data) { } } } + if (isList) { + logOtherInfo("mqtt", "received list topic for %s %s", circuit.c_str(), name.c_str()); + deque messages; + m_messages->findAll(circuit, name, m_levels, true, true, true, true, true, true, 0, 0, false, &messages); + bool onlyWithData = !data.empty(); + for (const auto message : messages) { + time_t lastup = message->getLastUpdateTime(); + if (onlyWithData && lastup == 0) { + continue; + } + ostringstream ostream; + publishMessage(message, &ostream, true); + } + return; + } if (name.empty()) { return; } - logOtherInfo("mqtt", "received topic for %s %s", circuit.c_str(), name.c_str()); + logOtherInfo("mqtt", "received %s topic for %s %s", direction.c_str(), circuit.c_str(), name.c_str()); Message* message = m_messages->find(circuit, name, m_levels, isWrite); if (message == nullptr) { message = m_messages->find(circuit, name, m_levels, isWrite, true); @@ -809,10 +837,15 @@ string MqttHandler::getTopic(const Message* message, const string& suffix, const return ret.str(); } -void MqttHandler::publishMessage(const Message* message, ostringstream* updates) { +void MqttHandler::publishMessage(const Message* message, ostringstream* updates, bool includeWithoutData) { OutputFormat outputFormat = g_publishFormat; bool json = outputFormat & OF_JSON; + bool noData = includeWithoutData && message->getLastUpdateTime() == 0; if (!m_publishByField) { + if (noData) { + publishEmptyTopic(getTopic(message)); // alternatively: , json ? "null" : ""); + return; + } if (json) { *updates << "{"; } @@ -833,6 +866,10 @@ void MqttHandler::publishMessage(const Message* message, ostringstream* updates) } for (size_t index = 0; index < message->getFieldCount(); index++) { string name = message->getFieldName(index); + if (noData) { + publishEmptyTopic(getTopic(message, "", name)); // alternatively: , json ? "null" : ""); + continue; + } result_t result = message->decodeLastData(false, nullptr, index, outputFormat, updates); if (result != RESULT_OK) { logOtherError("mqtt", "decode %s %s %s: %s", message->getCircuit().c_str(), message->getName().c_str(), @@ -850,8 +887,14 @@ void MqttHandler::publishTopic(const string& topic, const string& data, bool ret const char* dataStr = data.c_str(); const size_t len = strlen(dataStr); logOtherDebug("mqtt", "publish %s %s", topicStr, dataStr); - check(mosquitto_publish(m_mosquitto, nullptr, topic.c_str(), (uint32_t)len, + check(mosquitto_publish(m_mosquitto, nullptr, topicStr, (uint32_t)len, reinterpret_cast(dataStr), 0, g_retain || retain), "publish"); } +void MqttHandler::publishEmptyTopic(const string& topic) { + const char* topicStr = topic.c_str(); + logOtherDebug("mqtt", "publish empty %s", topicStr); + check(mosquitto_publish(m_mosquitto, nullptr, topicStr, 0, nullptr, 0, g_retain), "publish empty"); +} + } // namespace ebusd diff --git a/src/ebusd/mqtthandler.h b/src/ebusd/mqtthandler.h index c98dfda5..12f70cf5 100644 --- a/src/ebusd/mqtthandler.h +++ b/src/ebusd/mqtthandler.h @@ -117,8 +117,9 @@ class MqttHandler : public DataSink, public DataSource, public WaitThread { * Prepare a @a Message and publish as topic. * @param message the @a Message to publish. * @param updates the @a ostringstream for preparation. + * @param includeWithoutData whether to publish messages without data as well. */ - void publishMessage(const Message* message, ostringstream* updates); + void publishMessage(const Message* message, ostringstream* updates, bool includeWithoutData = false); /** * Publish a topic update to MQTT. @@ -128,6 +129,12 @@ class MqttHandler : public DataSink, public DataSource, public WaitThread { */ void publishTopic(const string& topic, const string& data, bool retain = false); + /** + * Publish a topic update to MQTT without any data. + * @param topic the topic string. + */ + void publishEmptyTopic(const string& topic); + /** the @a MessageMap instance. */ MessageMap* m_messages;