domain value duplication error; tests

viacheslauK · viacheslauK · commit ccde7dc5f1b2 · 2026-03-13T16:29:46.000+01:00
diff --git a/modules/mqtt_streaming_module/include/mqtt_streaming_module/mqtt_json_decoder_fb_impl.h b/modules/mqtt_streaming_module/include/mqtt_streaming_module/mqtt_json_decoder_fb_impl.h
@@ -39,7 +39,7 @@ class MqttJsonDecoderFbImpl final : public FunctionBlock
                                 const PropertyObjectPtr& config = nullptr);
 
     DAQ_MQTT_STREAM_MODULE_API static FunctionBlockTypePtr CreateType();
-    void processMessage(const std::string& json, const uint64_t externalTs);
+    DAQ_MQTT_STREAM_MODULE_API void processMessage(const std::string& json, const uint64_t externalTs);
 protected:
 
     struct FbConfig {
@@ -58,7 +58,9 @@ class MqttJsonDecoderFbImpl final : public FunctionBlock
     std::atomic<bool> configValid;
     std::string configMsg;
     std::atomic<bool> parsingSucceeded;
+    std::atomic<bool> externalTsDuplicate;
     std::string parsingMsg;
+    uint64_t lastExternalTs;
 
     FbConfig config;
 
@@ -73,6 +75,7 @@ class MqttJsonDecoderFbImpl final : public FunctionBlock
     void propertyChanged();
 
     void updateStatuses();
+    void checkExternalTs(const uint64_t externalTs);
 };
 
 END_NAMESPACE_OPENDAQ_MQTT_STREAMING_MODULE
diff --git a/modules/mqtt_streaming_module/include/mqtt_streaming_module/mqtt_subscriber_fb_impl.h b/modules/mqtt_streaming_module/include/mqtt_streaming_module/mqtt_subscriber_fb_impl.h
@@ -77,7 +77,7 @@ class MqttSubscriberFbImpl final : public FunctionBlock
     void reconfigureSignal();
     void clearSubscribedTopic();
 
-    DataPacketPtr createDomainDataPacket(const uint64_t epochTime);
+    DAQ_MQTT_STREAM_MODULE_API DataPacketPtr createDomainDataPacket(const uint64_t epochTime);
 
     void processMessage(const mqtt::MqttMessage& msg);
     void initProperties(const PropertyObjectPtr& config);
diff --git a/modules/mqtt_streaming_module/src/mqtt_json_decoder_fb_impl.cpp b/modules/mqtt_streaming_module/src/mqtt_json_decoder_fb_impl.cpp
@@ -13,7 +13,8 @@ MqttJsonDecoderFbImpl::MqttJsonDecoderFbImpl(const ContextPtr& ctx,
                                        const FunctionBlockTypePtr& type,
                                        const PropertyObjectPtr& config)
     : FunctionBlock(type, ctx, parent, generateLocalId()),
-      jsonDataWorker()
+      jsonDataWorker(),
+      lastExternalTs(0)
 {
     initComponentStatus();
     if (config.assigned())
@@ -147,6 +148,7 @@ void MqttJsonDecoderFbImpl::readProperties()
     jsonDataWorker.setTimestampFieldName(config.tsFieldName);
     jsonDataWorker.setDomainSignalMode(config.tsMode);
     waitingData = configValid.load();
+    externalTsDuplicate = false;
     updateStatuses();
 }
 
@@ -168,13 +170,29 @@ void MqttJsonDecoderFbImpl::updateStatuses()
     {
         setComponentStatusWithMessage(ComponentStatus::Ok, "Waiting for data");
     }
-    else if (parsingSucceeded)
+    else if (parsingSucceeded == false)
     {
-        setComponentStatusWithMessage(ComponentStatus::Ok, "Parsing succeeded");
+        setComponentStatusWithMessage(ComponentStatus::Error, "Parsing failed: " + parsingMsg);
+    }
+    else if (externalTsDuplicate)
+    {
+        setComponentStatusWithMessage(ComponentStatus::Warning,
+                                      "Domain signal value for one of the received messages is the same as previous. "
+                                      "Data may be lost!");
     }
     else
     {
-        setComponentStatusWithMessage(ComponentStatus::Error, "Parsing failed: " + parsingMsg);
+        setComponentStatusWithMessage(ComponentStatus::Ok, "Parsing succeeded");
+    }
+}
+
+void MqttJsonDecoderFbImpl::checkExternalTs(const uint64_t externalTs)
+{
+    if (config.tsMode == mqtt::MqttDataWrapper::DomainSignalMode::ExternalTimestamp)
+    {
+        if (externalTs == lastExternalTs)
+            externalTsDuplicate = true;
+        lastExternalTs = externalTs;
     }
 }
 
