mqtt: publishingStatus for publiher FB

viacheslauK · viacheslauK · commit 0b1ae5e2ef0b · 2025-12-23T10:02:31.000+01:00
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -62,6 +62,7 @@ if(OPENDAQ_MQTT_MODULE_ENABLE_SSL)
 endif()
 
 add_subdirectory(external)
+add_subdirectory(helper_utils)
 add_subdirectory(mqtt_streaming_protocol)
 add_subdirectory(mqtt_streaming_module)
 
diff --git a/helper_utils/CMakeLists.txt b/helper_utils/CMakeLists.txt
@@ -0,0 +1,19 @@
+cmake_minimum_required(VERSION 3.10)
+
+set(HELPER_LIB mqtt_streaming_helper)
+
+set(HELPER_PRJ_VERSION "1.0.0")
+set(HELPER_PRJ_NAME "HelperUtils")
+
+project(${HELPER_PRJ_NAME} VERSION ${HELPER_PRJ_VERSION} LANGUAGES C CXX)
+
+
+
+add_library(${HELPER_LIB} INTERFACE)
+
+target_include_directories(${HELPER_LIB} INTERFACE ${CMAKE_CURRENT_SOURCE_DIR}/include
+)
+
+target_sources(${HELPER_LIB} INTERFACE
+    ${CMAKE_CURRENT_SOURCE_DIR}/include/mqtt_streaming_helper/timer.h
+)
diff --git a/helper_utils/include/mqtt_streaming_helper/timer.h b/helper_utils/include/mqtt_streaming_helper/timer.h
@@ -2,31 +2,49 @@
 
 #include <chrono>
 
+namespace helper::utils
+{
 class Timer
 {
 public:
-    Timer(int ms)
+    Timer(size_t ms, bool start = true)
+        : period(ms),
+          firstStart(true)
     {
-        start = std::chrono::steady_clock::now();
-        timeout = std::chrono::milliseconds(ms);
+        if (start)
+            restart();
     }
+
     std::chrono::milliseconds remain() const
     {
         auto now = std::chrono::steady_clock::now();
         const auto elapsed_ms = std::chrono::duration_cast<std::chrono::milliseconds>(now - start);
         std::chrono::milliseconds newTout = (elapsed_ms >= timeout) ? std::chrono::milliseconds(0) : timeout - elapsed_ms;
         return newTout;
     }
+
     bool expired()
     {
-        return remain() == std::chrono::milliseconds(0);
+        return (firstStart) ? true : (remain() == std::chrono::milliseconds(0));
     }
+
     explicit operator std::chrono::milliseconds() const noexcept
     {
         return remain();
     }
 
+    void restart()
+    {
+        firstStart = false;
+        start = std::chrono::steady_clock::now();
+        timeout = std::chrono::milliseconds(period);
+    }
+
 protected:
     std::chrono::steady_clock::time_point start;
     std::chrono::milliseconds timeout;
+    std::chrono::milliseconds period;
+    bool firstStart;
 };
+
+} // namespace helper::utils
diff --git a/mqtt_streaming_module/include/mqtt_streaming_module/constants.h b/mqtt_streaming_module/include/mqtt_streaming_module/constants.h
@@ -50,5 +50,9 @@ static const char* MQTT_LOCAL_JSON_FB_ID_PREFIX = "jsonMqttFb";
 static const char* MQTT_ROOT_FB_CON_STATUS_TYPE = "BrokerConnectionStatusType";
 static const char* MQTT_RAW_FB_SUB_STATUS_TYPE = "MqttSubscriptionStatusType";
 static const char* MQTT_PUB_FB_SIG_STATUS_TYPE = "MqttSignalStatusType";
+static const char* MQTT_PUB_FB_PUB_STATUS_TYPE = "MqttPublishingStatusType";
+
+static const char* MQTT_PUB_FB_SIG_STATUS_NAME = "SignalStatus";
+static const char* MQTT_PUB_FB_PUB_STATUS_NAME = "PublishingStatus";
 
 END_NAMESPACE_OPENDAQ_MQTT_STREAMING_MODULE
diff --git a/mqtt_streaming_module/include/mqtt_streaming_module/mqtt_publisher_fb_impl.h b/mqtt_streaming_module/include/mqtt_streaming_module/mqtt_publisher_fb_impl.h
@@ -18,6 +18,7 @@
 #include "MqttAsyncClient.h"
 #include "MqttDataWrapper.h"
 #include "mqtt_streaming_module/handler_base.h"
+#include "mqtt_streaming_helper/timer.h"
 #include <mqtt_streaming_module/common.h>
 #include <mqtt_streaming_module/types.h>
 #include <opendaq/function_block_impl.h>
