mqtt: disconnecting on device removing

Author: viacheslauK

mqtt_streaming_client_module/include/mqtt_streaming_client_module/mqtt_streaming_device_impl.h

@@ -47,7 +47,7 @@ class MqttStreamingDeviceImpl : public Device
     DictPtr<IString, IFunctionBlockType> onGetAvailableFunctionBlockTypes() override;
     FunctionBlockPtr onAddFunctionBlock(const StringPtr& typeId, const PropertyObjectPtr& config) override;
 
-    void setupMqttSubscriber();
+    void initMqttSubscriber();
     bool waitForConnection(const int timeoutMs);
     void receiveSignalTopics(const int timeoutMs);
     void onSignalsMessage(const mqtt::MqttAsyncClient& subscriber, mqtt::MqttMessage& msg);

mqtt_streaming_client_module/src/mqtt_streaming_device_impl.cpp

@@ -17,6 +17,8 @@ BEGIN_NAMESPACE_OPENDAQ_MQTT_STREAMING_CLIENT_MODULE
 
 std::atomic<int> MqttStreamingDeviceImpl::localIndex = 0;
 
+constexpr int MQTT_CLIENT_SYNC_DISCONNECT_TOUT = 3000;
+
 MqttStreamingDeviceImpl::MqttStreamingDeviceImpl(const ContextPtr& ctx, const ComponentPtr& parent, const PropertyObjectPtr& config)
     : Device(ctx, parent, getLocalId()),
       connectionStatus(Enumeration("ConnectionStatusType", "Connected", this->context.getTypeManager())),
@@ -37,7 +39,7 @@ MqttStreamingDeviceImpl::MqttStreamingDeviceImpl(const ContextPtr& ctx, const Co
 
     initComponentStatus();
 
-    setupMqttSubscriber();
+    initMqttSubscriber();
     if (!waitForConnection(initTimeout))
     {
         LOG_E("MQTT: could not connect to MQTT broker within {} ms", initTimeout);
@@ -50,6 +52,9 @@ MqttStreamingDeviceImpl::MqttStreamingDeviceImpl(const ContextPtr& ctx, const Co
 
 void MqttStreamingDeviceImpl::removed()
 {
+    bool disRes = subscriber->syncDisconnect(MQTT_CLIENT_SYNC_DISCONNECT_TOUT);
+    if (!disRes)
+        LOG_E("MQTT: disconnection was unsuccessful");
     Device::removed();
 }
 
@@ -58,9 +63,8 @@ DeviceInfoPtr MqttStreamingDeviceImpl::onGetInfo()
     return DeviceInfo(connectionString, MQTT_DEVICE_NAME);
 }
 
-void MqttStreamingDeviceImpl::setupMqttSubscriber()
+void MqttStreamingDeviceImpl::initMqttSubscriber()
 {
-    subscriber->disconnect();
     subscriber->setServerURL(connectionSettings.mqttUrl);
     subscriber->setClientId(connectionSettings.clientId);
     subscriber->setUsernamePasswrod(connectionSettings.username, connectionSettings.password);

mqtt_streaming_protocol/include/MqttAsyncClient.h

@@ -40,6 +40,7 @@ class MqttAsyncClient final {
 
     bool connect();
     bool disconnect();
+    bool syncDisconnect(int timeoutMs);
 
     bool publish(const std::string &topic,
                  void *data,
@@ -84,6 +85,7 @@ class MqttAsyncClient final {
     std::function<void()> onConnectedCb;
     std::function<void(int, bool)> onSentCb;
     std::function<void(bool)> onDisconnectCb;
+    std::function<void(bool)> onInternalDisconnectCb;
     std::function<void(int)> onDeliveryCompletedCb;
     std::function<MsgArrivedCb_type> onMsgArrivedCmnCb;
     std::unordered_map<std::string, std::function<MsgArrivedCb_type>> onMsgArrivedCbs;

mqtt_streaming_protocol/src/MqttAsyncClient.cpp

@@ -1,4 +1,5 @@
 #include "MqttAsyncClient.h"
+#include <future>
 
 namespace mqtt
 {
@@ -90,8 +91,47 @@ bool MqttAsyncClient::connect()
 
 bool MqttAsyncClient::disconnect()
 {
+    if (client == nullptr)
+        return true;
+
     // It is only the result of the request to disconnect (queuing)
-    return MQTTAsync_disconnect(client, &disconnOpts) == MQTTASYNC_SUCCESS;
+    auto status = MQTTAsync_disconnect(client, &disconnOpts);
+    bool result = (status == MQTTASYNC_SUCCESS || status == MQTTASYNC_DISCONNECTED);
+    return result;
+}
+
+bool MqttAsyncClient::syncDisconnect(int timeoutMs)
+{
+    if (client == nullptr)
+        return true;
+
+    bool result = disconnect();
+    if (result)
+    {
+        std::atomic<bool> done{false};
+        std::promise<bool> disconnectedPromise;
+        auto disconnectedFuture = disconnectedPromise.get_future();
+        {
+            auto lock = getCbLock();
+            onInternalDisconnectCb = [promise = &disconnectedPromise, &done](bool result) {
+                bool expected = false;
+                if (done.compare_exchange_strong(expected, true)) {
+                    promise->set_value(result);
+                }
+            };
+        }
+        auto status = disconnectedFuture.wait_for(std::chrono::milliseconds(timeoutMs));
+        {
+            auto lock = getCbLock();
+            onInternalDisconnectCb = nullptr;
+        }
+        if (status == std::future_status::ready && disconnectedFuture.get() == true)
+        {
+            MQTTAsync_destroy(&client);
+            return true;
+        }
+    }
+    return false;
 }
 
 MqttConnectionStatus MqttAsyncClient::isConnected() const
@@ -336,6 +376,8 @@ void MqttAsyncClient::onDisconnectSuccess(void* context, MQTTAsync_successData*
     {
         auto clienttInst = (MqttAsyncClient*)context;
         auto lock = clienttInst->getCbLock();
+        if (clienttInst->onInternalDisconnectCb)
+            clienttInst->onInternalDisconnectCb(true);
         if (clienttInst->onDisconnectCb)
             clienttInst->onDisconnectCb(true);
     }
@@ -348,6 +390,8 @@ void MqttAsyncClient::onDisconnectFailure(void* context, MQTTAsync_failureData*
     {
         auto clienttInst = (MqttAsyncClient*)context;
         auto lock = clienttInst->getCbLock();
+        if (clienttInst->onInternalDisconnectCb)
+            clienttInst->onInternalDisconnectCb(false);
         if (clienttInst->onDisconnectCb)
             clienttInst->onDisconnectCb(false);
     }