@@ -184,6 +202,7 @@ void MqttJsonDecoderFbImpl::processMessage(const std::string& json, const uint64
     {
         auto lock = this->getRecursiveConfigLock();
         waitingData = false;
+        checkExternalTs(externalTs);
         auto status = jsonDataWorker.createAndSendDataPacket(json, externalTs);
         parsingSucceeded = status.success;
         if (status.success)
diff --git a/modules/mqtt_streaming_module/src/mqtt_subscriber_fb_impl.cpp b/modules/mqtt_streaming_module/src/mqtt_subscriber_fb_impl.cpp
@@ -431,7 +431,11 @@ DataPacketPtr MqttSubscriberFbImpl::createDomainDataPacket(const uint64_t epochT
         if (lastTsValue == epochTime)
         {
             if (statusContainer.getStatus("ComponentStatus") != ComponentStatus::Error)
-                setComponentStatusWithMessage(ComponentStatus::Warning, "Domain signal value is the same as previous. Data may be lost");
+            {
+                setComponentStatusWithMessage(ComponentStatus::Warning,
+                                              "Domain signal value for one of the received messages is the same as previous. "
+                                              "Data may be lost!");
+            }
         }
         lastTsValue = epochTime;
     }
diff --git a/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp b/modules/mqtt_streaming_module/tests/test_mqtt_json_decoder_fb.cpp
@@ -1127,3 +1127,54 @@ TEST_F(MqttJsonDecoderFbTest, RemovingNestedFunctionBlock)
     ASSERT_EQ(subMqttFb.getStatusContainer().getStatus("ComponentStatus"),
               Enumeration("ComponentStatusType", "Ok", daqInstance.getContext().getTypeManager()));
 }
+
+TEST_F(MqttJsonDecoderFbTest, PacketWithTheSameTS)
+{
+    using namespace std::chrono;
+    const std::string warnMsg("Domain signal value for one of the received messages is the same as previous.");
+    StartUp();
+    const std::string topic = buildTopicName();
+    const auto msgTemplate = VALID_JSON_DATA_1;
+    const std::string valueF = extractFieldName(msgTemplate, "<placeholder_value>");
+    AddSubFb(topic);
+    AddDecoderFb(valueF, DDSM::ExternalTimestamp, "");
+
+    auto fb = reinterpret_cast<MqttJsonDecoderFbImpl*>(*decoderObj);
+    auto buildMsg = [](int64_t value)
+    {
+        return replacePlaceholder(VALID_JSON_DATA_1, "<placeholder_value>", value);
+    };
+    auto getTime = []() { return duration_cast<microseconds>(system_clock::now().time_since_epoch()).count(); };
+    auto getComponentStatus = [&]() { return decoderObj.getStatusContainer().getStatus("ComponentStatus"); };
+    auto getStatusMsg = [&]() { return decoderObj.getStatusContainer().getStatusMessage("ComponentStatus").toStdString(); };
+
+    const auto warning = Enumeration("ComponentStatusType", "Warning", daqInstance.getContext().getTypeManager());
+    const auto ok = Enumeration("ComponentStatusType", "Ok", daqInstance.getContext().getTypeManager());
+
+    const auto ts = getTime();
+
+    fb->processMessage(buildMsg(1), ts);
+
+    ASSERT_EQ(getComponentStatus(), ok);
+
+    fb->processMessage(buildMsg(2), ts);
+    EXPECT_EQ(getComponentStatus(), warning);
+    EXPECT_NE(getStatusMsg().find(warnMsg), std::string::npos);
+
+    fb->processMessage(buildMsg(3), getTime());
+    EXPECT_EQ(getComponentStatus(), warning);
+    EXPECT_NE(getStatusMsg().find(warnMsg), std::string::npos);
+
+    // reconfiguring should reset warning
+    decoderObj.setPropertyValue(PROPERTY_NAME_DEC_UNIT, "ppm");
+    EXPECT_EQ(getComponentStatus(), ok);
+    EXPECT_EQ(getStatusMsg().find(warnMsg), std::string::npos);
+
+    fb->processMessage(buildMsg(3), getTime());
+    EXPECT_EQ(getComponentStatus(), ok);
+    EXPECT_EQ(getStatusMsg().find(warnMsg), std::string::npos);
+
+    fb->processMessage(buildMsg(3), getTime());
+    EXPECT_EQ(getComponentStatus(), ok);
+    EXPECT_EQ(getStatusMsg().find(warnMsg), std::string::npos);
+}
diff --git a/modules/mqtt_streaming_module/tests/test_mqtt_subscriber_fb.cpp b/modules/mqtt_streaming_module/tests/test_mqtt_subscriber_fb.cpp
@@ -61,6 +61,12 @@ class MqttSubscriberFbHelper
         mqtt::MqttAsyncClient unused;
         obj->onSignalsMessage(unused, msg);
     }