@@ -34,6 +35,12 @@ class MqttPublisherFbImpl final : public FunctionBlock
         Valid
     };
 
+    enum class PublishingStatus : EnumType
+    {
+        Ok = 0,
+        SampleSkipped
+    };
+
     explicit MqttPublisherFbImpl(const ContextPtr& ctx,
                                  const ComponentPtr& parent,
                                  const FunctionBlockTypePtr& type,
@@ -47,10 +54,11 @@ class MqttPublisherFbImpl final : public FunctionBlock
     void onConnected(const InputPortPtr& port) override;
     void onDisconnected(const InputPortPtr& port) override;
 
-    static void addTypeToTypeManager(daq::TypeManagerPtr manager);
+    static void addTypesToTypeManager(daq::TypeManagerPtr manager);
 
 private:
     static const std::vector<std::pair<SignalStatus, std::string>> signalStatusMap;
+    static const std::vector<std::pair<PublishingStatus, std::string>> publishingStatusMap;
 
     static std::atomic<int> localIndex;
     std::shared_ptr<mqtt::MqttAsyncClient> mqttClient;
@@ -63,10 +71,16 @@ class MqttPublisherFbImpl final : public FunctionBlock
     std::atomic<bool> hasError;
     std::unique_ptr<HandlerBase> handler;
     EnumerationPtr signalStatus;
+    EnumerationPtr publishingStatus;
+    uint64_t skippedMsgCnt;
+    uint64_t publishedMsgCnt;
+    std::string lastSkippedReason;
+    helper::utils::Timer publishingStatusTimer;
 
     static std::string getLocalId();
-    void initSignalStatus();
-    void setSignalStatus(const SignalStatus status, std::string message = "");
+    void setSignalStatus(const SignalStatus status, std::string message = "", bool init = false);
+    void setPublishingStatus(const PublishingStatus status, std::string message = "", bool init = false);
+    void updatePublishingStatus();
     void initProperties(const PropertyObjectPtr& config);
     void readProperties();
     void propertyChanged();
diff --git a/mqtt_streaming_module/src/CMakeLists.txt b/mqtt_streaming_module/src/CMakeLists.txt
@@ -86,6 +86,7 @@ endif()
 
 target_link_libraries(${LIB_NAME} PUBLIC  daq::opendaq
                                           mqtt_streaming_protocol
+                                          mqtt_streaming_helper
                                   PRIVATE $<BUILD_INTERFACE:Boost::algorithm>
 )
 
diff --git a/mqtt_streaming_module/src/mqtt_publisher_fb_impl.cpp b/mqtt_streaming_module/src/mqtt_publisher_fb_impl.cpp
@@ -14,6 +14,10 @@ const std::vector<std::pair<MqttPublisherFbImpl::SignalStatus, std::string>> Mqt
      {SignalStatus::Invalid, "Invalid"},
      {SignalStatus::Valid, "Valid"}};
 
+const std::vector<std::pair<MqttPublisherFbImpl::PublishingStatus, std::string>> MqttPublisherFbImpl::publishingStatusMap =
+    {{PublishingStatus::Ok, "Ok"},
+     {PublishingStatus::SampleSkipped, "SampleSkipped"}};
+
 MqttPublisherFbImpl::MqttPublisherFbImpl(const ContextPtr& ctx,
                                          const ComponentPtr& parent,
                                          const FunctionBlockTypePtr& type,
@@ -24,10 +28,14 @@ MqttPublisherFbImpl::MqttPublisherFbImpl(const ContextPtr& ctx,
       jsonDataWorker(loggerComponent),
       inputPortCount(0),
       running(true),
-      hasError(false)
+      hasError(false),
+      skippedMsgCnt(0),
+      publishingStatusTimer(helper::utils::Timer(1000, false))
 {
     initComponentStatus();
-    initSignalStatus();
+    addTypesToTypeManager(context.getTypeManager());
+    setSignalStatus(SignalStatus::NotConnected, "", true);
+    setPublishingStatus(PublishingStatus::Ok, "", true);
     if (config.assigned())
         initProperties(populateDefaultConfig(type.createDefaultConfig(), config));
     else
@@ -171,8 +179,12 @@ void MqttPublisherFbImpl::updateInputPorts()
 
 void MqttPublisherFbImpl::validateInputPorts()
 {
+    skippedMsgCnt = 0;
+    publishedMsgCnt = 0;
+    setPublishingStatus(PublishingStatus::Ok);
     if (signalContexts.size() == 1)     // no one input port is connected
     {
+        setComponentStatus(ComponentStatus::Ok);
         setSignalStatus(SignalStatus::NotConnected);
     }
     else
@@ -280,6 +292,7 @@ void MqttPublisherFbImpl::readerLoop()
             msgs = handler->processSignalContexts(signalContexts);
         }
         sendMessages(msgs);
+        updatePublishingStatus();
         std::this_thread::sleep_for(std::chrono::milliseconds(config.periodMs));
     }
 }
@@ -291,7 +304,13 @@ void MqttPublisherFbImpl::sendMessages(const MqttData& data)
         auto status = mqttClient->publish(topic, (void*)msg.c_str(), msg.length(), config.qos);
         if (!status.success)
         {
-            LOG_W("Failed to publish data to {}; reason - {}", topic, status.msg);
+            ++skippedMsgCnt;
+            lastSkippedReason = std::move(status.msg);
+            LOG_W("Failed to publish data to {}; reason - {}", topic, lastSkippedReason);
+        }
+        else
+        {
+            ++publishedMsgCnt;
         }
     }
 }
