Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,7 @@ Contributions are welcome - feel free to open issues and pull requests!

- Auto-update of EPICS PVS via `I/O Intr` records;
- Support for read/write flat MQTT topics (i.e, topics where the payload is a single value or array);
- Support for reading arbitrarily nested fields from JSON topic payloads;
- Support for reading JSON payloads;
- Support for MQTT QoS levels;
- Checks and reject invalid messages (based mostly on type-checking);
- Auto reconnection of broker;
Expand Down Expand Up @@ -117,7 +117,7 @@ Where:
- `<FORMAT>` is the format of the payload: `FLAT` or `JSON`.
- `<TYPE>` is the general type of the expected value [`INT|FLOAT|DIGITAL|STRING|INTARRAY|FLOATARRAY`].
- `<TOPIC>` is the MQTT topic to which the record will be subscribed/published.
- `<FIELD>` is the dot-separated path to the field to extract from a JSON payload (e.g. `sensor.temperature`). Arbitrary nesting is supported. Required when `FORMAT` is `JSON`.
- `<FIELD>` is JSON pointer to extract value from a JSON payload (e.g. `/sensor/temperature`). If empty, JSON root is used.

> **Note on JSON write support:** Writing to JSON-formatted topics is currently **not supported**. At the moment the driver has no way of knowing the JSON structure expected by the broker ahead of time for write records. For this reason, only `FLAT` format can be used for output records.

Expand Down
52 changes: 13 additions & 39 deletions mqttSup/src/drvMqtt.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -73,23 +73,25 @@ DeviceAddress* MqttDriver::parseDeviceAddress(std::string const& function, std::
}
else if (prefix == JSON_FUNC_PREFIX) {
auto spacePos = arguments.find(' ');
std::string topicName;
std::string jsonField;
if (spacePos == std::string::npos) {
fprintf(stderr, "%s::%s: JSON field not specified: %s\n", driverName, functionName, arguments.c_str());
delete addr;
return nullptr;
}
if (spacePos + 1 >= arguments.size()) {
fprintf(stderr, "%s::%s: JSON field is empty: %s\n", driverName, functionName, arguments.c_str());
delete addr;
return nullptr;
topicName = arguments;
jsonField = "";
} else {
if (spacePos + 1 >= arguments.size()) {
fprintf(stderr, "%s::%s: JSON field is empty: %s\n", driverName, functionName, arguments.c_str());
delete addr;
return nullptr;
}
topicName = arguments.substr(0, spacePos);
jsonField = arguments.substr(spacePos + 1, arguments.size());
}
std::string topicName = arguments.substr(0, spacePos);
if (!isValidTopicName(topicName)) {
fprintf(stderr, "%s::%s: Invalid topic name: %s\n", driverName, functionName, topicName.c_str());
delete addr;
return nullptr;
}
std::string jsonField = arguments.substr(spacePos + 1, arguments.size());
addr->format = MqttTopicAddr::JSON;
addr->topicName = topicName;
addr->jsonField = jsonField;
Expand Down Expand Up @@ -256,13 +258,7 @@ void MqttDriver::onMessageCb(Autoparam::Driver* driver, const std::string& topic
if (addr.format == MqttTopicAddr::JSON) {
try {
json root = json::parse(payload);
const json* fieldAddr = findJsonField(root, addr.jsonField);
if (!fieldAddr || fieldAddr->is_null())
throw std::invalid_argument("JSON field not found: " + addr.jsonField);
if (fieldAddr->is_string())
val = fieldAddr->get<std::string>();
else
val = fieldAddr->dump();
val = to_string(root.at(json::json_pointer(addr.jsonField)));
}
catch (const std::exception& e) {
asynPrint(pself->pasynUserSelf, ASYN_TRACE_ERROR,
Expand Down Expand Up @@ -336,28 +332,6 @@ void MqttDriver::onMessageCb(Autoparam::Driver* driver, const std::string& topic
//#############################################################################################
// Helper methods

// Recursively search for a key anywhere in the JSON structure
const json* MqttDriver::findJsonField(const json& payload, const std::string& targetKey) {
if (payload.is_object()) {
for (auto it = payload.begin(); it != payload.end(); ++it) {
if (it.key() == targetKey) {
return &it.value();
}
else {
const json* found = findJsonField(it.value(), targetKey);
if (found) return found;
}
}
}
else if (payload.is_array()) {
for (const auto& el : payload) {
const json* found = findJsonField(el, targetKey);
if (found) return found;
}
}
return nullptr;
}

/* Checks if a string corresponds to one of the supported topic types */
bool MqttDriver::isSupportedTopicType(const std::string& type) {
return MqttDriver::supportedTopicTypes.find(type) != MqttDriver::supportedTopicTypes.end();
Expand Down
2 changes: 1 addition & 1 deletion mqttSup/src/drvMqtt.h
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

using namespace Autoparam::Convenience;
using json = nlohmann::json;
using std::to_string;

static const char* driverName = "MqttDriver";

Expand Down Expand Up @@ -59,7 +60,6 @@ class MqttDriver : public Autoparam::Driver {
DeviceAddress* parseDeviceAddress(std::string const& function, std::string const& arguments);
DeviceVariable* createDeviceVariable(DeviceVariable* baseVar);
/* helper methods */
static const json* findJsonField(const json& payload, const std::string& targetKey);
static bool isInteger(const std::string& s, bool isSigned = true);
static bool isBoolean(const std::string& s);
static bool isFloat(const std::string& s);
Expand Down
Loading