Merge pull request #8 from openDAQ/json-subscriber-reworking

Json subscriber reworking
- Rework of JSON subscription flow: JSON receiver FB no longer parses all signals itself — it now creates nested JSON decoder function blocks (jsonDecoderMqttFb) that parse individual values/timestamps and produce signals.
- New JSON decoder FB: added mqtt_json_decoder_fb_impl.{h,cpp} to extract a single value + optional timestamp from incoming JSON and emit data/domain samples.
- JSON config handling changed: SignalList (big JSON string) → JsonConfig (string) and JsonConfigFile (path). JSON FB reads config (or file), extracts topic and per-signal descriptors and creates nested decoders. New "Topic" property.
- Stronger topic validation and subscription QoS: added PROPERTY_NAME_SUB_QOS / DEFAULT_SUB_QOS; subscription/unsubscribe now use the configured QoS.
- Status system refactor: introduced a generic StatusHelper to manage enumerated statuses (component / subscription / parsing / publishing / setting), replacing many manual EnumerationWithIntValue usages.
Publisher FB improvements: cleaner status reporting (SignalStatus / PublishingStatus / SettingStatus), validation of properties (QoS, topic when sharedTs or multi), rename of publishing modes to descriptive enum names, and more robust publishing status tracking.
- Raw FB changes: Topic property replaced SignalList.
- MqttDataWrapper changes: JSON parsing & descriptor extraction reworked (now returns per-signal MqttMsgDescriptor with unit info), value/timestamp extraction refactored; API signatures changed.
- Tests and examples: many unit tests added / updated (including new test for json decoder), new example JSON files for tests, updates to examples (custom-mqtt-sub, raw-mqtt-sub, ref-dev-mqtt-pub) to use the new FB/type/property names and behavior.
- Renames & cleanups: several handler/source/header files renamed to clearer names (single_handler → atomic_signal_atomic_sample_handler, multisingle → atomic_signal_sample_arr_handler, multiple_shared → group_signal_shared_ts_handler, multiple → signal_arr_atomic_sample_handler), constants and FB type IDs removed leading '@' (e.g. @rootMqttFb → rootMqttFb), and general code cleanup (status helper, removal of old helper functions).

Author: viacheslauK

README.md

@@ -2,17 +2,17 @@
 
 ## Description
 
-MQTT module for the [OpenDAQ SDK](https://github.com/openDAQ/openDAQ). The module is designed for software communication via the *MQTT 3.1.1* protocol using an external broker. It allows publishing and subscribing to openDAQ signal data over MQTT. The module consists of four key openDAQ components: the *MQTT device* and its nested function blocks — the *publisher* (**@publisherMqttFb**), the *raw subscriber* (**@rawMqttFb**), and the *JSON subscriber* (**@jsonMqttFb**).   
+MQTT module for the [OpenDAQ SDK](https://github.com/openDAQ/openDAQ). The module is designed for software communication via the *MQTT 3.1.1* protocol using an external broker. It allows publishing and subscribing to openDAQ signal data over MQTT. The module consists of five key openDAQ components: the *MQTT root function block* (**rootMqttFb**) and its nested function blocks — the *publisher* (**publisherMqttFb**) with its nested block *JSON decoder* (**jsonDecoderMqttFb**) , the *raw subscriber* (**rawMqttFb**), and the *JSON subscriber* (**jsonMqttFb**).   
 
 ### Functional
 - Connecting to an MQTT broker;
 - Publishing openDAQ signals as MQTT messages (*publisher FB*);
-- Subscribing to MQTT topics and converting incoming messages into openDAQ signals (*raw FB and JSON FB*);
+- Subscribing to MQTT topics and converting incoming messages into openDAQ signals (*raw FB and JSON FB + JSON decoder FB*);
 - Support for multiple message types and formats for both publishing and subscribing;
 - A set of examples and *gtests* for verifying functionality.
 
 ### Key components
-1) **MQTT root Function Block (@rootMqttFb)**:
+1) **MQTT root Function Block (rootMqttFb)**:
    - **Where**: *mqtt_streaming_module/src/mqtt_root_fb_impl.cpp, include/mqtt_streaming_module/...*
    - **Purpose**: Represents the MQTT broker as an openDAQ function block - the connection point through which function blocks are created.
    - **Main properties:**