@@ -301,27 +320,55 @@ std::string MqttPublisherFbImpl::getLocalId()
     return std::string(MQTT_LOCAL_PUB_FB_ID_PREFIX + std::to_string(localIndex++));
 }
 
-void MqttPublisherFbImpl::initSignalStatus()
+void MqttPublisherFbImpl::setSignalStatus(const SignalStatus status, std::string message, bool init)
 {
+    signalStatus = EnumerationWithIntValue(MQTT_PUB_FB_SIG_STATUS_TYPE, static_cast<Int>(status), context.getTypeManager());
+    if (init)
+        statusContainer.template asPtr<IComponentStatusContainerPrivate>(true).addStatusWithMessage(MQTT_PUB_FB_SIG_STATUS_NAME,
+                                                                                                    signalStatus,
+                                                                                                    message);
+    else
+        statusContainer.template asPtr<IComponentStatusContainerPrivate>(true).setStatusWithMessage(MQTT_PUB_FB_SIG_STATUS_NAME,
+                                                                                                    signalStatus,
+                                                                                                    message);
+}
 
-    addTypeToTypeManager(context.getTypeManager());
-
-    signalStatus = EnumerationWithIntValue(MQTT_PUB_FB_SIG_STATUS_TYPE,
-                                                 static_cast<Int>(SignalStatus::NotConnected),
-                                                 this->context.getTypeManager());
-    statusContainer.template asPtr<IComponentStatusContainerPrivate>(true).addStatus("SignalStatus",
-                                                                                     signalStatus);
+void MqttPublisherFbImpl::updatePublishingStatus()
+{
+    if (publishingStatusTimer.expired())
+    {
+        publishingStatusTimer.restart();
+        if (skippedMsgCnt != 0)
+        {
+            if (statusContainer.getStatus("ComponentStatus") == ComponentStatus::Ok)
+                setComponentStatusWithMessage(ComponentStatus::Warning, "Some messages were not published!");
+            setPublishingStatus(PublishingStatus::SampleSkipped,
+                                fmt::format("Published: {}; Skipped: {}; last reason - {}",
+                                            publishedMsgCnt,
+                                            skippedMsgCnt,
+                                            lastSkippedReason));
+        }
+        else
+        {
+            setPublishingStatus(PublishingStatus::Ok, fmt::format("Published: {};", publishedMsgCnt));
+        }
+    }
 }
 
-void MqttPublisherFbImpl::setSignalStatus(const SignalStatus status, std::string message)
+void MqttPublisherFbImpl::setPublishingStatus(const PublishingStatus status, std::string message, bool init)
 {
-    signalStatus = EnumerationWithIntValue(MQTT_PUB_FB_SIG_STATUS_TYPE, static_cast<Int>(status), this->context.getTypeManager());
-    statusContainer.template asPtr<IComponentStatusContainerPrivate>(true).setStatusWithMessage("SignalStatus",
-                                                                                                signalStatus,
-                                                                                                message);
+    publishingStatus = EnumerationWithIntValue(MQTT_PUB_FB_PUB_STATUS_TYPE, static_cast<Int>(status), context.getTypeManager());
+    if (init)
+        statusContainer.template asPtr<IComponentStatusContainerPrivate>(true).addStatusWithMessage(MQTT_PUB_FB_PUB_STATUS_NAME,
+                                                                                                    publishingStatus,
+                                                                                                    message);
+    else
+        statusContainer.template asPtr<IComponentStatusContainerPrivate>(true).setStatusWithMessage(MQTT_PUB_FB_PUB_STATUS_NAME,
+                                                                                                    publishingStatus,
+                                                                                                    message);
 }
 
