Root client FB as a device

README.md

@@ -2,7 +2,7 @@
 
 ## 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 five key openDAQ components: the *MQTT client function block* (**MQTTClientFB**) and its nested function blocks — the *publisher* (**MQTTJSONPublisherFB**) and the *subscriber* (**MQTTSubscriberFB**) with its nested block *JSON decoder* (**MQTTJSONDecoderFB**).   
+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 client device* (**OpenDAQMQTTDevice**) and its nested function blocks — the *publisher* (**MQTTJSONPublisherFB**) and the *subscriber* (**MQTTSubscriberFB**) with its nested block *JSON decoder* (**MQTTJSONDecoderFB**).   
 
 ### Functional
 - Connecting to an MQTT broker;
@@ -12,15 +12,16 @@ MQTT module for the [OpenDAQ SDK](https://github.com/openDAQ/openDAQ). The modul
 - A set of examples and *gtests* for verifying functionality.
 
 ### Key components
-1) **MQTT client Function Block (MQTTClientFB)**:
-   - **Where**: *mqtt_streaming_module/src/mqtt_client_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.
+1) **MQTT client Device (OpenDAQMQTTDevice)**:
+   - **Where**: *mqtt_streaming_module/src/mqtt_client_device_impl.cpp, include/mqtt_streaming_module/...*
+   - **Purpose**: Represents the MQTT broker as an openDAQ device - the connection point through which function blocks are created.
+   - **Connection string**: `daq.mqtt://<host>[:<port>]`, for example `daq.mqtt://127.0.0.1:1883`. The device is added with `instance.addDevice(...)`. The broker host always comes from the connection string; a port given there takes precedence over the *Port* property.
    - **Main properties:**
-      - *BrokerAddress* (string) — MQTT broker address. It can be an IP address or a hostname. By default, it is set to *"127.0.0.1"*.
-      - *BrokerPort* (integer) — Port number for the MQTT broker connection. By default, it is set to *1883*.
+      - *Port* (integer) — Port number for the MQTT broker connection, used only when the connection string carries no port. By default, it is set to *1883*.
       - *Username* (string) — Username for MQTT broker authentication. By default, it is empty.
       - *Password* (string) — Password for MQTT broker authentication. By default, it is empty.
       - *ConnectionTimeout* (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*.
+   - **Connection status**: exposed through the device's connection status container under the *ConfigurationStatus* alias, e.g. `device.getConnectionStatusContainer().getStatus("ConfigurationStatus")`.
 2) **MQTT publisher Function Block (MQTTJSONPublisherFB)**:
    - **Where**: *include/mqtt_streaming_module/mqtt_publisher_fb_impl.h, src/mqtt_publisher_fb_impl.cpp*
    - **Purpose**: Publishes openDAQ signal data to MQTT topics.  
@@ -192,15 +193,15 @@ cmake --build .
 ## Examples   
 
 There are 3 example C++ application:
- - **custom-mqtt-sub** - demonstrates how to work with the *MQTT subscriber MQTT FB* and *MQTT JSON decoder MQTT FB*. The application creates an *MQTTClientFB* and a *MQTTSubscriberFB* with nested *MQTTJSONDecoderFB* function blocks 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 *MQTTSubscriberFB* 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:
+ - **custom-mqtt-sub** - demonstrates how to work with the *MQTT subscriber MQTT FB* and *MQTT JSON decoder MQTT FB*. The application creates an *OpenDAQMQTTDevice* and a *MQTTSubscriberFB* with nested *MQTTJSONDecoderFB* function blocks 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 *MQTTSubscriberFB* 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 *MQTT subscriber MQTT FB* in a raw mode (binary data without parsing). The application creates an *MQTTClientFB* and a *MQTTSubscriberFB* 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 *MQTTSubscriberFB* is filled from the application arguments. Usage:
+ - **raw-mqtt-sub**  - demonstrates how to work with the *MQTT subscriber MQTT FB* in a raw mode (binary data without parsing). The application creates an *OpenDAQMQTTDevice* and a *MQTTSubscriberFB* 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 *MQTTSubscriberFB* is filled from the application arguments. Usage:
  ```bash
  ./raw-mqtt-sub --address broker.emqx.io /mirip/UNet3AC2/sensor/data
  ```
