From 96a57615b4779514704dc81510b5636f3ecff7ab Mon Sep 17 00:00:00 2001 From: Viacheslau Date: Tue, 25 Aug 2026 09:02:56 +0200 Subject: [PATCH 1/4] subscriber fb fix: propagate DomainMode from JSON config to decoder FBs --- .../src/mqtt_subscriber_fb_impl.cpp | 4 + .../tests/test_mqtt_subscriber_fb.cpp | 75 +++++++++++++++++++ 2 files changed, 79 insertions(+) diff --git a/modules/mqtt_streaming_module/src/mqtt_subscriber_fb_impl.cpp b/modules/mqtt_streaming_module/src/mqtt_subscriber_fb_impl.cpp index 203d2c2..5564769 100644 --- a/modules/mqtt_streaming_module/src/mqtt_subscriber_fb_impl.cpp +++ b/modules/mqtt_streaming_module/src/mqtt_subscriber_fb_impl.cpp @@ -361,11 +361,15 @@ void MqttSubscriberFbImpl::setJsonConfig(const std::string config) } if (const auto signalDscs = jsonConfigWrapper.extractDescription(); !signalDscs.empty()) { + using DSM = mqtt::MqttDataWrapper::DomainSignalMode; auto fbConfig = MqttJsonDecoderFbImpl::CreateType().createDefaultConfig(); for (const auto& [signalName, descriptor] : signalDscs) { LOG_I("Creating a decoder FB for the signal \"{}\":", signalName); fbConfig.setPropertyValue(PROPERTY_NAME_DEC_VALUE_NAME, descriptor.valueFieldName); + + const auto tsMode = descriptor.tsFieldName.empty() ? DSM::None : DSM::ExtractFromMessage; + fbConfig.setPropertyValue(PROPERTY_NAME_DEC_TS_MODE, static_cast(tsMode)); fbConfig.setPropertyValue(PROPERTY_NAME_DEC_TS_NAME, descriptor.tsFieldName); if (descriptor.unit.assigned()) fbConfig.setPropertyValue(PROPERTY_NAME_DEC_UNIT, descriptor.unit.getSymbol()); diff --git a/modules/mqtt_streaming_module/tests/test_mqtt_subscriber_fb.cpp b/modules/mqtt_streaming_module/tests/test_mqtt_subscriber_fb.cpp index ac27cf9..910bd5d 100644 --- a/modules/mqtt_streaming_module/tests/test_mqtt_subscriber_fb.cpp +++ b/modules/mqtt_streaming_module/tests/test_mqtt_subscriber_fb.cpp @@ -1,3 +1,4 @@ +#include "mqtt_streaming_module/mqtt_json_decoder_fb_impl.h" #include "mqtt_streaming_module/mqtt_subscriber_fb_impl.h" #include "test_daq_test_helper.h" #include "test_data.h" @@ -47,6 +48,13 @@ class MqttSubscriberFbHelper return signalList; } + auto getNestedFbs() + { + auto fbList = List(); + obj->getFunctionBlocks(&fbList); + return fbList; + } + std::string buildTopicName(const std::string& postfix = "") { return std::string("test/topic/") + std::string(::testing::UnitTest::GetInstance()->current_test_info()->name()) + postfix; @@ -494,6 +502,73 @@ TEST_F(MqttSubscriberFbTest, JsonInitFromFileWithChecking) } +TEST_F(MqttSubscriberFbTest, JsonInitDomainMode) +{ + using DDSM = mqtt::MqttDataWrapper::DomainSignalMode; + + auto config = MqttSubscriberFbImpl::CreateType().createDefaultConfig(); + config.setPropertyValue(PROPERTY_NAME_SUB_JSON_CONFIG, String(VALID_JSON_1_TOPIC_0)); + CreateSubFB(config); + + auto nestedFbs = getNestedFbs(); + ASSERT_EQ(nestedFbs.getCount(), 3u); + + auto lambda = [](FunctionBlockPtr nestedFb, std::string value, std::string ts, DDSM mode) + { + EXPECT_EQ(nestedFb.getPropertyValue(PROPERTY_NAME_DEC_VALUE_NAME).asPtr().toStdString(), value); + EXPECT_EQ(static_cast(nestedFb.getPropertyValue(PROPERTY_NAME_DEC_TS_MODE).asPtr()), static_cast(mode)); + EXPECT_EQ(nestedFb.getPropertyValue(PROPERTY_NAME_DEC_TS_NAME).asPtr().toStdString(), ts); + ASSERT_EQ(nestedFb.getSignals().getCount(), 1u); + EXPECT_EQ(nestedFb.getSignals()[0].getDomainSignal().assigned(), mode != DDSM::None); + }; + + lambda(nestedFbs[0], "value", "timestamp", DDSM::ExtractFromMessage); + lambda(nestedFbs[1], "value1", "", DDSM::None); + lambda(nestedFbs[2], "value2", "", DDSM::None); +} + +TEST_F(MqttSubscriberFbTest, JsonInitDomainModeDataTransfer) +{ + auto config = MqttSubscriberFbImpl::CreateType().createDefaultConfig(); + config.setPropertyValue(PROPERTY_NAME_SUB_JSON_CONFIG, String(VALID_JSON_1_TOPIC_0)); + CreateSubFB(config); + + const auto topic = obj->getSubscribedTopic(); + auto nestedFbs = getNestedFbs(); + ASSERT_EQ(nestedFbs.getCount(), 3u); + + auto signal = nestedFbs[0].getSignals()[0]; + ASSERT_TRUE(signal.getDomainSignal().assigned()); + auto reader = daq::PacketReader(signal); + + constexpr uint64_t tsToSend = 1761567115; + constexpr double valueToSend = 12.5; + const std::string payload = R"json({"value": 12.5, "timestamp": 1761567115, "value1": 7})json"; + + onSignalsMessage({topic, std::vector(payload.begin(), payload.end()), 1, 0}); + const auto timeout = std::chrono::steady_clock::now() + std::chrono::milliseconds(1000); + + std::vector dataToReceive; + std::vector tsToReceive; + while ((!reader.getEmpty() || std::chrono::steady_clock::now() < timeout) && tsToReceive.empty()) + { + auto packet = reader.read(); + const auto dataPacket = packet.asPtrOrNull(); + if (!dataPacket.assigned()) + continue; + + dataToReceive.push_back(*(static_cast(dataPacket.getRawData()))); + const auto domainPacket = dataPacket.getDomainPacket(); + ASSERT_TRUE(domainPacket.assigned()); + tsToReceive.push_back(*(static_cast(domainPacket.getRawData()))); + } + + ASSERT_EQ(dataToReceive.size(), 1u); + EXPECT_DOUBLE_EQ(dataToReceive[0], valueToSend); + ASSERT_EQ(tsToReceive.size(), 1u); + EXPECT_EQ(tsToReceive[0], tsToSend * 1'000'000ull); +} + TEST_F(MqttSubscriberFbTest, JsonInitFromFileWrongPath) { StartUp(); From f58b3215fdfa993e7254572f267823f6b222d75d Mon Sep 17 00:00:00 2001 From: Viacheslau Date: Tue, 25 Aug 2026 09:03:11 +0200 Subject: [PATCH 2/4] decoder: promote mixed integer/double value arrays to double --- README.md | 2 + .../tests/test_mqtt_json_decoder_fb.cpp | 76 +++++++++++++++++ .../src/MqttDataWrapper.cpp | 84 ++++++++++++++----- 3 files changed, 142 insertions(+), 20 deletions(-) diff --git a/README.md b/README.md index b7773ea..8d0b6a9 100644 --- a/README.md +++ b/README.md @@ -135,6 +135,8 @@ MQTT module for the [OpenDAQ SDK](https://github.com/openDAQ/openDAQ). The modul - *DomainKey* (string) — Specifies the JSON field name (or dot-separated path for nested objects) from which the timestamp will be extracted. Dot notation is supported, e.g. `"info.timestamp"` extracts `timestamp` from inside the `info` object. This property is optional. If it is set it should be contained in the incoming JSON messages. Otherwise, a parsing error will occur. - *Unit* (string) — Specifies the unit symbol for the decoded value. This property is optional. + - **Supported value types**: the field addressed by *ValueKey* may hold a single value (integer, floating-point number or string) or an array of values. An array produces several samples at once, so in the *Extract from message* domain mode the *DomainKey* field has to be an array of the same size. The sample type of the output signal follows the type found in the message, and it is updated if the type changes. An array which mixes integers and floating-point numbers is decoded as a floating-point (`Float64`) array; an array of integers only is decoded as an `Int64` array. Any other mix of types within one array (for example numbers and strings) causes a parsing error. + Dot-notation paths support arbitrary nesting depth. For example, `"sensor.values.temperature"` traverses `sensor` → `values` → `temperature`. Example of a nested JSON MQTT message and the corresponding property values: ```json diff --git a/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp b/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp index df6c40c..d6db4ef 100644 --- a/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp +++ b/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp @@ -481,6 +481,20 @@ class MqttJsonDecoderFbHelper : public DaqTestHelper return transferData>(data, jsonDataTemplate); } + // Sends a ready-made JSON message (no placeholders) to a freshly created decoder FB + template + std::vector transferRawMessage(const std::string& json, const std::string& valueF, DDSM mode, const std::string& tsF = "") + { + const auto topic = buildTopicName(); + CreateDecoderFB(topic, valueF, mode, tsF); + + auto signal = getSignals()[0]; + auto reader = daq::PacketReader(signal); + + onSignalsMessage({topic, std::vector(json.begin(), json.end()), 1, 0}); + return read(reader, signal, 1000); + } + template std::vector, std::vector>> transferData(const std::vector, std::vector>>& data, const std::string& jsonDataTemplate) @@ -891,6 +905,68 @@ TEST_F(MqttJsonDecoderFbTest, DataTransferOneSignalIntArrayWithoutDomain) EXPECT_NE(decoderObj.getStatusContainer().getStatusMessage("ComponentStatus").toStdString().find("Parsing succeeded"), std::string::npos); } +TEST_F(MqttJsonDecoderFbTest, DataTransferMixedNumericArray) +{ + // An array which mixes integers and doubles has to be promoted to a double array + const std::string json = R"json({"value": [1, 2.5, 3, -4.25], "ts": [1761567115, 1761567116, 1761567117, 1761567118]})json"; + const std::vector expectedValues{1.0, 2.5, 3.0, -4.25}; + const std::vector expectedTs{1761567115000000ull, 1761567116000000ull, 1761567117000000ull, 1761567118000000ull}; + + auto dataToReceive = + transferRawMessage, std::vector>>(json, "value", DDSM::ExtractFromMessage, "ts"); + + ASSERT_EQ(dataToReceive.size(), 1u); + EXPECT_TRUE(equal(dataToReceive[0].first, expectedValues)); + EXPECT_EQ(dataToReceive[0].second, expectedTs); + EXPECT_EQ(getSignals()[0].getDescriptor().getSampleType(), SampleType::Float64); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Ok", decoderObj.getContext().getTypeManager())); +} + +TEST_F(MqttJsonDecoderFbTest, DataTransferMixedNumericArrayDoubleFirst) +{ + // The type of the array must not depend on the type of its first element + const std::string json = R"json({"value": [2.5, 1, 3, 4]})json"; + const std::vector expectedValues{2.5, 1.0, 3.0, 4.0}; + + auto dataToReceive = transferRawMessage>(json, "value", DDSM::None); + + ASSERT_EQ(dataToReceive.size(), 1u); + EXPECT_TRUE(equal(dataToReceive[0], expectedValues)); + EXPECT_EQ(getSignals()[0].getDescriptor().getSampleType(), SampleType::Float64); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Ok", decoderObj.getContext().getTypeManager())); +} + +TEST_F(MqttJsonDecoderFbTest, DataTransferIntArrayIsNotPromoted) +{ + // An array of integers stays an integer array + const std::string json = R"json({"value": [1, -2, 3]})json"; + const std::vector expectedValues{1, -2, 3}; + + auto dataToReceive = transferRawMessage>(json, "value", DDSM::None); + + ASSERT_EQ(dataToReceive.size(), 1u); + EXPECT_TRUE(equal(dataToReceive[0], expectedValues)); + EXPECT_EQ(getSignals()[0].getDescriptor().getSampleType(), SampleType::Int64); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Ok", decoderObj.getContext().getTypeManager())); +} + +TEST_F(MqttJsonDecoderFbTest, DataTransferMixedIncompatibleArray) +{ + // Mixing numbers with other types is still not supported + const std::string json = R"json({"value": [1, "two", 3.5]})json"; + + auto dataToReceive = transferRawMessage>(json, "value", DDSM::None); + + ASSERT_EQ(dataToReceive.size(), 0u); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Error", decoderObj.getContext().getTypeManager())); + EXPECT_NE(decoderObj.getStatusContainer().getStatusMessage("ComponentStatus").toStdString().find("Unsupported or mixed value types"), + std::string::npos); +} + TEST_F(MqttJsonDecoderFbTest, DataTransferOneSignalDoubleArrayDomainString) { std::vector, std::vector>> dataToSend; diff --git a/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp b/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp index 5fbbb8b..aee293e 100644 --- a/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp +++ b/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp @@ -38,8 +38,10 @@ template <> std::pair> parseHomogeneousArray(const rapidjson::Value::ConstArray& arr) { std::pair> result{{true, {}}, {}}; - result.second = parseHomogeneousArrayInternal< - int64_t>(arr, [](const auto& x) { return x.IsInt64() || x.IsUint64(); }, [](const auto& x) { return x.GetInt64(); }); + result.second = parseHomogeneousArrayInternal( + arr, + [](const auto& x) { return x.IsInt64() || x.IsUint64(); }, + [](const auto& x) { return x.IsInt64() ? x.GetInt64() : static_cast(x.GetUint64()); }); if (result.second.empty()) { result.first.addError("Mixed types in value array (expected integers). "); @@ -64,11 +66,12 @@ template <> std::pair> parseHomogeneousArray(const rapidjson::Value::ConstArray& arr) { std::pair> result{{true, {}}, {}}; + // Any JSON number is accepted here: an array which mixes integers and doubles is promoted to doubles result.second = - parseHomogeneousArrayInternal(arr, [](const auto& x) { return x.IsDouble(); }, [](const auto& x) { return x.GetDouble(); }); + parseHomogeneousArrayInternal(arr, [](const auto& x) { return x.IsNumber(); }, [](const auto& x) { return x.GetDouble(); }); if (result.second.empty()) { - result.first.addError("Mixed types in value array (expected doubles). "); + result.first.addError("Mixed types in value array (expected numbers). "); } return result; } @@ -86,6 +89,38 @@ std::pair> parseHomogeneousArray(const return result; } +enum class ArrayValueType +{ + Integer, + Double, + String, + Unsupported +}; + +// The type of an array is derived from all of its elements, not from the first one only: an array which +// mixes integers and doubles is a valid double array, while any other mix is not supported. +ArrayValueType detectArrayValueType(const rapidjson::Value::ConstArray& arr) +{ + bool allIntegers = true; + bool allNumbers = true; + bool allStrings = true; + + for (const auto& x : arr) + { + allIntegers = allIntegers && (x.IsInt64() || x.IsUint64()); + allNumbers = allNumbers && x.IsNumber(); + allStrings = allStrings && x.IsString(); + } + + if (allIntegers) + return ArrayValueType::Integer; + if (allNumbers) + return ArrayValueType::Double; + if (allStrings) + return ArrayValueType::String; + return ArrayValueType::Unsupported; +} + const rapidjson::Value* resolveJsonPath(const rapidjson::Value& root, const std::string& dotPath) { const rapidjson::Value* cur = &root; @@ -214,24 +249,33 @@ bool MqttDataWrapper::extractValue(ExtractionContext& ctx, const rapidjson::Valu { ctx.result.addError("Value field is an array but it is empty. "); } - else if (arr[0].IsInt64() || arr[0].IsUint64()) - { - auto [parsingStatus, out] = parseHomogeneousArray(arr); - fillContext(parsingStatus, std::move(out)); - } - else if (arr[0].IsDouble()) - { - auto [parsingStatus, out] = parseHomogeneousArray(arr); - fillContext(parsingStatus, std::move(out)); - } - else if (arr[0].IsString()) - { - auto [parsingStatus, out] = parseHomogeneousArray(arr); - fillContext(parsingStatus, std::move(out)); - } else { - ctx.result.addError(fmt::format("Unsupported value type for '{}' array. ", msgDescriptor.valueFieldName)); + switch (detectArrayValueType(arr)) + { + case ArrayValueType::Integer: + { + auto [parsingStatus, out] = parseHomogeneousArray(arr); + fillContext(parsingStatus, std::move(out)); + break; + } + case ArrayValueType::Double: + { + auto [parsingStatus, out] = parseHomogeneousArray(arr); + fillContext(parsingStatus, std::move(out)); + break; + } + case ArrayValueType::String: + { + auto [parsingStatus, out] = parseHomogeneousArray(arr); + fillContext(parsingStatus, std::move(out)); + break; + } + case ArrayValueType::Unsupported: + ctx.result.addError( + fmt::format("Unsupported or mixed value types for '{}' array. ", msgDescriptor.valueFieldName)); + break; + } } } else From bfeaf0d84c3eff75130deec83e4b5b12d68b884b Mon Sep 17 00:00:00 2001 From: Viacheslau Date: Tue, 25 Aug 2026 09:03:13 +0200 Subject: [PATCH 3/4] decoder: support escaped dots in JSON field paths --- README.md | 1 + .../src/mqtt_json_decoder_fb_impl.cpp | 12 ++-- .../tests/test_mqtt_json_decoder_fb.cpp | 66 +++++++++++++++++++ .../src/MqttDataWrapper.cpp | 59 +++++++++++++++-- 4 files changed, 127 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index 8d0b6a9..9d905f2 100644 --- a/README.md +++ b/README.md @@ -138,6 +138,7 @@ MQTT module for the [OpenDAQ SDK](https://github.com/openDAQ/openDAQ). The modul - **Supported value types**: the field addressed by *ValueKey* may hold a single value (integer, floating-point number or string) or an array of values. An array produces several samples at once, so in the *Extract from message* domain mode the *DomainKey* field has to be an array of the same size. The sample type of the output signal follows the type found in the message, and it is updated if the type changes. An array which mixes integers and floating-point numbers is decoded as a floating-point (`Float64`) array; an array of integers only is decoded as an `Int64` array. Any other mix of types within one array (for example numbers and strings) causes a parsing error. Dot-notation paths support arbitrary nesting depth. For example, `"sensor.values.temperature"` traverses `sensor` → `values` → `temperature`. + A dot which is a part of a field name has to be escaped with a backslash: `"data.a\.b"` addresses the `"a.b"` field of the `"data"` object, and `"\\"` stands for a single backslash in a field name. Inside a JSON configuration file the backslash itself has to be escaped as well, e.g. `"Value": "data.a\\.b"`. Example of a nested JSON MQTT message and the corresponding property values: ```json {"data": {"temperature": 25.68, "humidity": 72.1}, "info": {"timestamp": 1776332277}} diff --git a/modules/mqtt_streaming_module/src/mqtt_json_decoder_fb_impl.cpp b/modules/mqtt_streaming_module/src/mqtt_json_decoder_fb_impl.cpp index 8799125..8cbf6d0 100644 --- a/modules/mqtt_streaming_module/src/mqtt_json_decoder_fb_impl.cpp +++ b/modules/mqtt_streaming_module/src/mqtt_json_decoder_fb_impl.cpp @@ -32,8 +32,10 @@ FunctionBlockTypePtr MqttJsonDecoderFbImpl::CreateType() { auto builder = StringPropertyBuilder(PROPERTY_NAME_DEC_VALUE_NAME, String("")) - .setDescription("Specifies the JSON field name from which value data will be extracted. This property is required. It " - "should be contained in the incoming JSON messages. Otherwise, a parsing error will occur."); + .setDescription("Specifies the JSON field name from which value data will be extracted. Use \'.\' to address a field " + "of a nested object, e.g. \"data.temperature\"; a dot which is a part of a field name has to be " + "escaped with a backslash, e.g. \"data.a\\.b\". This property is required. It should be " + "contained in the incoming JSON messages. Otherwise, a parsing error will occur."); defaultConfig.addProperty(builder.build()); } @@ -56,8 +58,10 @@ FunctionBlockTypePtr MqttJsonDecoderFbImpl::CreateType() .setVisible(EvalValue(std::string("$") + PROPERTY_NAME_DEC_TS_MODE + " == " + std::to_string(static_cast(DSM::ExtractFromMessage)))) .setDescription( - "Specifies the JSON field name from which timestamp will be extracted. This property is " - "optional. If it is set it should be contained in the incoming JSON messages. Otherwise, a parsing error will occur."); + "Specifies the JSON field name from which timestamp will be extracted. Use \'.\' to address a field of a nested " + "object, e.g. \"info.timestamp\"; a dot which is a part of a field name has to be escaped with a backslash. " + "This property is optional. If it is set it should be contained in the incoming JSON messages. Otherwise, a " + "parsing error will occur."); defaultConfig.addProperty(builder.build()); } diff --git a/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp b/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp index d6db4ef..3a603c5 100644 --- a/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp +++ b/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp @@ -1364,6 +1364,72 @@ TEST_F(MqttJsonDecoderFbTest, NestedValueFieldWithoutDomain) EXPECT_NE(decoderObj.getStatusContainer().getStatusMessage("ComponentStatus").toStdString().find("Parsing succeeded"), std::string::npos); } +TEST_F(MqttJsonDecoderFbTest, EscapedDotInValueFieldName) +{ + // "data.a\.b" addresses the "a.b" field of the "data" object, not the "b" field of the "data.a" object + const std::string json = R"json({"data": {"a.b": 1.5, "a": {"b": 99.5}}})json"; + + auto dataToReceive = transferRawMessage(json, "data.a\\.b", DDSM::None); + + ASSERT_EQ(dataToReceive.size(), 1u); + EXPECT_DOUBLE_EQ(dataToReceive[0], 1.5); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Ok", decoderObj.getContext().getTypeManager())); +} + +TEST_F(MqttJsonDecoderFbTest, EscapedDotInTopLevelFieldName) +{ + const std::string json = R"json({"a.b": 2.5})json"; + + auto dataToReceive = transferRawMessage(json, "a\\.b", DDSM::None); + + ASSERT_EQ(dataToReceive.size(), 1u); + EXPECT_DOUBLE_EQ(dataToReceive[0], 2.5); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Ok", decoderObj.getContext().getTypeManager())); +} + +TEST_F(MqttJsonDecoderFbTest, EscapedDotInDomainFieldName) +{ + const std::string json = R"json({"value": 3.5, "info.ts": 1761567115})json"; + + auto dataToReceive = + transferRawMessage>(json, "value", DDSM::ExtractFromMessage, "info\\.ts"); + + ASSERT_EQ(dataToReceive.size(), 1u); + EXPECT_DOUBLE_EQ(dataToReceive[0].first, 3.5); + EXPECT_EQ(dataToReceive[0].second, 1761567115000000ull); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Ok", decoderObj.getContext().getTypeManager())); +} + +TEST_F(MqttJsonDecoderFbTest, EscapedBackslashInFieldName) +{ + // The JSON field name is "a\b", the escaped path for it is "a\\b" + const std::string json = R"json({"a\\b": 4.5})json"; + + auto dataToReceive = transferRawMessage(json, "a\\\\b", DDSM::None); + + ASSERT_EQ(dataToReceive.size(), 1u); + EXPECT_DOUBLE_EQ(dataToReceive[0], 4.5); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Ok", decoderObj.getContext().getTypeManager())); +} + +TEST_F(MqttJsonDecoderFbTest, UnescapedDotStaysAPathSeparator) +{ + // Without the escaping "a.b" still means the "b" field of the "a" object + const std::string json = R"json({"a.b": 5.5})json"; + + auto dataToReceive = transferRawMessage(json, "a.b", DDSM::None); + + ASSERT_EQ(dataToReceive.size(), 0u); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Error", decoderObj.getContext().getTypeManager())); + EXPECT_NE(decoderObj.getStatusContainer().getStatusMessage("ComponentStatus").toStdString().find("Parsing failed"), + std::string::npos); +} + TEST_F(MqttJsonDecoderFbTest, NestedMissingField) { const auto topic = buildTopicName(); diff --git a/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp b/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp index aee293e..471869c 100644 --- a/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp +++ b/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp @@ -121,20 +121,65 @@ ArrayValueType detectArrayValueType(const rapidjson::Value::ConstArray& arr) return ArrayValueType::Unsupported; } +bool isEscapeSequence(const std::string& dotPath, std::string::size_type pos, std::string::size_type end) +{ + return dotPath[pos] == '\\' && (pos + 1) < end && (dotPath[pos + 1] == '.' || dotPath[pos + 1] == '\\'); +} + +const rapidjson::Value* findMember(const rapidjson::Value& node, const char* name, std::string::size_type size) +{ + if (!node.IsObject()) + return nullptr; + + const auto member = node.FindMember(rapidjson::Value(rapidjson::StringRef(name, size))); + return member != node.MemberEnd() ? &member->value : nullptr; +} + +// Resolves a dot-separated path, e.g. "sensor.values.temperature". A dot which belongs to a field name has +// to be escaped with a backslash ("data.a\.b" addresses the "a.b" field of the "data" object), "\\" stands +// for a single backslash. A backslash in any other position is a part of the field name. const rapidjson::Value* resolveJsonPath(const rapidjson::Value& root, const std::string& dotPath) { const rapidjson::Value* cur = &root; + // Reused by the segments which contain escaped characters + // the segments without them are looked up in place + std::string unescaped; + std::string::size_type start = 0; while (start < dotPath.size()) { - auto dot = dotPath.find('.', start); - if (dot == std::string::npos) - dot = dotPath.size(); - const std::string segment(dotPath, start, dot - start); - if (!cur->IsObject() || !cur->HasMember(segment)) + auto end = start; + bool escaped = false; + while (end < dotPath.size() && dotPath[end] != '.') + { + if (isEscapeSequence(dotPath, end, dotPath.size())) + { + escaped = true; + ++end; + } + ++end; + } + + if (escaped) + { + unescaped.clear(); + for (auto i = start; i < end; ++i) + { + if (isEscapeSequence(dotPath, i, end)) + ++i; + unescaped += dotPath[i]; + } + cur = findMember(*cur, unescaped.data(), unescaped.size()); + } + else + { + cur = findMember(*cur, dotPath.data() + start, end - start); + } + + if (!cur) return nullptr; - cur = &(*cur)[segment]; - start = dot + 1; + + start = end + 1; } return cur; } From 0c4b1821b13cb13f6c368d4078deef9eb007b5e6 Mon Sep 17 00:00:00 2001 From: Viacheslau Date: Tue, 25 Aug 2026 09:03:17 +0200 Subject: [PATCH 4/4] decoder: support array indexes in JSON field paths --- README.md | 3 +- .../src/mqtt_json_decoder_fb_impl.cpp | 10 +- .../tests/test_mqtt_json_decoder_fb.cpp | 113 +++++++++++++ .../src/MqttDataWrapper.cpp | 155 ++++++++++++++---- 4 files changed, 246 insertions(+), 35 deletions(-) diff --git a/README.md b/README.md index 9d905f2..d54bc1c 100644 --- a/README.md +++ b/README.md @@ -138,7 +138,8 @@ MQTT module for the [OpenDAQ SDK](https://github.com/openDAQ/openDAQ). The modul - **Supported value types**: the field addressed by *ValueKey* may hold a single value (integer, floating-point number or string) or an array of values. An array produces several samples at once, so in the *Extract from message* domain mode the *DomainKey* field has to be an array of the same size. The sample type of the output signal follows the type found in the message, and it is updated if the type changes. An array which mixes integers and floating-point numbers is decoded as a floating-point (`Float64`) array; an array of integers only is decoded as an `Int64` array. Any other mix of types within one array (for example numbers and strings) causes a parsing error. Dot-notation paths support arbitrary nesting depth. For example, `"sensor.values.temperature"` traverses `sensor` → `values` → `temperature`. - A dot which is a part of a field name has to be escaped with a backslash: `"data.a\.b"` addresses the `"a.b"` field of the `"data"` object, and `"\\"` stands for a single backslash in a field name. Inside a JSON configuration file the backslash itself has to be escaped as well, e.g. `"Value": "data.a\\.b"`. + An element of an array is addressed by its index in square brackets: `"sensors[1].temperature"` takes the second element of the `sensors` array, and the indexes may be chained, e.g. `"matrix[1][0]"`. An index which is out of range, or an index applied to a field which is not an array, causes a parsing error. + A dot or an opening bracket which is a part of a field name has to be escaped with a backslash: `"data.a\.b"` addresses the `"a.b"` field of the `"data"` object, `"a\[0]"` addresses the `"a[0]"` field, and `"\\"` stands for a single backslash in a field name. Brackets which do not form a valid index (e.g. `"a[x]"`) are a part of the field name and need no escaping. Inside a JSON configuration file the backslash itself has to be escaped as well, e.g. `"Value": "data.a\\.b"`. Example of a nested JSON MQTT message and the corresponding property values: ```json {"data": {"temperature": 25.68, "humidity": 72.1}, "info": {"timestamp": 1776332277}} diff --git a/modules/mqtt_streaming_module/src/mqtt_json_decoder_fb_impl.cpp b/modules/mqtt_streaming_module/src/mqtt_json_decoder_fb_impl.cpp index 8cbf6d0..cdfb965 100644 --- a/modules/mqtt_streaming_module/src/mqtt_json_decoder_fb_impl.cpp +++ b/modules/mqtt_streaming_module/src/mqtt_json_decoder_fb_impl.cpp @@ -33,9 +33,10 @@ FunctionBlockTypePtr MqttJsonDecoderFbImpl::CreateType() auto builder = StringPropertyBuilder(PROPERTY_NAME_DEC_VALUE_NAME, String("")) .setDescription("Specifies the JSON field name from which value data will be extracted. Use \'.\' to address a field " - "of a nested object, e.g. \"data.temperature\"; a dot which is a part of a field name has to be " - "escaped with a backslash, e.g. \"data.a\\.b\". This property is required. It should be " - "contained in the incoming JSON messages. Otherwise, a parsing error will occur."); + "of a nested object, e.g. \"data.temperature\", and an index in square brackets to address an " + "element of an array, e.g. \"sensors[1].temperature\". A dot or a bracket which is a part of a " + "field name has to be escaped with a backslash, e.g. \"data.a\\.b\". This property is required. " + "It should be contained in the incoming JSON messages. Otherwise, a parsing error will occur."); defaultConfig.addProperty(builder.build()); } @@ -59,7 +60,8 @@ FunctionBlockTypePtr MqttJsonDecoderFbImpl::CreateType() " == " + std::to_string(static_cast(DSM::ExtractFromMessage)))) .setDescription( "Specifies the JSON field name from which timestamp will be extracted. Use \'.\' to address a field of a nested " - "object, e.g. \"info.timestamp\"; a dot which is a part of a field name has to be escaped with a backslash. " + "object, e.g. \"info.timestamp\", and an index in square brackets to address an element of an array, e.g. " + "\"info.ts[0]\". A dot or a bracket which is a part of a field name has to be escaped with a backslash. " "This property is optional. If it is set it should be contained in the incoming JSON messages. Otherwise, a " "parsing error will occur."); defaultConfig.addProperty(builder.build()); diff --git a/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp b/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp index 3a603c5..513902a 100644 --- a/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp +++ b/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp @@ -1430,6 +1430,119 @@ TEST_F(MqttJsonDecoderFbTest, UnescapedDotStaysAPathSeparator) std::string::npos); } +TEST_F(MqttJsonDecoderFbTest, ArrayIndexInValueFieldPath) +{ + const std::string json = R"json({"sensors": [{"temp": 1.5}, {"temp": 2.5}, {"temp": 3.5}]})json"; + + auto dataToReceive = transferRawMessage(json, "sensors[1].temp", DDSM::None); + + ASSERT_EQ(dataToReceive.size(), 1u); + EXPECT_DOUBLE_EQ(dataToReceive[0], 2.5); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Ok", decoderObj.getContext().getTypeManager())); +} + +TEST_F(MqttJsonDecoderFbTest, ArrayIndexAtTheEndOfThePath) +{ + const std::string json = R"json({"data": {"values": [1.5, 2.5, 3.5]}})json"; + + auto dataToReceive = transferRawMessage(json, "data.values[2]", DDSM::None); + + ASSERT_EQ(dataToReceive.size(), 1u); + EXPECT_DOUBLE_EQ(dataToReceive[0], 3.5); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Ok", decoderObj.getContext().getTypeManager())); +} + +TEST_F(MqttJsonDecoderFbTest, ChainedArrayIndexes) +{ + const std::string json = R"json({"matrix": [[1.5, 2.5], [3.5, 4.5]]})json"; + + auto dataToReceive = transferRawMessage(json, "matrix[1][0]", DDSM::None); + + ASSERT_EQ(dataToReceive.size(), 1u); + EXPECT_DOUBLE_EQ(dataToReceive[0], 3.5); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Ok", decoderObj.getContext().getTypeManager())); +} + +TEST_F(MqttJsonDecoderFbTest, ArrayIndexSelectsAnArrayOfSamples) +{ + // An indexed element may be an array itself, and then it produces several samples + const std::string json = R"json({"matrix": [[1.5, 2.5], [3.5, 4.5]]})json"; + + auto dataToReceive = transferRawMessage>(json, "matrix[0]", DDSM::None); + + ASSERT_EQ(dataToReceive.size(), 1u); + EXPECT_TRUE(equal(dataToReceive[0], std::vector{1.5, 2.5})); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Ok", decoderObj.getContext().getTypeManager())); +} + +TEST_F(MqttJsonDecoderFbTest, ArrayIndexInDomainFieldPath) +{ + const std::string json = R"json({"value": 3.5, "info": {"ts": [1761567115, 1761567116]}})json"; + + auto dataToReceive = + transferRawMessage>(json, "value", DDSM::ExtractFromMessage, "info.ts[1]"); + + ASSERT_EQ(dataToReceive.size(), 1u); + EXPECT_DOUBLE_EQ(dataToReceive[0].first, 3.5); + EXPECT_EQ(dataToReceive[0].second, 1761567116000000ull); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Ok", decoderObj.getContext().getTypeManager())); +} + +TEST_F(MqttJsonDecoderFbTest, ArrayIndexOutOfRange) +{ + const std::string json = R"json({"values": [1.5, 2.5]})json"; + + auto dataToReceive = transferRawMessage(json, "values[2]", DDSM::None); + + ASSERT_EQ(dataToReceive.size(), 0u); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Error", decoderObj.getContext().getTypeManager())); + EXPECT_NE(decoderObj.getStatusContainer().getStatusMessage("ComponentStatus").toStdString().find("Parsing failed"), + std::string::npos); +} + +TEST_F(MqttJsonDecoderFbTest, ArrayIndexOnNonArrayField) +{ + const std::string json = R"json({"value": 1.5})json"; + + auto dataToReceive = transferRawMessage(json, "value[0]", DDSM::None); + + ASSERT_EQ(dataToReceive.size(), 0u); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Error", decoderObj.getContext().getTypeManager())); +} + +TEST_F(MqttJsonDecoderFbTest, EscapedBracketInFieldName) +{ + // The field is named "a[0]", so the bracket has to be escaped to keep it a part of the name + const std::string json = R"json({"a[0]": 6.5, "a": [7.5]})json"; + + auto dataToReceive = transferRawMessage(json, "a\\[0]", DDSM::None); + + ASSERT_EQ(dataToReceive.size(), 1u); + EXPECT_DOUBLE_EQ(dataToReceive[0], 6.5); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Ok", decoderObj.getContext().getTypeManager())); +} + +TEST_F(MqttJsonDecoderFbTest, BracketsWhichAreNotAnIndexStayInTheFieldName) +{ + // "[x]" is not a valid index, so the whole segment is a field name and needs no escaping + const std::string json = R"json({"a[x]": 8.5})json"; + + auto dataToReceive = transferRawMessage(json, "a[x]", DDSM::None); + + ASSERT_EQ(dataToReceive.size(), 1u); + EXPECT_DOUBLE_EQ(dataToReceive[0], 8.5); + ASSERT_EQ(decoderObj.getStatusContainer().getStatus("ComponentStatus"), + Enumeration("ComponentStatusType", "Ok", decoderObj.getContext().getTypeManager())); +} + TEST_F(MqttJsonDecoderFbTest, NestedMissingField) { const auto topic = buildTopicName(); diff --git a/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp b/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp index 471869c..3f26ff3 100644 --- a/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp +++ b/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include @@ -123,7 +124,99 @@ ArrayValueType detectArrayValueType(const rapidjson::Value::ConstArray& arr) bool isEscapeSequence(const std::string& dotPath, std::string::size_type pos, std::string::size_type end) { - return dotPath[pos] == '\\' && (pos + 1) < end && (dotPath[pos + 1] == '.' || dotPath[pos + 1] == '\\'); + if (dotPath[pos] != '\\' || (pos + 1) >= end) + return false; + + const char next = dotPath[pos + 1]; + return next == '.' || next == '[' || next == '\\'; +} + +// Reads the "[]" group which starts at pos and returns the position right after it, +// or npos if there is no well-formed group there +std::string::size_type +parseIndex(const std::string& dotPath, std::string::size_type pos, std::string::size_type end, rapidjson::SizeType& index) +{ + if (pos >= end || dotPath[pos] != '[') + return std::string::npos; + + constexpr uint64_t maxIndex = std::numeric_limits::max(); + auto digit = pos + 1; + uint64_t value = 0; + while (digit < end && dotPath[digit] >= '0' && dotPath[digit] <= '9') + { + value = value * 10 + static_cast(dotPath[digit] - '0'); + if (value > maxIndex) + return std::string::npos; + ++digit; + } + + // an empty index ("[]") or a non-digit inside the brackets is not an index + if (digit == (pos + 1) || digit >= end || dotPath[digit] != ']') + return std::string::npos; + + index = static_cast(value); + return digit + 1; +} + +struct PathSegment +{ + std::string::size_type nameEnd; // the field name is [start, nameEnd) + std::string::size_type end; // the chain of indexes is [nameEnd, end), the next segment starts at end + 1 + bool nameHasEscapes; +}; + +// Reads the segment which starts at `start` and splits it into the field name and the chain of array +// indexes which follows it: "sensors[1][0]" is the name "sensors" plus two indexes. The chain has to +// occupy the whole tail of the segment, so any ordinary character cancels the chain started before it and +// makes it a part of the name: "a[x]" and "a[0]b" are plain field names. +PathSegment readSegment(const std::string& dotPath, std::string::size_type start) +{ + constexpr auto noChain = std::string::npos; + + const auto pathEnd = dotPath.size(); + auto chainStart = noChain; + bool nameHasEscapes = false; + + auto pos = start; + while (pos < pathEnd && dotPath[pos] != '.') + { + if (isEscapeSequence(dotPath, pos, pathEnd)) + { + nameHasEscapes = true; + chainStart = noChain; + pos += 2; + continue; + } + + if (dotPath[pos] == '[') + { + rapidjson::SizeType index = 0; + if (const auto afterIndex = parseIndex(dotPath, pos, pathEnd, index); afterIndex != std::string::npos) + { + if (chainStart == noChain) + chainStart = pos; + pos = afterIndex; + continue; + } + } + + chainStart = noChain; + ++pos; + } + + return {chainStart != noChain ? chainStart : pos, pos, nameHasEscapes}; +} + +// Copies [from, to) into `out` dropping the escaping backslashes +void unescape(const std::string& dotPath, std::string::size_type from, std::string::size_type to, std::string& out) +{ + out.clear(); + for (auto pos = from; pos < to; ++pos) + { + if (isEscapeSequence(dotPath, pos, to)) + ++pos; + out += dotPath[pos]; + } } const rapidjson::Value* findMember(const rapidjson::Value& node, const char* name, std::string::size_type size) @@ -135,51 +228,53 @@ const rapidjson::Value* findMember(const rapidjson::Value& node, const char* nam return member != node.MemberEnd() ? &member->value : nullptr; } -// Resolves a dot-separated path, e.g. "sensor.values.temperature". A dot which belongs to a field name has -// to be escaped with a backslash ("data.a\.b" addresses the "a.b" field of the "data" object), "\\" stands -// for a single backslash. A backslash in any other position is a part of the field name. +// Applies the chain of array indexes [from, to) to the node +const rapidjson::Value* +applyIndexes(const rapidjson::Value* node, const std::string& dotPath, std::string::size_type from, std::string::size_type to) +{ + auto pos = from; + while (node != nullptr && pos < to) + { + rapidjson::SizeType index = 0; + const auto afterIndex = parseIndex(dotPath, pos, to, index); + if (afterIndex == std::string::npos) + return nullptr; + + node = (node->IsArray() && index < node->Size()) ? &(*node)[index] : nullptr; + pos = afterIndex; + } + return node; +} + +// Resolves a dot-separated path, e.g. "sensor.values.temperature". An element of an array is addressed by +// its index, e.g. "sensors[1].temperature" or "matrix[1][0]". A dot or an opening bracket which belongs to +// a field name has to be escaped with a backslash ("data.a\.b" addresses the "a.b" field of the "data" +// object), "\\" stands for a single backslash. A backslash in any other position is a part of the name. const rapidjson::Value* resolveJsonPath(const rapidjson::Value& root, const std::string& dotPath) { const rapidjson::Value* cur = &root; - // Reused by the segments which contain escaped characters - // the segments without them are looked up in place - std::string unescaped; + std::string unescapedName; std::string::size_type start = 0; while (start < dotPath.size()) { - auto end = start; - bool escaped = false; - while (end < dotPath.size() && dotPath[end] != '.') - { - if (isEscapeSequence(dotPath, end, dotPath.size())) - { - escaped = true; - ++end; - } - ++end; - } + const auto segment = readSegment(dotPath, start); - if (escaped) + if (segment.nameHasEscapes) { - unescaped.clear(); - for (auto i = start; i < end; ++i) - { - if (isEscapeSequence(dotPath, i, end)) - ++i; - unescaped += dotPath[i]; - } - cur = findMember(*cur, unescaped.data(), unescaped.size()); + unescape(dotPath, start, segment.nameEnd, unescapedName); + cur = findMember(*cur, unescapedName.data(), unescapedName.size()); } else { - cur = findMember(*cur, dotPath.data() + start, end - start); + cur = findMember(*cur, dotPath.data() + start, segment.nameEnd - start); } - if (!cur) + cur = applyIndexes(cur, dotPath, segment.nameEnd, segment.end); + if (cur == nullptr) return nullptr; - start = end + 1; + start = segment.end + 1; } return cur; }