-void MqttPublisherFbImpl::addTypeToTypeManager(daq::TypeManagerPtr manager)
+void MqttPublisherFbImpl::addTypesToTypeManager(daq::TypeManagerPtr manager)
 {
     if (!manager.hasType(MQTT_PUB_FB_SIG_STATUS_TYPE))
     {
@@ -331,5 +378,14 @@ void MqttPublisherFbImpl::addTypeToTypeManager(daq::TypeManagerPtr manager)
 
         manager.addType(EnumerationType(MQTT_PUB_FB_SIG_STATUS_TYPE, list));
     }
+
+    if (!manager.hasType(MQTT_PUB_FB_PUB_STATUS_TYPE))
+    {
+        auto list = List<IString>();
+        for (const auto& [_, st] : publishingStatusMap)
+            list.pushBack(st);
+
+        manager.addType(EnumerationType(MQTT_PUB_FB_PUB_STATUS_TYPE, list));
+    }
 }
 END_NAMESPACE_OPENDAQ_MQTT_STREAMING_MODULE
diff --git a/mqtt_streaming_module/tests/CMakeLists.txt b/mqtt_streaming_module/tests/CMakeLists.txt
@@ -18,6 +18,7 @@ target_link_libraries(${TEST_APP} PRIVATE daq::test_utils
                                           ${MODULE_NAME}
                                           mqtt_stream_module
                                           mqtt_streaming_protocol_test_helper
+                                          mqtt_streaming_helper
 )
 
 add_test(NAME ${TEST_APP}
diff --git a/mqtt_streaming_module/tests/test_mqtt_json_fb.cpp b/mqtt_streaming_module/tests/test_mqtt_json_fb.cpp
@@ -1,5 +1,5 @@
 #include "MqttAsyncClientWrapper.h"
-#include "Timer.h"
+#include "mqtt_streaming_helper/timer.h"
 #include "mqtt_streaming_module/mqtt_json_receiver_fb_impl.h"
 #include "test_daq_test_helper.h"
 #include "test_data.h"
@@ -236,7 +236,7 @@ class MqttJsonFbHelper
     {
         std::vector<T> result;
 
-        auto timer = Timer(timeoutMs);
+        auto timer = helper::utils::Timer(timeoutMs);
         while (!reader.getEmpty() || !timer.expired())
         {
             if (reader.getEmpty())
diff --git a/mqtt_streaming_module/tests/test_mqtt_publisher_fb.cpp b/mqtt_streaming_module/tests/test_mqtt_publisher_fb.cpp
diff --git a/mqtt_streaming_protocol/tests/CMakeLists.txt b/mqtt_streaming_protocol/tests/CMakeLists.txt
diff --git a/mqtt_streaming_protocol/tests/MqttAsyncClientWrapper.cpp b/mqtt_streaming_protocol/tests/MqttAsyncClientWrapper.cpp
diff --git a/mqtt_streaming_protocol/tests/test_mqtt_streaming_protocol.cpp b/mqtt_streaming_protocol/tests/test_mqtt_streaming_protocol.cpp

Original file line number	Diff line number	Diff line change
`@@ -86,6 +86,7 @@ endif()`
`86`	`86`
`87`	`87`	`target_link_libraries(${LIB_NAME} PUBLIC daq::opendaq`
`88`	`88`	`mqtt_streaming_protocol`
	`89`	`+ mqtt_streaming_helper`
`89`	`90`	`PRIVATE $<BUILD_INTERFACE:Boost::algorithm>`
`90`	`91`	`)`
`91`	`92`
Original file line number	Diff line number	Diff line change
`@@ -18,6 +18,7 @@ target_link_libraries(${TEST_APP} PRIVATE daq::test_utils`
`18`	`18`	`${MODULE_NAME}`
`19`	`19`	`mqtt_stream_module`
`20`	`20`	`mqtt_streaming_protocol_test_helper`
	`21`	`+ mqtt_streaming_helper`
`21`	`22`	`)`
`22`	`23`
`23`	`24`	`add_test(NAME ${TEST_APP}`
Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,5 @@`
`1`	`1`	`#include "MqttAsyncClientWrapper.h"`
`2`		`-#include "Timer.h"`
	`2`	`+#include "mqtt_streaming_helper/timer.h"`
`3`	`3`	`#include "mqtt_streaming_module/mqtt_json_receiver_fb_impl.h"`
`4`	`4`	`#include "test_daq_test_helper.h"`
`5`	`5`	`#include "test_data.h"`
`@@ -236,7 +236,7 @@ class MqttJsonFbHelper`
`236`	`236`	`{`
`237`	`237`	`std::vector<T> result;`
`238`	`238`
`239`		`- auto timer = Timer(timeoutMs);`
	`239`	`+ auto timer = helper::utils::Timer(timeoutMs);`
`240`	`240`	`while (!reader.getEmpty() \|\| !timer.expired())`
`241`	`241`	`{`
`242`	`242`	`if (reader.getEmpty())`