mqtt: refactoring

Author: viacheslauK

mqtt_streaming_client_module/src/mqtt_json_receiver_fb_impl.cpp

@@ -33,7 +33,7 @@ MqttJsonReceiverFbImpl::~MqttJsonReceiverFbImpl()
 
 void MqttJsonReceiverFbImpl::readProperties()
 {
-    auto lock = std::lock_guard<std::mutex>(sync);
+    auto lock = std::scoped_lock<std::mutex>(sync);
     subscribedSignals.clear();
     signalIdList.clear();
     bool isPresent = false;
@@ -66,7 +66,7 @@ void MqttJsonReceiverFbImpl::readProperties()
 
 void MqttJsonReceiverFbImpl::createDataPacket(const std::string& topic, const std::string& json)
 {
-    auto lock = std::lock_guard<std::mutex>(sync);
+    auto lock = std::scoped_lock<std::mutex>(sync);
     jsonDataWorker.createAndSendDataPacket(topic, json);
 }
 
@@ -79,7 +79,7 @@ void MqttJsonReceiverFbImpl::processMessage(const mqtt::MqttMessage& msg)
 
 void MqttJsonReceiverFbImpl::createSignals()
 {
-    auto lock = std::lock_guard<std::mutex>(sync);
+    auto lock = std::scoped_lock<std::mutex>(sync);
     if (!subscribedSignals.empty())
         LOG_I("Creating signals...");
 
@@ -131,7 +131,7 @@ void MqttJsonReceiverFbImpl::createSignals()
 
 std::vector<std::string> MqttJsonReceiverFbImpl::getSubscribedTopics() const
 {
-    auto lock = std::lock_guard<std::mutex>(sync);
+    auto lock = std::scoped_lock<std::mutex>(sync);
     std::set<std::string> topicsSet;
     for (const auto& [signalId, _] : subscribedSignals)
     {
@@ -142,7 +142,7 @@ std::vector<std::string> MqttJsonReceiverFbImpl::getSubscribedTopics() const
 
 void MqttJsonReceiverFbImpl::clearSubscribedTopics()
 {
-    auto lock = std::lock_guard<std::mutex>(sync);
+    auto lock = std::scoped_lock<std::mutex>(sync);
     subscribedSignals.clear();
     signalIdList.clear();
 }

mqtt_streaming_client_module/src/mqtt_raw_receiver_fb_impl.cpp

@@ -33,7 +33,7 @@ MqttRawReceiverFbImpl::~MqttRawReceiverFbImpl()
 
 void MqttRawReceiverFbImpl::readProperties()
 {
-    auto lock = std::lock_guard<std::mutex>(sync);
+    auto lock = std::scoped_lock<std::mutex>(sync);
     topicsForSubscribing.clear();
     bool isPresent = false;
     if (objPtr.hasProperty(PROPERTY_NAME_SIGNAL_LIST))
@@ -69,7 +69,7 @@ void MqttRawReceiverFbImpl::processMessage(const mqtt::MqttMessage& msg)
 {
     std::string topic(msg.getTopic());
 
-    auto lock = std::lock_guard<std::mutex>(sync);
+    auto lock = std::scoped_lock<std::mutex>(sync);
     auto signalIter = outputSignals.find(topic);
     if (signalIter == outputSignals.end())
     {
@@ -84,7 +84,7 @@ void MqttRawReceiverFbImpl::processMessage(const mqtt::MqttMessage& msg)
 
 void MqttRawReceiverFbImpl::createSignals()
 {
-    auto lock = std::lock_guard<std::mutex>(sync);
+    auto lock = std::scoped_lock<std::mutex>(sync);
     if (!topicsForSubscribing.empty())
         LOG_I("Creating signals...");
     for (const auto& topic : topicsForSubscribing)

mqtt_streaming_client_module/src/mqtt_streaming_device_impl.cpp

@@ -223,7 +223,7 @@ FunctionBlockPtr MqttStreamingDeviceImpl::onAddFunctionBlock(const StringPtr& ty
         }
         else
         {
-            setComponentStatusWithMessage(ComponentStatus::Error, "Function block type is not available: " + typeId.toStdString());
+            DAQ_THROW_EXCEPTION(NotFoundException, "Function block type is not available: " + typeId.toStdString());
         }
     }
     return nestedFunctionBlock;

mqtt_streaming_client_module/tests/test_daq_test_helper.h

@@ -7,7 +7,7 @@ class DaqTestHelper
 {
 public:
     daq::InstancePtr daqInstance;
-    daq::GenericDevicePtr<daq::IDevice> device;
+    daq::DevicePtr device;
 
     void StartUp(std::string connectionStr = "daq.mqtt://127.0.0.1", int discoveryTimeoutMs = 0)
     {

mqtt_streaming_protocol/include/MqttAsyncClient.h

@@ -116,7 +116,7 @@ class MqttAsyncClient final {
     std::function<MsgArrivedCb_type> onMsgArrivedCmnCb;
     std::unordered_map<std::string, std::function<MsgArrivedCb_type>> onMsgArrivedCbs;
 
-    std::lock_guard<std::recursive_mutex> getCbLock();
+    std::scoped_lock<std::recursive_mutex> getCbLock();
 
     static void onDeliveryCompleted(void *context, MQTTAsync_token token);
     static void onConnected(void *context, char *cause);

mqtt_streaming_protocol/src/MqttAsyncClient.cpp

@@ -147,9 +147,9 @@ MqttConnectionStatus MqttAsyncClient::isConnected() const
     return MQTTAsync_isConnected(client) ? MqttConnectionStatus::connected : MqttConnectionStatus::not_connected;
 }
 
-std::lock_guard<std::recursive_mutex> MqttAsyncClient::getCbLock()
+std::scoped_lock<std::recursive_mutex> MqttAsyncClient::getCbLock()
 {
-    return std::lock_guard<decltype(cbMtx)>(cbMtx);
+    return std::scoped_lock<decltype(cbMtx)>(cbMtx);
 }
 
 void MqttAsyncClient::setUsernamePasswrod(std::string username, std::string password)