nils/ebusd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix for mqtt topic including field

Browse Source
This commit is contained in:
john30
2017-06-03 12:17:18 +02:00
parent 6f5f94238d
commit 23d930d653
7 changed files with 87 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ebusd/mqtthandler.cpp
+30 -21
View File
@@ -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) {
src/ebusd/mqtthandler.h
+2 -2
View File
@@ -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.
src/lib/ebus/data.cpp
+20 -8
View File
@@ -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<signed>(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;
src/lib/ebus/data.h
+17 -6
View File
@@ -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<const SingleDataField*> fields)
: DataField(name), m_fields(fields) {
bool uniqueNames = true;
size_t ignoredCount = 0;
map<string, string> 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;
};
src/lib/ebus/message.cpp
+1 -1
View File
@@ -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,
src/lib/ebus/message.h
+5 -5
View File
@@ -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.
*/
src/lib/ebus/test/test_data.cpp
+12 -3
View File
@@ -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) {