@@ -21,7 +21,7 @@ MQTT module for the [OpenDAQ SDK](https://github.com/openDAQ/openDAQ). The modul
       - *MqttUsername* (string) — Username for MQTT broker authentication. By default, it is empty.
       - *MqttPassword* (string) — Password for MQTT broker authentication. By default, it is empty.
       - *ConnectTimeout* (integer) — Timeout in milliseconds for the initial connection to the MQTT broker. If the connection fails, an exception is thrown. By default, it is set to *3000 ms*.
-2) **Publisher MQTT Function Block (@publisherMqttFb)**:
+2) **Publisher MQTT Function Block (publisherMqttFb)**:
    - **Where**: *include/mqtt_streaming_module/mqtt_publisher_fb_impl.h, src/mqtt_publisher_fb_impl.cpp*
    - **Purpose**: Publishes openDAQ signal data to MQTT topics. There are **four** general data publishing schemes:
       1) One MQTT message per signal / one message per sample / one topic per signal / one timestamp for each sample. Example: *{"AI0": 1.1, "timestamp": 1763716736100000}*
@@ -37,6 +37,7 @@ MQTT module for the [OpenDAQ SDK](https://github.com/openDAQ/openDAQ). The modul
    - **Main properties**:
       - *TopicMode* (list) — Selects whether to publish all signals to separate MQTT topics (one per signal, *single-topic mode*) or to a single topic (*multiple-topic mode*), one for all signals. Choose *0* for *single-topic* mode and *1* for *multiple-topic* mode. By default, it is set to *single-topic* mode.
       - *MqttQoS* (integer) — MQTT Quality of Service level. It can be *0* (at most once), *1* (at least once), or *2* (exactly once). By default, it is set to *1*.
+      - *Topic* (string) — Topic name for publishing in multiple-topic mode. If left empty, the Publisher's *Global ID* is used as the topic name.
       - *SharedTimestamp* (bool) — Enables the use of a shared timestamp for all signals when publishing in *multiple-topic* mode. By default, it is set to *false*.
       - *GroupValues* (bool) — Enables the use of a sample pack for a signal when publishing in *single-topic* mode. By default, it is set to *false*.
       - *UseSignalNames* (bool) — Uses signal names as JSON field names instead of Global IDs. By default, it is set to *false*.
@@ -50,18 +51,22 @@ MQTT module for the [OpenDAQ SDK](https://github.com/openDAQ/openDAQ). The modul
         4) *TopicMode(1), SharedTimestamp(true), GroupValues(false)*;
 
 
-3) **Raw MQTT Function Block (@rawMqttFb)**:
+3) **Raw MQTT Function Block (rawMqttFb)**:
 
    - **Where**: *include/mqtt_streaming_module/mqtt_raw_receiver_fb_impl.h, src/mqtt_raw_receiver_fb_impl.cpp*
    - **Purpose**: Subscribes to raw MQTT messages and converts them into openDAQ signals (binary data) without any parsing — suitable for binary/unstructured messages or simple numeric values.
    - **Main properties**:
-      - *SignalList* (list of strings) — List of MQTT topics to subscribe to for receiving raw binary data.
+      - *Topic* (string) — MQTT topic to subscribe to for receiving raw binary data.
+      - *MqttQoS* (integer) — MQTT Quality of Service level. It can be *0* (at most once), *1* (at least once), or *2* (exactly once). By default, it is set to *1*.
 
-4) **JSON MQTT Function Block (@jsonMqttFb)**:
+4) **JSON MQTT Function Block (jsonMqttFb)**:
    - **Where**: *include/mqtt_streaming_module/mqtt_json_receiver_fb_impl.h, src/mqtt_json_receiver_fb_impl.cpp*
-   - **Purpose**: Subscribes to MQTT topics, extracts values and timestamps from MQTT JSON messages, and converts them into openDAQ signal data samples.
+   - **Purpose**: Subscribes to MQTT topics, extracts values and timestamps from MQTT JSON messages via nested *JSON decoder MQTT Function Blocks*.
    - **Main properties**:
-      - *SignalList* (string) — **JSON configuration string** that defines the list of MQTT topics and the corresponding signals to subscribe to. A typical string structure:
+      - *Topic* (string) — MQTT topic to subscribe to for receiving JSON data.
+      - *MqttQoS* (integer) — MQTT Quality of Service level. It can be *0* (at most once), *1* (at least once), or *2* (exactly once). By default, it is set to *1*.
+      - *JsonConfigFile* (string) — path to file with **JSON configuration string**. See the *JsonConfig* property for more details. This property could be set only at creation. It is not visible.
+      - *JsonConfig* (string) — **JSON configuration string** that defines the MQTT topic and the corresponding signals to subscribe to. This property could be set only at creation. It is not visible. A typical string structure:
       ```json
       {
         "<topic>":[
@@ -71,8 +76,8 @@ MQTT module for the [OpenDAQ SDK](https://github.com/openDAQ/openDAQ). The modul
                     "Timestamp":"<field_name_in_JSON_MQTT_message_for_extracting_sample_timestamp>",
                     "Unit":[
                         "<unit_symbol>",
-                        "<unit_name>",
-                        "<unit_quantity>"
+                        "<unit_name>",      // is not used
+                        "<unit_quantity>"   // is not used
                     ]
                 }
             },
@@ -112,11 +117,21 @@ MQTT module for the [OpenDAQ SDK](https://github.com/openDAQ/openDAQ). The modul
         ]
     }
     ```
