New approach for MQTT module

CMakeLists.txt

@@ -4,6 +4,7 @@
 set(CMAKE_POLICY_VERSION_MINIMUM 3.5)
 cmake_minimum_required(VERSION 3.25)
 
+set(SDK_TARGET_NAMESPACE daq)
 set(REPO_NAME mqtt_module)
 set(REPO_OPTION_PREFIX MQTT_MODULE)
 
@@ -30,6 +31,7 @@ set(CMAKE_MESSAGE_CONTEXT_SHOW ON CACHE BOOL "Show CMake message context")
 
 list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/cmake")
 
+option(OPENDAQ_MQTT_MODULE_ENABLE_SSL "Enable building with openSSL" OFF)
 option(OPENDAQ_DEVICE_EXAMPLE_ENABLE_EXAMPLE_APPS "Enable building example applications" OFF)
 
 if ((CMAKE_COMPILER_IS_GNUCXX OR CMAKE_COMPILER_IS_CLANGXX) AND NOT MSVC)
@@ -42,17 +44,20 @@ include(CommonUtils)
 setup_repo(${REPO_OPTION_PREFIX})
 
 if(OPENDAQ_DEVICE_EXAMPLE_ENABLE_EXAMPLE_APPS)
-	set(DAQMODULES_REF_DEVICE_MODULE ON CACHE BOOL "" FORCE)
+    set(DAQMODULES_REF_DEVICE_MODULE ON CACHE BOOL "" FORCE)
 endif()
 
 option(OPENDAQ_MQTT_ENABLE_TESTS "Enable module testing" OFF)
 option(OPENDAQ_MQTT_ENABLE_EXAMPLE_APPS "Enable example applications building" OFF)
 
-find_package(OpenSSL REQUIRED)
-if (OPENSSL_FOUND)
-  message(STATUS "Found OpenSSL ${OPENSSL_VERSION}")
-else()
-  message(STATUS "OpenSSL Not Found")
+if(OPENDAQ_MQTT_MODULE_ENABLE_SSL)
+    find_package(OpenSSL REQUIRED)
+    if (OPENSSL_FOUND)
+        message(STATUS "Found OpenSSL ${OPENSSL_VERSION}")
+    else()
+        message(FATAL_ERROR "OpenSSL Not Found")
+    endif()
+    add_compile_definitions(OPENDAQ_MQTT_MODULE_ENABLE_SSL)
 endif()
 
 add_subdirectory(external)

examples/CMakeLists.txt

@@ -3,4 +3,6 @@ cmake_minimum_required(VERSION 3.16)
 if (OPENDAQ_MQTT_ENABLE_EXAMPLE_APPS)
     add_subdirectory(ref-dev-mqtt-pub)
     add_subdirectory(ref-dev-mqtt-sub)
+    add_subdirectory(ref-dev-mqtt-raw-sub)
+    add_subdirectory(custom-mqtt-sub)
 endif()

examples/InputArgs.h

@@ -78,8 +78,15 @@ class InputArgs
         return false;
     }
 
+    void setUsageHelp(const std::string& str)
+    {
+        usageString = str;
+    }
+
     void printHelp() const
     {
+        if (!usageString.empty())
+            std::cout << "Usage: " << usageString << std::endl;
         std::cout << "Available arguments:" << std::endl;
         for (const auto& [name, descStruct] : argDescriptions)
         {
@@ -100,4 +107,5 @@ class InputArgs
     std::vector<std::string> parsedArgs;
     std::unordered_map<std::string, std::string> argValues;
     std::vector<std::string> positionalArgs;
-};
+    std::string usageString;
+};

examples/custom-mqtt-sub/CMakeLists.txt

@@ -0,0 +1,12 @@
+cmake_minimum_required(VERSION 3.16)
+
+set(EXAMPLE_PROJECT_NAME "custom-mqtt-sub")
+
+project(${EXAMPLE_PROJECT_NAME} LANGUAGES CXX)
+
+set(CMAKE_CXX_STANDARD 17)
+set(CMAKE_CXX_STANDARD_REQUIRED ON)
+
+set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/bin)
+
+add_subdirectory(src)

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