+
+    DataPacketPtr createDomainDataPacket(daq::FunctionBlockPtr subFb, const uint64_t epochTime)
+    {
+        auto fb = reinterpret_cast<MqttSubscriberFbImpl*>(*subFb);
+        return fb->createDomainDataPacket(epochTime);
+    }
 };
 
 class MqttSubscriberFbTest : public testing::Test, public DaqTestHelper, public MqttSubscriberFbHelper
@@ -658,3 +664,55 @@ TEST_F(MqttSubscriberFbTest, CheckRawFbFullDataTransferWithReconfiguring)
     ASSERT_EQ(dataToSend[1], dataToReceive[0]);
 }
 
+TEST_F(MqttSubscriberFbTest, DomainDataPacketWithTheSameTS)
+{
+    using namespace std::chrono;
+    const std::string warnMsg("Domain signal value for one of the received messages is the same as previous.");
+    const auto topic = buildTopicName();
+
+    StartUp();
+
+    auto config = clientMqttFb.getAvailableFunctionBlockTypes().get(SUB_FB_NAME).createDefaultConfig();
+    config.setPropertyValue(PROPERTY_NAME_SUB_TOPIC, topic);
+    config.setPropertyValue(PROPERTY_NAME_SUB_PREVIEW_SIGNAL, True);
+    config.setPropertyValue(PROPERTY_NAME_SUB_PREVIEW_SIGNAL_TS_MODE, static_cast<int>(SDSM::SystemTime));
+    auto fb = clientMqttFb.addFunctionBlock(SUB_FB_NAME, config);
+
+    auto getTime = []() { return duration_cast<microseconds>(system_clock::now().time_since_epoch()).count(); };
+    auto getComponentStatus = [&]() { return fb.getStatusContainer().getStatus("ComponentStatus"); };
+    auto getStatusMsg = [&]() { return fb.getStatusContainer().getStatusMessage("ComponentStatus").toStdString(); };
+
+    const auto warning = Enumeration("ComponentStatusType", "Warning", daqInstance.getContext().getTypeManager());
+    const auto ok = Enumeration("ComponentStatusType", "Ok", daqInstance.getContext().getTypeManager());
+
+    const auto ts = getTime();
+    auto packet = createDomainDataPacket(fb, ts);
+
+    ASSERT_EQ(getComponentStatus(), ok);
+
+    packet = createDomainDataPacket(fb, ts);
+
+    EXPECT_EQ(getComponentStatus(), warning);
+    EXPECT_NE(getStatusMsg().find(warnMsg), std::string::npos);
+
+    packet = createDomainDataPacket(fb, getTime());
+
+    EXPECT_EQ(getComponentStatus(), warning);
+    EXPECT_NE(getStatusMsg().find(warnMsg), std::string::npos);
+
+    // reconfiguring should reset warning
+    fb.setPropertyValue(PROPERTY_NAME_SUB_QOS, 2);
+
+    EXPECT_EQ(getComponentStatus(), ok);
+    EXPECT_EQ(getStatusMsg().find(warnMsg), std::string::npos);
+
+    packet = createDomainDataPacket(fb, getTime());
+
+    EXPECT_EQ(getComponentStatus(), ok);
+    EXPECT_EQ(getStatusMsg().find(warnMsg), std::string::npos);
+
+    packet = createDomainDataPacket(fb, getTime());
+
+    EXPECT_EQ(getComponentStatus(), ok);
+    EXPECT_EQ(getStatusMsg().find(warnMsg), std::string::npos);
+}

