diff --git a/README.md b/README.md index b7773ea..d54bc1c 100644 --- a/README.md +++ b/README.md @@ -135,7 +135,11 @@ 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`. + 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 8799125..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 @@ -32,8 +32,11 @@ 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\", 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()); } @@ -56,8 +59,11 @@ 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\", 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/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_json_decoder_fb.cpp b/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp index df6c40c..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 @@ -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; @@ -1288,6 +1364,185 @@ 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, 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/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(); diff --git a/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp b/shared/mqtt_streaming_protocol/src/MqttDataWrapper.cpp index 5fbbb8b..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 @@ -38,8 +39,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 +67,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,20 +90,191 @@ 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; +} + +bool isEscapeSequence(const std::string& dotPath, std::string::size_type pos, std::string::size_type end) +{ + 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) +{ + if (!node.IsObject()) + return nullptr; + + const auto member = node.FindMember(rapidjson::Value(rapidjson::StringRef(name, size))); + return member != node.MemberEnd() ? &member->value : nullptr; +} + +// 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; + std::string unescapedName; + 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)) + const auto segment = readSegment(dotPath, start); + + if (segment.nameHasEscapes) + { + unescape(dotPath, start, segment.nameEnd, unescapedName); + cur = findMember(*cur, unescapedName.data(), unescapedName.size()); + } + else + { + cur = findMember(*cur, dotPath.data() + start, segment.nameEnd - start); + } + + cur = applyIndexes(cur, dotPath, segment.nameEnd, segment.end); + if (cur == nullptr) return nullptr; - cur = &(*cur)[segment]; - start = dot + 1; + + start = segment.end + 1; } return cur; } @@ -214,24 +389,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