-      In this example, the *JSON MQTT Function Block* creates 3 signals, subscribes to the *"/mirip/UNet3AC2/sensor/data"* topic, and extracts 3 signal samples from each message (one sample per signal). The signals are named *“temp”*, *“humidity”*, and *“tds”*. The *“temp”* signal is created with a domain signal because the *“Timestamp”* field is present. Each domain-signal sample is extracted from the *“ts”* field of the JSON MQTT message. The value of the *“ts”* field (the timestamp field) may be in **ISO8601** format or **Unix epoch time** in seconds, milliseconds, or microseconds. The value of the *“temp”* signal sample is extracted from the *“temp”* field of the JSON message. The unit of the values is “°C”.
+      In this example, the *JSON MQTT Function Block* creates 3 nested *jsonDecoderMqttFb*, subscribes to the *"/mirip/UNet3AC2/sensor/data"* topic, and extracts 3 signal samples from each message (one sample per *jsonDecoderMqttFb*). The signals are named *“temp”*, *“humidity”*, and *“tds”*. The *“temp”* signal is created with a domain signal because the *“Timestamp”* field is present. Each domain-signal sample is extracted from the *“ts”* field of the JSON MQTT message. The value of the *“ts”* field (the timestamp field) may be in **ISO8601** format or **Unix epoch time** in seconds, milliseconds, or microseconds. The value of the *“temp”* signal sample is extracted from the *“temp”* field of the JSON message. The unit of the values is “°C”.
       Example of JSON MQTT message for this configuration:
       ```json
       {"ts":"2025-10-08 20:35:43", "bdn":"SanbonFishTank3", "temp":27.20,"humi":72.40, "tds_value":275.22, "fan_status":"off", "auto_mode":"on", "fan_comp":"26.3", "humi_comp":"55"}
       ```
+
+5) **JSON decoder MQTT Function Block (jsonDecoderMqttFb)**:
+
+   - **Where**: *include/mqtt_streaming_module/mqtt_json_decoder_fb_impl.h, src/mqtt_json_decoder_fb_impl.cpp*
+   - **Purpose**: To parse JSON string data to extract a value and a timestamp, and to send data and domain samples based on this data.
+   - **Main properties**:
+      - *ValueName* (string) — indicates which JSON field contains the sample value.
+      - *TimestampName* (string) — indicates which JSON field contains the timestamp.
+      - *Unit* (string) — describes the unit symbol of the decoded signal value.
+      - *SignalName* (string) — specifies the name to assign to the signal created by a *jsonDecoderMqttFb*.
 ---
 
 ## Building MQTTStreamingModule
@@ -164,13 +179,13 @@ cmake --build .
 ## Examples   
 
 There are 3 example C++ application:
- - **custom-mqtt-sub** - demonstrates how to work with the *JSON MQTT FB*. The application creates an *MQTT root FB* and a *JSON MQTT FB* to receive JSON MQTT messages, parse them, and create openDAQ signals to send the parsed data. The application also creates *packet readers* for all FB signals and prints the samples to standard output. The *SignalList* property of the JSON MQTT FB is set to the value read from a file whose path is provided as a command-line argument when the application starts (see the **Key components** section). Usage:
+ - **custom-mqtt-sub** - demonstrates how to work with the *JSON receiver MQTT FB* and *JSON decoder MQTT FB*. The application creates an *MQTT root FB* and a *JSON MQTT FB* to receive JSON MQTT messages, parse them, and create openDAQ signals to send the parsed data. The application also creates *packet readers* for all FB signals and prints the samples to standard output. The *JsonConfigFile* property of the JSON MQTT FB is set to the value of path whose is provided as a command-line argument when the application starts (see the **Key components** section). Usage:
   ```bash
  ./custom-mqtt-sub --address broker.emqx.io examples/custom-mqtt-sub/public-example0.json
  ```  
- - **raw-mqtt-sub**  - demonstrates how to work with the *raw MQTT FB*. The application creates an *MQTT root FB* and a *raw MQTT FB* to receive MQTT messages and create openDAQ signals to send the data as binary packets. The application also creates packet readers for all FB signals and prints the binary packets as strings to standard output. The *SignalList* property of the raw MQTT FB is filled from the application arguments. Usage:
+ - **raw-mqtt-sub**  - demonstrates how to work with the *raw MQTT FB*. The application creates an *MQTT root FB* and a *raw MQTT FB* to receive MQTT messages and create openDAQ signals to send the data as binary packets. The application also creates packet readers for all FB signals and prints the binary packets as strings to standard output. The *Topic* property of the raw MQTT FB is filled from the application arguments. Usage:
  ```bash
- ./raw-mqtt-sub --address broker.emqx.io /agvstate /mirip/UNet3AC2/sensor/data
+ ./raw-mqtt-sub --address broker.emqx.io /mirip/UNet3AC2/sensor/data
  ```
  - **ref-dev-mqtt-pub** - demonstrates how to work with the *publisher MQTT FB*. The application creates an *openDAQ ref-device* with four channels, an *MQTT root FB*, and a *publisher MQTT FB* to publish JSON MQTT messages with the channels’ data. The properties of the *publisher MQTT FB* are set according to the selected mode, which can be specified via the *--mode* option. Posible values are:
     - 0 - One MQTT message per signal / one message per sample / one topic per signal / one timestamp for each sample;