@@ -0,0 +1,54 @@
+{
+    "openDAQ/RefDev0/IO/AI/RefCh0/Sig/AI0": [
+        {
+            "AI0": {
+                "Value": "value",
+                "Timestamp": "timestamp",
+                "Unit": [
+                    "V",
+                    "volts",
+                    "voltage"
+                ]
+            }
+        }
+    ],
+    "openDAQ/RefDev0/IO/AI/RefCh1/Sig/AI1": [
+        {
+            "AI1": {
+                "Value": "value",
+                "Timestamp": "timestamp",
+                "Unit": [
+                    "V",
+                    "volts",
+                    "voltage"
+                ]
+            }
+        }
+    ],
+    "openDAQ/RefDev0/IO/AI/RefCh2/Sig/AI2": [
+        {
+            "AI2": {
+                "Value": "value",
+                "Timestamp": "timestamp",
+                "Unit": [
+                    "V",
+                    "volts",
+                    "voltage"
+                ]
+            }
+        }
+    ],
+    "openDAQ/RefDev0/IO/AI/RefCh3/Sig/AI3": [
+        {
+            "AI3": {
+                "Value": "value",
+                "Timestamp": "timestamp",
+                "Unit": [
+                    "V",
+                    "volts",
+                    "voltage"
+                ]
+            }
+        }
+    ]
+}

examples/custom-mqtt-sub/src/CMakeLists.txt

@@ -0,0 +1,8 @@
+cmake_minimum_required(VERSION 3.16)
+
+add_compile_definitions(MODULE_PATH="${OPENDAQ_MODULES_DIR}")
+add_compile_definitions(APP_NAME="${EXAMPLE_PROJECT_NAME}")
+
+add_executable(${EXAMPLE_PROJECT_NAME} custom-mqtt-sub.cpp)
+target_link_libraries(${EXAMPLE_PROJECT_NAME} PRIVATE daq::opendaq)
+target_include_directories(${EXAMPLE_PROJECT_NAME} PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/../../)

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