Original file line number	Diff line number	Diff line change
`@@ -13,7 +13,8 @@ MqttJsonDecoderFbImpl::MqttJsonDecoderFbImpl(const ContextPtr& ctx,`
`13`	`13`	`const FunctionBlockTypePtr& type,`
`14`	`14`	`const PropertyObjectPtr& config)`
`15`	`15`	`: FunctionBlock(type, ctx, parent, generateLocalId()),`
`16`		`- jsonDataWorker()`
	`16`	`+ jsonDataWorker(),`
	`17`	`+ lastExternalTs(0)`
`17`	`18`	`{`
`18`	`19`	`initComponentStatus();`
`19`	`20`	`if (config.assigned())`
`@@ -147,6 +148,7 @@ void MqttJsonDecoderFbImpl::readProperties()`
`147`	`148`	`jsonDataWorker.setTimestampFieldName(config.tsFieldName);`
`148`	`149`	`jsonDataWorker.setDomainSignalMode(config.tsMode);`
`149`	`150`	`waitingData = configValid.load();`
	`151`	`+ externalTsDuplicate = false;`
`150`	`152`	`updateStatuses();`
`151`	`153`	`}`
`152`	`154`
`@@ -168,13 +170,29 @@ void MqttJsonDecoderFbImpl::updateStatuses()`
`168`	`170`	`{`
`169`	`171`	`setComponentStatusWithMessage(ComponentStatus::Ok, "Waiting for data");`
`170`	`172`	`}`
`171`		`- else if (parsingSucceeded)`
	`173`	`+ else if (parsingSucceeded == false)`
`172`	`174`	`{`
`173`		`- setComponentStatusWithMessage(ComponentStatus::Ok, "Parsing succeeded");`
	`175`	`+ setComponentStatusWithMessage(ComponentStatus::Error, "Parsing failed: " + parsingMsg);`
	`176`	`+ }`
	`177`	`+ else if (externalTsDuplicate)`
	`178`	`+ {`
	`179`	`+ setComponentStatusWithMessage(ComponentStatus::Warning,`
	`180`	`+ "Domain signal value for one of the received messages is the same as previous. "`
	`181`	`+ "Data may be lost!");`
`174`	`182`	`}`
`175`	`183`	`else`
`176`	`184`	`{`
`177`		`- setComponentStatusWithMessage(ComponentStatus::Error, "Parsing failed: " + parsingMsg);`
	`185`	`+ setComponentStatusWithMessage(ComponentStatus::Ok, "Parsing succeeded");`
	`186`	`+ }`
	`187`	`+}`
	`188`	`+`
	`189`	`+void MqttJsonDecoderFbImpl::checkExternalTs(const uint64_t externalTs)`
	`190`	`+{`
	`191`	`+ if (config.tsMode == mqtt::MqttDataWrapper::DomainSignalMode::ExternalTimestamp)`
	`192`	`+ {`
	`193`	`+ if (externalTs == lastExternalTs)`
	`194`	`+ externalTsDuplicate = true;`
	`195`	`+ lastExternalTs = externalTs;`
`178`	`196`	`}`
`179`	`197`	`}`
`180`	`198`
`@@ -184,6 +202,7 @@ void MqttJsonDecoderFbImpl::processMessage(const std::string& json, const uint64`
`184`	`202`	`{`
`185`	`203`	`auto lock = this->getRecursiveConfigLock();`
`186`	`204`	`waitingData = false;`
	`205`	`+ checkExternalTs(externalTs);`
`187`	`206`	`auto status = jsonDataWorker.createAndSendDataPacket(json, externalTs);`
`188`	`207`	`parsingSucceeded = status.success;`
`189`	`208`	`if (status.success)`
Original file line number	Diff line number	Diff line change
`@@ -431,7 +431,11 @@ DataPacketPtr MqttSubscriberFbImpl::createDomainDataPacket(const uint64_t epochT`
`431`	`431`	`if (lastTsValue == epochTime)`
`432`	`432`	`{`
`433`	`433`	`if (statusContainer.getStatus("ComponentStatus") != ComponentStatus::Error)`
`434`		`- setComponentStatusWithMessage(ComponentStatus::Warning, "Domain signal value is the same as previous. Data may be lost");`
	`434`	`+ {`
	`435`	`+ setComponentStatusWithMessage(ComponentStatus::Warning,`
	`436`	`+ "Domain signal value for one of the received messages is the same as previous. "`
	`437`	`+ "Data may be lost!");`
	`438`	`+ }`
`435`	`439`	`}`
`436`	`440`	`lastTsValue = epochTime;`
`437`	`441`	`}`