Add call unified reusable workflow

.github/workflows/ci-downstream.yml

@@ -0,0 +1,41 @@
+name: CI Downstream - MQTT Streaming
+
+on:
+  workflow_dispatch:
+    inputs:
+      opendaq-ref:
+        description: "openDAQ SDK commit, branch or tag"
+        required: false
+        default: ""
+  pull_request:
+
+jobs:
+  call-opendaq-reusable:
+    name: MQTT Streaming
+    uses: openDAQ/openDAQ-CI/.github/workflows/reusable.yml@v1
+    with:
+      opendaq-ref: ${{ github.event.inputs.opendaq-ref || '' }}
+      packages: >
+        [
+          {
+            "match-jobs": ["ubuntu-*"],
+            "apt-install": ["mosquitto"],
+            "run": "mosquitto -d"
+          },
+          {
+            "match-jobs": ["windows-*"],
+            "winget-install": ["EclipseFoundation.Mosquitto"]
+          },
+          {
+            "match-jobs": ["macos-*"],
+            "brew-install": ["mosquitto"],
+            "run": "$(brew --prefix mosquitto)/sbin/mosquitto -d"
+          }
+        ]
+      cmake-presets: >
+        [
+          {
+            "configure-preset": "module",
+            "test-preset": "module-test"
+          }
+        ]

CMakeLists.txt

@@ -37,6 +37,10 @@ option(${REPO_OPTION_PREFIX}_ENABLE_SSL "Enable building with openSSL" OFF)
 opendaq_common_compile_targets_settings()
 opendaq_setup_compiler_flags(${REPO_OPTION_PREFIX})
 
+if (CMAKE_CXX_COMPILER_ID MATCHES "Clang|AppleClang" AND CMAKE_CXX_COMPILER_VERSION VERSION_LESS 10)
+    add_compile_options(-Wno-unknown-warning-option)
+endif()
+
 if (${REPO_OPTION_PREFIX}_ENABLE_TESTS)
     message(STATUS "Unit tests in ${REPO_NAME} are ENABLED")
     enable_testing()

CMakePresets.json

@@ -178,6 +178,20 @@
                 "clang",
                 "ninja"
             ]
+        },
+        {
+            "name": "module",
+            "hidden": true,
+            "cacheVariables": {
+                "DAQMODULES_MQTT_ENABLE_TESTS": "ON"
+            }
+        }
+    ],
+    "testPresets": [
+        {
+            "name": "module-test",
+            "hidden": true,
+            "configurePreset": "module"
         }
     ]
 }

modules/mqtt_streaming_module/include/mqtt_streaming_module/mqtt_subscriber_fb_impl.h

@@ -86,6 +86,7 @@ class MqttSubscriberFbImpl final : public FunctionBlock
     std::mutex queueMutex;
     std::condition_variable queueCv;
     mutable std::recursive_mutex processingMutex;
+    std::mutex componentStatusMutex;
 
     DAQ_MQTT_STREAM_MODULE_API void onSignalsMessage(const mqtt::MqttAsyncClient& subscriber, const mqtt::MqttMessage& msg);
 

modules/mqtt_streaming_module/src/mqtt_subscriber_fb_impl.cpp

@@ -101,6 +101,8 @@ void MqttSubscriberFbImpl::updateStatuses()
     if (!statuses->isUpdated())
         return;
 
+    std::scoped_lock lock(componentStatusMutex);
+
     if (!jsonConfigErr.ok())
     {
         setComponentStatusWithMessage(ComponentStatus::Error, jsonConfigErr.buildStatusMessage());
@@ -508,7 +510,6 @@ void MqttSubscriberFbImpl::processMessageImpl(const mqtt::MqttMessage& msg, cons
             decoderFb->processMessage(jsonObjStr, epochTime);
         }
     }
-    updateStatuses();
 }
 
 void MqttSubscriberFbImpl::processingLoop()

modules/mqtt_streaming_module/tests/test_mqtt_publisher_fb.cpp

@@ -451,7 +451,7 @@ class MqttPublisherFbHelper : public DaqTestHelper
     }
 
     template <typename vT>
-    static std::string valueToString(const vT& value, bool quoteString = true)
+    static std::string valueToString(const vT& value, [[maybe_unused]] bool quoteString = true)
     {
         std::string result;
         if constexpr (std::is_same_v<vT, std::string>)
@@ -1709,7 +1709,7 @@ TEST_F(MqttPublisherFbTest, DISABLED_MultiReaderTest)
     auto send = [&]()
     {
         std::cout << "---------------sending-data---------------" << std::endl;
-        for (size_t i = 0; i < samples * divider; i++)
+        for (size_t i = 0; i < static_cast<size_t>(samples * divider); i++)
         {
             uint64_t ts0, ts1;
             if (i % divider == 0)