@@ -0,0 +1,183 @@
+#include "../../InputArgs.h"
+#include <iomanip>
+#include <opendaq/opendaq.h>
+
+#include <fstream>
+#include <iostream>
+#include <sstream>
+
+using namespace daq;
+
+struct ConfigStruct {
+    std::string brokerAddress;
+    std::string configFilePath;
+    bool exit = true;
+    int error = 0;
+};
+
+std::string to_string(uint64_t ts)
+{
+    using namespace std::chrono;
+
+    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);
+
+    auto us = duration_cast<milliseconds>(tp.time_since_epoch()) % 1000;
+
+    std::ostringstream oss;
+    oss << std::put_time(&tm, "%Y-%m-%d %H:%M:%S") << '.' << std::setfill('0') << std::setw(3) << us.count();
+    return oss.str();
+}
+
+std::string to_string(daq::DataPacketPtr packet)
+{
+    std::string result;
+    std::string data;
+    switch (packet.getDataDescriptor().getSampleType())
+    {
+    case SampleType::Float64:
+        data = std::to_string(*(static_cast<double*>(packet.getData())));
+        break;
+    case SampleType::UInt64:
+        data = std::to_string(*(static_cast<uint64_t*>(packet.getData())));
+        break;
+    case SampleType::Int64:
+        data = std::to_string(*(static_cast<int64_t*>(packet.getData())));
+        break;
+    case SampleType::Binary:
+        data = '\"' + std::string(static_cast<char*>(packet.getData()), packet.getDataSize()) + '\"';
+        break;
+    default:
+        break;
+    }
+    std::string unitStr;
+    if (auto unit = packet.getDataDescriptor().getUnit(); unit.assigned())
+    {
+        if (auto s = unit.getSymbol(); s.assigned())
+            unitStr = " " + s.toStdString();
+    }
+
+    result = fmt::format("SampleType : {}; Data: {}{};", convertSampleTypeToString(packet.getDataDescriptor().getSampleType()), data, unitStr);
+    if (auto domainPacket = packet.getDomainPacket(); domainPacket.assigned())
+    {
+        uint64_t ts = *(static_cast<uint64_t*>(domainPacket.getData()));
+        result += fmt::format(" Time : {};", to_string(ts));
+    }
+    return result;
+}
+
+std::string readFileToString(const std::string& filePath)
+{
+    std::ifstream file(filePath);
+    if (!file)
+        throw std::runtime_error("Failed to open file: " + filePath);
+
+    std::ostringstream buffer;
+    buffer << file.rdbuf(); // Read the entire file buffer
+    return buffer.str();
+}
+
+ConfigStruct StartUp(int argc, char* argv[])
+{
+    ConfigStruct config;
+    InputArgs args;
+    args.addArg("--help", "Show help message");
+    args.addArg("--address", "MQTT broker address", true);
+    args.setUsageHelp(APP_NAME " [options] <config file>");
+    args.parse(argc, argv);
+
+    if (args.hasArg("--help") || args.hasUnknownArgs())
+    {
+        args.printHelp();
+        config.error = 0;
+        return config;
+    }
+
+    config.brokerAddress = args.getArgValue("--address", "127.0.0.1");
+    auto configFilePath = args.getPositionalArgs();
+    if (configFilePath.size() != 1)
+    {
+        std::cout << "Configuration file path is required." << std::endl;
+        config.error = -1;
+        return config;
+    }
+    if (configFilePath.size() > 1)
+    {
+        std::cout << "Only one configuration file path is allowed. The first one will be used - " << configFilePath[0] << std::endl;
+    }
+    config.configFilePath = std::move(configFilePath[0]);
+    config.exit = false;
+    return config;
+}
+
+int main(int argc, char* argv[])
+{
+    // Parse input arguments
+    auto appConfig = StartUp(argc, argv);
+    if (appConfig.exit)
+    {
+        return appConfig.error;
+    }
+
+    // Create OpenDAQ instance and add MQTT broker device
+    const InstancePtr instance = InstanceBuilder().addModulePath(MODULE_PATH).build();
+    auto brokerDevice = instance.addDevice("daq.mqtt://" + appConfig.brokerAddress);
+    auto availableDeviceNodes = brokerDevice.getAvailableFunctionBlockTypes();
+
+    const std::string fbName = "@jsonMqttFb";
+    std::cout << "Try to add the " << fbName << 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 = availableDeviceNodes.get(fbName).createDefaultConfig();
+    config.setPropertyValue("SignalList", jsonConfig);
+
+    // Add the JSON function block to the broker device
+    daq::FunctionBlockPtr jsonFb = brokerDevice.addFunctionBlock(fbName, config);
+
+    // Create packet readers for all signals
+    const auto signals = jsonFb.getSignals();
+    std::map<std::string, PacketReaderPtr> packetReaders;
+    for (const auto& s : signals)
+    {
+        packetReaders.emplace(std::pair<std::string, PacketReaderPtr>(s.getName().toStdString(), daq::PacketReader(s)));
+    }
+
+    // Start a thread to read packets from the readers
+    std::atomic<bool> running = true;
+    std::thread readerThread(
+        [&packetReaders, &running]()
+        {
+            while (running)
+            {
+                for (const auto& [signal, reader] : packetReaders)
+                {
+                    while (!reader.getEmpty() && running)
+                    {
+                        auto packet = reader.read();
+                        if (packet.getType() == PacketType::Event)
+                        {
+                            std::cout << "Event packet is skipped!" << std::endl;
+                        }
+                        else if (packet.getType() == PacketType::Data)
+                        {
+                            const auto dataPacket = packet.asPtr<IDataPacket>();
+                            std::cout << signal << " - " << to_string(dataPacket) << std::endl;
+                        }
+                    }
+                }
+                std::this_thread::sleep_for(std::chrono::milliseconds(20));
+            }
+        });
+
+    std::cout << "Press \"enter\" to exit the application..." << std::endl;
+    std::cin.get();
+
+    running = false;
+    readerThread.join();
+    std::cout << "Reader thread finished. Exiting.\n";
+
+    return 0;
+}