From 23d930d65333a936c4d91f9d9bfc646f6b573a70 Mon Sep 17 00:00:00 2001 From: john30 Date: Sat, 3 Jun 2017 12:17:18 +0200 Subject: [PATCH] fix for mqtt topic including field --- src/ebusd/mqtthandler.cpp | 51 +++++++++++++++++++-------------- src/ebusd/mqtthandler.h | 4 +-- src/lib/ebus/data.cpp | 28 ++++++++++++------ src/lib/ebus/data.h | 23 +++++++++++---- src/lib/ebus/message.cpp | 2 +- src/lib/ebus/message.h | 10 +++---- src/lib/ebus/test/test_data.cpp | 15 ++++++++-- 7 files changed, 87 insertions(+), 46 deletions(-) diff --git a/src/ebusd/mqtthandler.cpp b/src/ebusd/mqtthandler.cpp index cefbd4cf..893763ad 100644 --- a/src/ebusd/mqtthandler.cpp +++ b/src/ebusd/mqtthandler.cpp @@ -286,13 +286,13 @@ MqttHandler::MqttHandler(UserInfo* userInfo, BusHandler* busHandler, MessageMap* m_topicFields.push_back("name"); } else { for (size_t i = 0; i < m_topicFields.size(); i++) { - if (m_topicFields[i] == "fields") { + if (m_topicFields[i] == "field") { m_publishByField = true; break; } } } - m_globalTopic = getTopic(NULL, -1, "global/"); + m_globalTopic = getTopic(NULL, "global/"); m_mosquitto = NULL; if (mosquitto_lib_init() != MOSQ_ERR_SUCCESS) { logOtherError("mqtt", "unable to initialize"); @@ -489,7 +489,7 @@ void MqttHandler::run() { publishTopic(m_globalTopic+"running", "true"); publishTopic(signalTopic, "false"); mosquitto_message_callback_set(m_mosquitto, on_message); - string subTopic = getTopic(NULL, -1, "#"); + string subTopic = getTopic(NULL, "#"); mosquitto_subscribe(m_mosquitto, NULL, subTopic.c_str(), 0); while (isRunning()) { handleTraffic(); @@ -557,7 +557,7 @@ void MqttHandler::handleTraffic() { } } -string MqttHandler::getTopic(const Message* message, ssize_t fieldIndex, const string& suffix) { +string MqttHandler::getTopic(const Message* message, const string& suffix, const string& fieldName) { ostringstream ret; for (size_t i = 0; i < m_topicStrs.size(); i++) { ret << m_topicStrs[i]; @@ -565,8 +565,8 @@ string MqttHandler::getTopic(const Message* message, ssize_t fieldIndex, const s break; } if (i < m_topicFields.size()) { - if (m_topicFields[i] == "fields" && fieldIndex >= 0) { - ret << message->getFieldName(fieldIndex); // TODO skip ignored fields + if (m_topicFields[i] == "field") { + ret << fieldName; } else { message->dumpField(m_topicFields[i], false, &ret); } @@ -579,24 +579,33 @@ string MqttHandler::getTopic(const Message* message, ssize_t fieldIndex, const s } void MqttHandler::publishMessage(const Message* message, ostringstream* updates) { - result_t result = message->decodeLastData(false, NULL, -1, 0, updates); - if (result != RESULT_OK) { - logOtherError("mqtt", "decode %s %s: %s", message->getCircuit().c_str(), message->getName().c_str(), - getResultCode(result)); + if (!m_publishByField) { + result_t result = message->decodeLastData(false, NULL, -1, 0, updates); + if (result != RESULT_OK) { + logOtherError("mqtt", "decode %s %s: %s", message->getCircuit().c_str(), message->getName().c_str(), + getResultCode(result)); + return; + } + publishTopic(getTopic(message), updates->str()); return; } - if (m_publishByField) { - ssize_t index = 0; - istringstream input(updates->str()); - string token; - while (getline(input, token, UI_FIELD_SEPARATOR)) { - string topic = getTopic(message, index); - publishTopic(topic, token); - index++; + ssize_t index = 0; + do { + string name = message->getFieldName(index); + if (name.empty()) { + break; } - } else { - publishTopic(getTopic(message), updates->str()); - } + result_t result = message->decodeLastData(false, NULL, index, 0, updates); + if (result != RESULT_OK) { + logOtherError("mqtt", "decode %s %s %s: %s", message->getCircuit().c_str(), message->getName().c_str(), + name.c_str(), getResultCode(result)); + return; + } + publishTopic(getTopic(message, "", name), updates->str()); + updates->str(""); + updates->clear(); + index++; + } while (index < MAX_LEN); } void MqttHandler::publishTopic(const string& topic, const string& data, bool retain) { diff --git a/src/ebusd/mqtthandler.h b/src/ebusd/mqtthandler.h index eccf23d8..ec3c38db 100644 --- a/src/ebusd/mqtthandler.h +++ b/src/ebusd/mqtthandler.h @@ -97,11 +97,11 @@ class MqttHandler : public DataSink, public DataSource, public Thread { /** * Build the MQTT topic string for the @a Message. * @param message the @a Message to build the topic string for. - * @param fieldIndex the optional field index for the field column, or -1. * @param suffix the optional suffix string to append. + * @param fieldName the name of the singular field, or empty. * @return the topic string. */ - string getTopic(const Message* message, ssize_t fieldIndex = -1, const string& suffix = ""); + string getTopic(const Message* message, const string& suffix = "", const string& fieldName = ""); /** * Prepare a @a Message and publish as topic. diff --git a/src/lib/ebus/data.cpp b/src/lib/ebus/data.cpp index aaf06f60..8e29ccac 100644 --- a/src/lib/ebus/data.cpp +++ b/src/lib/ebus/data.cpp @@ -494,7 +494,7 @@ result_t SingleDataField::read(const SymbolString& data, size_t offset, if (offset + (remainder?1:m_length) > data.getDataSize()) { return RESULT_ERR_INVALID_POS; } - if (isIgnored() || (fieldName != NULL && (m_name != fieldName || fieldIndex > 0))) { + if (isIgnored() || (fieldName != NULL && m_name != fieldName) || fieldIndex > 0) { return RESULT_EMPTY; } result_t res = m_dataType->readRawValue(offset, m_length, data, output); @@ -514,7 +514,7 @@ result_t SingleDataField::read(const SymbolString& data, size_t offset, if (offset + (remainder?1:m_length) > data.getDataSize()) { return RESULT_ERR_INVALID_POS; } - if (isIgnored() || (fieldName != NULL && (m_name != fieldName || fieldIndex > 0))) { + if (isIgnored() || (fieldName != NULL && m_name != fieldName) || fieldIndex > 0) { return RESULT_EMPTY; } bool shortFormat = outputFormat & OF_SHORT; @@ -900,14 +900,26 @@ size_t DataFieldSet::getLength(PartType partType, size_t maxLength) const { } string DataFieldSet::getName(ssize_t fieldIndex) const { - if (fieldIndex < 0) { + if (fieldIndex < (ssize_t)m_ignoredCount) { return m_name; } - if ((size_t)fieldIndex >= m_fields.size()) { + if ((size_t)fieldIndex + m_ignoredCount >= m_fields.size()) { return ""; } if (m_uniqueNames) { - return m_fields[fieldIndex]->getName(-1); + if (m_ignoredCount == 0) { + return m_fields[fieldIndex]->getName(-1); + } + ssize_t remain = fieldIndex; + for (const auto field : m_fields) { + if (field->isIgnored()) { + continue; + } + remain--; + if (remain == 0) { + return field->getName(-1); + } + } } ostringstream ostream; ostream << static_cast(fieldIndex); @@ -973,7 +985,7 @@ result_t DataFieldSet::read(const SymbolString& data, size_t offset, if (result != RESULT_EMPTY) { found = true; } - if (findFieldIndex && fieldName == field->getName(-1)) { + if (findFieldIndex && !field->isIgnored() && (fieldName == NULL || fieldName == field->getName(-1))) { if (fieldIndex == 0) { if (!found) { return RESULT_ERR_NOTFOUND; @@ -994,7 +1006,7 @@ result_t DataFieldSet::read(const SymbolString& data, size_t offset, result_t DataFieldSet::read(const SymbolString& data, size_t offset, bool leadingSeparator, const char* fieldName, ssize_t fieldIndex, OutputFormat outputFormat, ssize_t outputIndex, ostream* output) const { - bool previousFullByteOffset = true, found = false, findFieldIndex = fieldName != NULL && fieldIndex >= 0; + bool previousFullByteOffset = true, found = false, findFieldIndex = fieldIndex >= 0; if (outputIndex < 0 && (!m_uniqueNames || ((outputFormat & OF_JSON) && !(outputFormat & OF_NAMES)))) { outputIndex = 0; } @@ -1020,7 +1032,7 @@ result_t DataFieldSet::read(const SymbolString& data, size_t offset, found = true; leadingSeparator = true; } - if (findFieldIndex && fieldName == field->getName(-1)) { + if (findFieldIndex && !field->isIgnored() && (fieldName == NULL || fieldName == field->getName(-1))) { if (fieldIndex == 0) { if (!found) { return RESULT_ERR_NOTFOUND; diff --git a/src/lib/ebus/data.h b/src/lib/ebus/data.h index 66b9c395..2e45e00a 100644 --- a/src/lib/ebus/data.h +++ b/src/lib/ebus/data.h @@ -258,10 +258,10 @@ class DataField : public AttributedItem { /** * Get the specified field name. - * @param fieldIndex the index of the field, or -1 for this. - * @return the field name, or the index as string if not unique or not available. + * @param fieldIndex the index of the field (excluding ignored fields), or -1 for this. + * @return the field name, or the index as string is not unique or not available. */ - virtual string getName(ssize_t fieldIndex) const { return m_name; } + virtual string getName(ssize_t fieldIndex) const = 0; /** * Dump the field settings to the output. @@ -282,7 +282,7 @@ class DataField : public AttributedItem { * @param data the data @a SymbolString for reading binary data. * @param offset the additional offset to add for reading binary data. * @param fieldName the name of the field to read, or NULL for the first field. - * @param fieldIndex the optional index of the named field, or -1. + * @param fieldIndex the optional index of the field (either named or overall), or -1. * @param output the variable in which to store the numeric value. * @return @a RESULT_OK on success, * or @a RESULT_EMPTY if the field was skipped (either if the partType does @@ -298,7 +298,7 @@ class DataField : public AttributedItem { * @param offset the additional offset to add for reading binary data. * @param leadingSeparator whether to prepend a separator before the formatted value. * @param fieldName the optional name of a field to limit the output to. - * @param fieldIndex the optional index of the named field to limit the output to, or -1. + * @param fieldIndex the optional index of the field to limit the output to (either named or overall), or -1. * @param outputFormat the @a OutputFormat options to use. * @param outputIndex the optional index of the field when using an indexed output format, or -1. * @param output the @a ostream to append the formatted value to. @@ -397,6 +397,11 @@ class SingleDataField : public DataField { */ bool hasFullByteOffset(bool after) const; + // @copydoc + virtual string getName(ssize_t fieldIndex) const { + return isIgnored() || fieldIndex > 0 ? "" : m_name; + } + /** * Dump the common prefix field settings to the output (name and part type). * @param output the @a ostream to dump to. @@ -404,7 +409,7 @@ class SingleDataField : public DataField { void dumpPrefix(ostream* output) const; /** - * Dump the common suffix field settings to the output (optiona unit and comment). + * Dump the common suffix field settings to the output (optional unit and comment). * @param output the @a ostream to dump to. */ void dumpSuffix(ostream* output) const; @@ -596,9 +601,11 @@ class DataFieldSet : public DataField { DataFieldSet(const string& name, const vector fields) : DataField(name), m_fields(fields) { bool uniqueNames = true; + size_t ignoredCount = 0; map names; for (auto field : fields) { if (field->isIgnored()) { + ignoredCount++; continue; } string name = field->getName(-1); @@ -609,6 +616,7 @@ class DataFieldSet : public DataField { names[name] = name; } m_uniqueNames = uniqueNames; + m_ignoredCount = ignoredCount; } /** @@ -677,6 +685,9 @@ class DataFieldSet : public DataField { /** whether all fields have a unique name. */ bool m_uniqueNames; + + /** the number of ignored fields. */ + size_t m_ignoredCount; }; diff --git a/src/lib/ebus/message.cpp b/src/lib/ebus/message.cpp index 33942530..562ee0a8 100644 --- a/src/lib/ebus/message.cpp +++ b/src/lib/ebus/message.cpp @@ -741,7 +741,7 @@ result_t Message::decodeLastData(bool master, bool leadingSeparator, const char* ssize_t fieldIndex, OutputFormat outputFormat, ostream* output) const { result_t result; if (master) { - result = m_data->read(m_lastMasterData, m_id.size() - 2, leadingSeparator, fieldName, fieldIndex, + result = m_data->read(m_lastMasterData, getIdLength(), leadingSeparator, fieldName, fieldIndex, outputFormat, -1, output); } else { result = m_data->read(m_lastSlaveData, 0, leadingSeparator, fieldName, fieldIndex, diff --git a/src/lib/ebus/message.h b/src/lib/ebus/message.h index 09649f81..25949522 100644 --- a/src/lib/ebus/message.h +++ b/src/lib/ebus/message.h @@ -269,7 +269,7 @@ class Message : public AttributedItem { /** * Get the specified field name. - * @param fieldIndex the index of the field. + * @param fieldIndex the index of the field (excluding ignored fields). * @return the field name, or the index as string if not unique or not available. */ virtual string getFieldName(ssize_t fieldIndex) const { return m_data->getName(fieldIndex); } @@ -456,10 +456,10 @@ class Message : public AttributedItem { /** * Decode the value from the last stored master or slave data. - * @param master true for deocding the master data, false for slave. + * @param master true for decoding the master data, false for slave. * @param leadingSeparator whether to prepend a separator before the formatted value. * @param fieldName the optional name of a field to limit the output to. - * @param fieldIndex the optional index of the named field to limit the output to, or -1. + * @param fieldIndex the optional index of the field to limit the output to (either named or overall), or -1. * @param outputFormat the @a OutputFormat options to use. * @param output the @a ostream to append the formatted value to. * @return @a RESULT_OK on success, or an error code. @@ -471,7 +471,7 @@ class Message : public AttributedItem { * Decode the value from the last stored master and slave data. * @param leadingSeparator whether to prepend a separator before the formatted value. * @param fieldName the optional name of a field to limit the output to. - * @param fieldIndex the optional index of the named field to limit the output to, or -1. + * @param fieldIndex the optional index of the field to limit the output to (either named or overall), or -1. * @param outputFormat the @a OutputFormat options to use. * @param output the @a ostream to append the formatted value to. * @return @a RESULT_OK on success, or an error code. @@ -482,7 +482,7 @@ class Message : public AttributedItem { /** * Decode a particular numeric field value from the last stored data. * @param fieldName the name of the field to decode, or NULL for the first field. - * @param fieldIndex the optional index of the named field, or -1. + * @param fieldIndex the optional index of the field (either named or overall), or -1. * @param output the variable in which to store the value. * @return @a RESULT_OK on success, or an error code. */ diff --git a/src/lib/ebus/test/test_data.cpp b/src/lib/ebus/test/test_data.cpp index 197d2d06..1089f74a 100644 --- a/src/lib/ebus/test/test_data.cpp +++ b/src/lib/ebus/test/test_data.cpp @@ -502,6 +502,10 @@ int main() { {"x,,temp;HEX:2", "18.004;13 14", "10fe07000401121314", "00", ""}, // reference to template and base type {"x,,temp;HEX:2", "temp=18.004;=13 14", "10fe07000401121314", "00", "v"}, // reference to template and base type {"x,,temp:degrees;HEX:2", "degrees=18.004;=13 14", "10fe07000401121314", "00", "v"}, // reference to template and base type + {"x,,uch;UCH;IGN;UCH", "41", "1008ffff00", "0426272829", "wi2"}, + {"x,,uch,,,,x,,uch,,,,x,,ign,,,,x,,uch,,,,", "41", "1008ffff00", "0426272829", "wi2"}, + {"x,,uch,,,,y,,uch,,,,x,,ign,,,,x,,uch,,,,", "41", "1008ffff00", "0426272829", "wIi1"}, + {"x,,uch,,,,y,,uch,,,,z,,ign,,,,x,,uch,,,,", "41", "1008ffff00", "0426272829", "wIi1"}, }; DataFieldTemplates* templates = new DataFieldTemplates(); unsigned int lineNo = 0; @@ -535,6 +539,11 @@ int main() { bool failedReadMatch = flags.find('R') != string::npos; bool failedWrite = flags.find('w') != string::npos; bool failedWriteMatch = flags.find('W') != string::npos; + const char* findName = flags.find('I') == string::npos ? NULL : "x"; + ssize_t findIndex = -1; + if (flags.find('i') != string::npos) { + findIndex = parseSignedInt(flags.substr(flags.find('i')+1).c_str(), 10, 0, 9, &result); + } OutputFormat verbosity = 0; if (flags.find("v") != string::npos) { verbosity |= OF_NAMES; @@ -572,7 +581,7 @@ int main() { dummystr.str("#"); result = reader.readLineFromStream(__FILE__, false, &dummystr, &lineNo, &row, &errorDescription, NULL, NULL); if (result != RESULT_OK) { - cout << "\"" << check[0] << "\": reader header error: " << getResultCode(result) << ", " << errorDescription + cout << "\"" << check[0] << "\": read header error: " << getResultCode(result) << ", " << errorDescription << endl; error = true; continue; @@ -616,9 +625,9 @@ int main() { cout << " parse \"" << sstr.getStr().substr(0, 2) << "\" error: " << getResultCode(result) << endl; error = true; } - result = fields->read(mstr, 0, false, NULL, -1, verbosity|(numeric?OF_NUMERIC:0), -1, &output); + result = fields->read(mstr, 0, false, findName, findIndex, verbosity|(numeric?OF_NUMERIC:0), -1, &output); if (result >= RESULT_OK) { - result = fields->read(sstr, 0, !output.str().empty(), NULL, -1, verbosity|(numeric?OF_NUMERIC:0), -1, &output); + result = fields->read(sstr, 0, !output.str().empty(), findName, findIndex, verbosity|(numeric?OF_NUMERIC:0), -1, &output); } if (failedRead) { if (result >= RESULT_OK) {