- - **ref-dev-mqtt-pub** - demonstrates how to work with the *MQTTJSONPublisherFB*. The application creates an *openDAQ ref-device* with four channels, an *MQTTClientFB*, and a *MQTTJSONPublisherFB* to publish JSON MQTT messages with the channels’ data. The properties of the *MQTTJSONPublisherFB* are set according to the selected mode, which can be specified via the *--mode* option, and array size,  which can be specified via the *--array* option with size. Posible *--mode* option values are:
+ - **ref-dev-mqtt-pub** - demonstrates how to work with the *MQTTJSONPublisherFB*. The application creates an *openDAQ ref-device* with four channels, an *OpenDAQMQTTDevice*, and a *MQTTJSONPublisherFB* to publish JSON MQTT messages with the channels’ data. The properties of the *MQTTJSONPublisherFB* are set according to the selected mode, which can be specified via the *--mode* option, and array size,  which can be specified via the *--array* option with size. Posible *--mode* option values are:
     - 0 - One MQTT topic per signal;
     - 1 - One MQTT message/topic for all signals.   
 ```bash

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

@@ -122,13 +122,11 @@ int main(int argc, char* argv[])
         return appConfig.error;
     }
 
-    // Create OpenDAQ instance and add MQTT broker FB
+    // Create OpenDAQ instance and add the MQTT broker device
     const InstancePtr instance = InstanceBuilder().addModulePath(MODULE_PATH).build();
-    const std::string clientFbName = "MQTTClientFB";
-    auto clientFbConfig = instance.getAvailableFunctionBlockTypes().get(clientFbName).createDefaultConfig();
-    clientFbConfig.setPropertyValue("BrokerAddress", appConfig.brokerAddress);
-    auto brokerFB = instance.addFunctionBlock(clientFbName, clientFbConfig);
-    auto availableFbs = brokerFB.getAvailableFunctionBlockTypes();
+    const std::string connectionString = "daq.mqtt://" + appConfig.brokerAddress;
+    auto brokerDevice = instance.addDevice(connectionString);
+    auto availableFbs = brokerDevice.getAvailableFunctionBlockTypes();
 
     const std::string subFbName = "MQTTSubscriberFB";
     std::cout << "Try to add the " << subFbName << std::endl;
@@ -137,7 +135,7 @@ int main(int argc, char* argv[])
     config.setPropertyValue("JSONConfigFile", appConfig.configFilePath);
 
     // Add the subscriber function block to the broker FB
-    daq::FunctionBlockPtr subFb = brokerFB.addFunctionBlock(subFbName, config);
+    daq::FunctionBlockPtr subFb = brokerDevice.addFunctionBlock(subFbName, config);
 
     // Create packet readers for all signals
     auto signals = List<daq::ISignal>();

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

@@ -54,13 +54,11 @@ int main(int argc, char* argv[])
         return appConfig.error;
     }
 
-    // Create OpenDAQ instance and add MQTT broker FB
+    // Create OpenDAQ instance and add the MQTT broker device
     const InstancePtr instance = InstanceBuilder().addModulePath(MODULE_PATH).build();
-    const std::string clientFbName = "MQTTClientFB";
-    auto clientFbConfig = instance.getAvailableFunctionBlockTypes().get(clientFbName).createDefaultConfig();
-    clientFbConfig.setPropertyValue("BrokerAddress", appConfig.brokerAddress);
-    auto brokerFB = instance.addFunctionBlock(clientFbName, clientFbConfig);
-    auto availableFbs = brokerFB.getAvailableFunctionBlockTypes();
+    const std::string connectionString = "daq.mqtt://" + appConfig.brokerAddress;
+    auto brokerDevice = instance.addDevice(connectionString);
+    auto availableFbs = brokerDevice.getAvailableFunctionBlockTypes();
 
     const std::string fbName = "MQTTSubscriberFB";
     std::cout << "Try to add the " << fbName << std::endl;
@@ -72,7 +70,7 @@ int main(int argc, char* argv[])
     config.setPropertyValue("MessageIsString", True);
 
     // Add the subscriber function block to the broker FB
-    daq::FunctionBlockPtr subFb = brokerFB.addFunctionBlock(fbName, config);
+    daq::FunctionBlockPtr subFb = brokerDevice.addFunctionBlock(fbName, config);
 
     // Create packet readers for a signal
     const auto signal = subFb.getSignals()[0];

examples/ref-dev-mqtt-pub/src/ref-dev-mqtt-pub.cpp

@@ -100,12 +100,10 @@ int main(int argc, char* argv[])
     channels[3].setPropertyValue("SampleRate", 100);
     channels[3].setPropertyValue("Frequency", 20);
 
-    // Create and configure MQTT server
-    const std::string clientFbName = "MQTTClientFB";
-    auto clientFbConfig = instance.getAvailableFunctionBlockTypes().get(clientFbName).createDefaultConfig();
-    clientFbConfig.setPropertyValue("BrokerAddress", appConfig.brokerAddress);
-    auto brokerFB = instance.addFunctionBlock(clientFbName, clientFbConfig);
-    auto availableFbs = brokerFB.getAvailableFunctionBlockTypes();
+    // Connect to the MQTT broker
+    const std::string connectionString = "daq.mqtt://" + appConfig.brokerAddress;
+    auto brokerDevice = instance.addDevice(connectionString);
+    auto availableFbs = brokerDevice.getAvailableFunctionBlockTypes();
     const std::string fbName = "MQTTJSONPublisherFB";
     std::cout << "Try to add the " << fbName << std::endl;
 
@@ -120,7 +118,7 @@ int main(int argc, char* argv[])
 
 
     // Add the publisher function block to the broker device
-    daq::FunctionBlockPtr fb = brokerFB.addFunctionBlock(fbName, config);
+    daq::FunctionBlockPtr fb = brokerDevice.addFunctionBlock(fbName, config);
     const auto signals = refDevice.getSignals(search::Recursive(search::Any()));
     for (const auto& s : signals)
     {

modules/mqtt_streaming_module/include/mqtt_streaming_module/constants.h

@@ -24,8 +24,7 @@ static constexpr const char* DEFAULT_VALUE_SIGNAL_LOCAL_ID = "MQTTValueSignal";
 static constexpr const char* DEFAULT_TS_SIGNAL_LOCAL_ID = "MQTTTimestampSignal";
 
 
-static constexpr const char* PROPERTY_NAME_CLIENT_BROKER_ADDRESS = "BrokerAddress";
-static constexpr const char* PROPERTY_NAME_CLIENT_BROKER_PORT = "BrokerPort";
+static constexpr const char* PROPERTY_NAME_CLIENT_BROKER_PORT = "Port";
 static constexpr const char* PROPERTY_NAME_CLIENT_USERNAME = "Username";
 static constexpr const char* PROPERTY_NAME_CLIENT_PASSWORD = "Password";
 static constexpr const char* PROPERTY_NAME_CLIENT_CONNECT_TIMEOUT = "ConnectionTimeout";
@@ -59,22 +58,23 @@ static constexpr const char* PUB_PREVIEW_SIGNAL_NAME = "PreviewSignal";
 
 static constexpr const char* SUB_FB_NAME = "MQTTSubscriberFB";
 static constexpr const char* PUB_FB_NAME = "MQTTJSONPublisherFB";
-static constexpr const char* CLIENT_FB_NAME = "MQTTClientFB";
 static constexpr const char* JSON_DECODER_FB_NAME = "MQTTJSONDecoderFB";
 
-static const char* MQTT_LOCAL_CLIENT_FB_ID_PREFIX = "MQTTClientFB";
+static constexpr const char* CLIENT_DEVICE_TYPE_ID = "OpenDAQMQTTDevice";
+static constexpr const char* CLIENT_DEVICE_TYPE_NAME = "MQTT client device";
+static constexpr const char* CLIENT_DEVICE_CONN_PREFIX = "daq.mqtt";
+
+static const char* MQTT_LOCAL_CLIENT_DEVICE_ID_PREFIX = "MQTTClientDevice";
 static const char* MQTT_LOCAL_PUB_FB_ID_PREFIX = "MQTTJSONPublisherFB";
 static const char* MQTT_LOCAL_SUB_FB_ID_PREFIX = "MQTTSubscriberFB";
 static const char* MQTT_LOCAL_JSON_DECODER_FB_ID_PREFIX = "MQTTJSONDecoderFB";
 
 
-static const char* MQTT_CLIENT_FB_CON_STATUS_TYPE = "DAQ_MQTT_ConnectionStatusType";
 static const char* MQTT_PUB_FB_SIG_STATUS_TYPE = "DAQ_MQTT_SignalStatusType";
 static const char* MQTT_PUB_FB_PUB_STATUS_TYPE = "DAQ_MQTT_PublishingStatusType";
 static const char* MQTT_PUB_FB_SET_STATUS_TYPE = "DAQ_MQTT_SettingStatusType";
 
 
-static const char* MQTT_CLIENT_FB_CON_STATUS_NAME = "ConnectionStatus";
 static const char* MQTT_PUB_FB_SIG_STATUS_NAME = "SignalStatus";
 static const char* MQTT_PUB_FB_PUB_STATUS_NAME = "PublishingStatus";
 static const char* MQTT_PUB_FB_SET_STATUS_NAME = "SettingStatus";

modules/mqtt_streaming_module/include/mqtt_streaming_module/mqtt_client_fb_impl.h → modules/mqtt_streaming_module/include/mqtt_streaming_module/mqtt_client_device_impl.h

@@ -19,30 +19,32 @@
 #include "mqtt_streaming_protocol/MqttSettings.h"
 #include <future>
 #include <mqtt_streaming_module/common.h>
-#include <opendaq/function_block_impl.h>
-#include <opendaq/streaming_ptr.h>
-#include <mqtt_streaming_module/status_adaptor.h>
+#include <opendaq/device_impl.h>
 
 
 BEGIN_NAMESPACE_OPENDAQ_MQTT_STREAMING_MODULE
 
-class MqttClientFbImpl : public FunctionBlock
+class MqttClientDeviceImpl : public Device
 {
 public:
-    explicit MqttClientFbImpl(const ContextPtr& ctx,
-                                       const ComponentPtr& parent,
-                                       const PropertyObjectPtr& config);
+    explicit MqttClientDeviceImpl(const ContextPtr& ctx,
+                                  const ComponentPtr& parent,
+                                  const StringPtr& localId,
+                                  const StringPtr& connectionString,
+                                  const std::string& brokerHost,
+                                  const PropertyObjectPtr& config);
 
-    static FunctionBlockTypePtr CreateType();
+    static DeviceTypePtr CreateType();
+    static PropertyObjectPtr CreateDefaultConfig();
 
 protected:
-    static std::atomic<int> localIndex;
-    static std::string generateLocalId();
-
     void removed() override;
 
+    DeviceInfoPtr onGetInfo() override;
+
     DictPtr<IString, IFunctionBlockType> onGetAvailableFunctionBlockTypes() override;
     FunctionBlockPtr onAddFunctionBlock(const StringPtr& typeId, const PropertyObjectPtr& config) override;
+    void onRemoveFunctionBlock(const FunctionBlockPtr& functionBlock) override;
 
     void initNestedFbTypes();
     void initMqttSubscriber();
@@ -51,10 +53,14 @@ class MqttClientFbImpl : public FunctionBlock
     void readProperties(const PropertyObjectPtr& config);
     bool waitForConnection(const int timeoutMs);
 
+    /// Pushes a ConnectionStatusType value into the device's connection status container.
+    /// Surfaces to clients under the "ConfigurationStatus" alias.
+    void setConnectionStatus(const std::string& value, const std::string& message = "");
+
     DictObjectPtr<IDict, IString, IFunctionBlockType> nestedFbTypes;
 
+    StringPtr connectionString;
     int connectTimeout;
-    StatusAdaptor connectionStatus;
 
     std::shared_ptr<mqtt::MqttAsyncClient> subscriber;
     Mqtt::Utils::Settings::MqttConnectionSettings connectionSettings;

modules/mqtt_streaming_module/include/mqtt_streaming_module/mqtt_streaming_module_impl.h

@@ -18,6 +18,7 @@
 #include <mqtt_streaming_module/common.h>
 #include <opendaq/module_impl.h>
 #include <opendaq/device_ptr.h>
+#include <mutex>
 
 BEGIN_NAMESPACE_OPENDAQ_MQTT_STREAMING_MODULE
 
@@ -27,14 +28,29 @@ class MqttStreamingModule final : public Module
 public:
     MqttStreamingModule(ContextPtr context);
 
-    DictPtr<IString, IFunctionBlockType> onGetAvailableFunctionBlockTypes() override;
-    FunctionBlockPtr onCreateFunctionBlock(const StringPtr& id,
-                                           const ComponentPtr& parent,
-                                           const StringPtr& localId,
-                                           const PropertyObjectPtr& config) override;
+    DictPtr<IString, IDeviceType> onGetAvailableDeviceTypes() override;
+    DevicePtr onCreateDevice(const StringPtr& connectionString,
+                             const ComponentPtr& parent,
+                             const PropertyObjectPtr& config) override;
+
+    /// Host and port taken from a `daq.mqtt://host[:port]` connection string.
+    struct BrokerAddress
+    {
+        std::string host;
+        uint16_t port{0};   // 0 when the connection string carries no port
+    };
+
+    /// Throws InvalidParameterException when the string is not a valid `daq.mqtt://` address.
+    DAQ_MQTT_STREAM_MODULE_API static BrokerAddress parseConnectionString(const StringPtr& connectionString);
+    DAQ_MQTT_STREAM_MODULE_API static StringPtr formatConnectionString(const MqttStreamingModule::BrokerAddress& conParam);
 
 private:
-    static FunctionBlockTypePtr createFbType();
+    static DeviceTypePtr createDeviceType();
+
+    static PropertyObjectPtr populateDefaultConfig(const PropertyObjectPtr& config);
+
+    std::mutex sync;
+    size_t deviceIndex{0};
 };
 
 END_NAMESPACE_OPENDAQ_MQTT_STREAMING_MODULE