examples/custom-mqtt-sub/pub-config.json

@@ -22,7 +22,7 @@ std::string to_string(uint64_t ts)
     system_clock::time_point tp = system_clock::time_point(microseconds(ts));
 
     auto tt = system_clock::to_time_t(tp);
-    std::tm tm = *std::localtime(&tt);
+    std::tm tm = *std::gmtime(&tt);
 
     auto us = duration_cast<milliseconds>(tp.time_since_epoch()) % 1000;
 
@@ -123,29 +123,37 @@ int main(int argc, char* argv[])
 
     // Create OpenDAQ instance and add MQTT broker FB
     const InstancePtr instance = InstanceBuilder().addModulePath(MODULE_PATH).build();
-    const std::string rootFbName = "@rootMqttFb";
+    const std::string rootFbName = "RootMqttFb";
     auto rootFbConfig = instance.getAvailableFunctionBlockTypes().get(rootFbName).createDefaultConfig();
     rootFbConfig.setPropertyValue("MqttBrokerAddress", appConfig.brokerAddress);
     auto brokerFB = instance.addFunctionBlock(rootFbName, rootFbConfig);
     auto availableFbs = brokerFB.getAvailableFunctionBlockTypes();
 
-    const std::string fbName = "@jsonMqttFb";
-    std::cout << "Try to add the " << fbName << std::endl;
+    const std::string jsonFbName = "JsonSubscriberMqttFb";
+    std::cout << "Try to add the " << jsonFbName << std::endl;
 
-    // Read JSON function block configuration from file and fill out the function block config
-    const std::string jsonConfig = readFileToString(appConfig.configFilePath);
-    auto config = availableFbs.get(fbName).createDefaultConfig();
-    config.setPropertyValue("SignalList", jsonConfig);
+    auto config = availableFbs.get(jsonFbName).createDefaultConfig();
+    config.setPropertyValue("JsonConfigFile", appConfig.configFilePath);
 
     // Add the JSON function block to the broker FB
-    daq::FunctionBlockPtr jsonFb = brokerFB.addFunctionBlock(fbName, config);
+    daq::FunctionBlockPtr jsonFb = brokerFB.addFunctionBlock(jsonFbName, config);
 
     // Create packet readers for all signals
-    const auto signals = jsonFb.getSignals();
-    std::map<std::string, PacketReaderPtr> packetReaders;
+    auto signals = List<daq::ISignal>();
+    const auto fbs = jsonFb.getFunctionBlocks();
+    for (const auto& fb : fbs)
+    {
+        const auto sig = fb.getSignals();
+        for (const auto& s : sig)
+        {
+            signals.pushBack(s);
+        }
+    }
+
+    std::vector<std::pair<std::string, PacketReaderPtr>> packetReaders;
     for (const auto& s : signals)
     {
-        packetReaders.emplace(std::pair<std::string, PacketReaderPtr>(s.getName().toStdString(), daq::PacketReader(s)));
+        packetReaders.push_back(std::pair<std::string, PacketReaderPtr>(s.getName().toStdString(), daq::PacketReader(s)));
     }
 
     // Start a thread to read packets from the readers

examples/custom-mqtt-sub/src/custom-mqtt-sub.cpp

@@ -55,13 +55,13 @@ int main(int argc, char* argv[])
 
     // Create OpenDAQ instance and add MQTT broker FB
     const InstancePtr instance = InstanceBuilder().addModulePath(MODULE_PATH).build();
-    const std::string rootFbName = "@rootMqttFb";
+    const std::string rootFbName = "RootMqttFb";
     auto rootFbConfig = instance.getAvailableFunctionBlockTypes().get(rootFbName).createDefaultConfig();
     rootFbConfig.setPropertyValue("MqttBrokerAddress", appConfig.brokerAddress);
     auto brokerFB = instance.addFunctionBlock(rootFbName, rootFbConfig);
     auto availableFbs = brokerFB.getAvailableFunctionBlockTypes();
 
-    const std::string fbName = "@rawMqttFb";
+    const std::string fbName = "RawSubscriberMqttFb";
     std::cout << "Try to add the " << fbName << std::endl;
 
     // Create RAW function block configuration