Fix deferred subscriptions

Peter M. Groen
1 parent 11fe0b09
Showing 3 changed files with 78 additions and 20 deletions
include/mqttclient.h
src/clientpaho.cpp
src/mqttclient.cpp
@@ -42,6 +42,26 @@ namespace osdev {
 namespace components {
 namespace mqtt {
+// Internal structure for storing subscriptions until a valid connection becomes available.
+class Subscription
+{
+public:
+    Subscription( const std::string &topic, int qos, const std::function<void(MqttMessage)>& call_back )
+        : m_topic( topic )
+        , m_qos( qos )
+        , m_call_back(call_back )
+    {}
+
+    std::string     getTopic() const { return m_topic; }
+    int             getQoS() const { return m_qos; }
+    std::function<void(MqttMessage)>&   getCallBack() const { return m_call_back; }
+
+private:
+    std::string m_topic;
+    int         m_qos;
+    std::function<void(MqttMessage)>& m_call_back;
+};
+
 // Forward definition
 class IMqttClientImpl;
@@ -215,6 +235,7 @@ private:
     mutable std::mutex m_interfaceMutex;                                    ///< Makes the interface mutual exclusive
     mutable std::mutex m_internalMutex;                                     ///< Protect the internal state.
+    mutable std::mutex m_subscriptionMutex;                                 ///< Protect the deferred Subscription Buffer
     std::string m_endpoint;                                                 ///< The endpoint uri.
     std::string m_clientId;                                                 ///< The main client identification.
     std::set<Token> m_activeTokens;                                         ///< Set with active command tokens. Callbacks still need to be made for these tokens.
@@ -225,6 +246,7 @@ private:
     std::vector<std::unique_ptr<IMqttClientImpl>> m_additionalClients;      ///< A vector of additional wrapper clients.
     SynchronizedQueue<std::function<void()>> m_eventQueue;                  ///< Synchronized queue for scheduling additional work.
     std::thread m_workerThread;                                             ///< A worker thread that is used to perform actions that cannot be done on the callback threads.
+    std::vector<Subscription> m_deferredSubscriptions;                       ///< A buffer to store subscription requests until the principal client comes online
 };
 }       // End namespace mqtt
@@ -680,7 +680,7 @@ std::int32_t ClientPaho::publishInternal( const MqttMessage&amp; message, int qos )
     if( !m_pendingOperations.insert( opts.token ).second )
     {
-        LogDebug( "[ClientPaho::publishInterval]", std::string( m_clientId + " publishInternal - token " + std::to_string( opts.token ) + " already in use" ) );
+        // LogDebug( "[ClientPaho::publishInterval]", std::string( m_clientId + " publishInternal - token " + std::to_string( opts.token ) + " already in use" ) );
     }
     m_operationResult.erase( opts.token );
     return opts.token;
@@ -718,10 +718,12 @@ std::int32_t ClientPaho::subscribeInternal( const std::string&amp; topic, int qos )
 void ClientPaho::setConnectionStatus( ConnectionStatus status )
 {
+    LogDebug( "[ClientPaho::setConnectionStatus]", std::string( m_clientId + " - " ) );
     ConnectionStatus curStatus = m_connectionStatus;
     m_connectionStatus = status;
     if( status != curStatus && m_connectionStatusCallback )
     {
+        LogDebug( "[ClientPaho::setConnectionStatus]", std::string( m_clientId + " - Calling m_connectionStatusCallback" ) );
         m_connectionStatusCallback( m_clientId, status );
     }
 }
@@ -787,6 +789,8 @@ void ClientPaho::onConnectOnInstance( const std::string&amp; cause )
 void ClientPaho::onConnectSuccessOnInstance()
 {
+    LogDebug( "[ClientPaho::onConnectSuccessOnInstance]",
+              std::string( m_clientId + " - onConnectSuccessOnInstance triggered." ) );
     {
         OSDEV_COMPONENTS_LOCKGUARD(m_mutex);
         // Register the connect callback that is used in reconnect scenarios.
@@ -892,7 +896,7 @@ void ClientPaho::onDisconnectFailureOnInstance( const MqttFailure&amp; response )
 void ClientPaho::onPublishSuccessOnInstance( const MqttSuccess& response )
 {
     auto pd = response.publishData();
-    LogDebug( "[ClientPaho::onPublishSuccessOnInstance]", std::string( m_clientId + " - publish with token " + std::to_string( response.token() ) + " succeeded ( message was " + pd.payload() + " )" ) );
+    // LogDebug( "[ClientPaho::onPublishSuccessOnInstance]", std::string( m_clientId + " - publish with token " + std::to_string( response.token() ) + " succeeded ( message was " + pd.payload() + " )" ) );
     {
         OSDEV_COMPONENTS_LOCKGUARD(m_mutex);
         // ("ClientPaho", "onPublishSuccessOnInstance %1 - pending operations : %2, removing operation %3", m_clientId, m_pendingOperations, response.token());
@@ -57,6 +57,7 @@ std::string generateUniqueClientId(const std::string&amp; clientId, std::size_t clie
 MqttClient::MqttClient(const std::string& _clientId, const std::function<void(const Token& token)>& deliveryCompleteCallback)
     : m_interfaceMutex()
     , m_internalMutex()
+    , m_subscriptionMutex()
     , m_endpoint()
     , m_clientId(_clientId)
     , m_activeTokens()
@@ -67,6 +68,7 @@ MqttClient::MqttClient(const std::string&amp; _clientId, const std::function&lt;void(co
     , m_additionalClients()
     , m_eventQueue(_clientId)
     , m_workerThread( std::thread( &MqttClient::eventHandler, this ) )
+    , m_deferredSubscriptions()
 {
     Log::init( "mqtt-library" );
     LogInfo( "MQTT Client started", "[MqttClient::MqttClient]");
@@ -248,12 +250,18 @@ Token MqttClient::subscribe(const std::string&amp; topic, int qos, const std::functi
     IMqttClientImpl* client(nullptr);
     {
         // OSDEV_COMPONENTS_LOCKGUARD(m_internalMutex);
-        if (!m_principalClient || m_principalClient->connectionStatus() == ConnectionStatus::Disconnected)
+        if (!m_principalClient || m_principalClient->connectionStatus() != ConnectionStatus::Connected)
         {
             LogError("MqttClient", std::string( m_clientId + " - Unable to subscribe, not connected" ) );
-            // throw (?)(MqttException, "Not connected");
+            // Store the subscription in the buffer for later processing.
+            {
+                OSDEV_COMPONENTS_LOCKGUARD(m_subscriptionMutex);
+                m_deferredSubscriptions.push_back( Subscription( topic, qos, cb ) );
+            }
+
             return Token(m_clientId, -1);
         }
+
         if (!m_principalClient->isOverlapping(topic))
         {
             client = m_principalClient.get();
@@ -270,6 +278,7 @@ Token MqttClient::subscribe(const std::string&amp; topic, int qos, const std::functi
                 }
             }
         }
+
         if (!clientFound)
         {
             LogDebug("[MqttClient::subscribe]", std::string( m_clientId + " - Creating new ClientPaho instance for subscription on topic " + topic ) );
@@ -282,9 +291,10 @@ Token MqttClient::subscribe(const std::string&amp; topic, int qos, const std::functi
             client = m_additionalClients.back().get();
         }
     }
+
     if (!clientFound)
     {
-        client->connect( false );
+        client->connect( true );
     }
     return Token{ client->clientId(), client->subscribe(topic, qos, cb) };
 }
@@ -386,32 +396,42 @@ void MqttClient::connectionStatusChanged(const std::string&amp; id, ConnectionStatus
     std::vector<ConnectionStatus> connectionStates{};
     {
         OSDEV_COMPONENTS_LOCKGUARD(m_internalMutex);
-        if (!m_principalClient) {
-            return;
-        }
-        if (m_principalClient) {
+
+        if (m_principalClient)
+        {
             principalClient = m_principalClient.get();
             clients.push_back(principalClient);
             connectionStates.push_back(m_principalClient->connectionStatus());
         }
-        for (const auto& c : m_additionalClients) {
+
+        for (const auto& c : m_additionalClients)
+        {
             clients.push_back(c.get());
             connectionStates.push_back(c->connectionStatus());
         }
     }
+
     auto newState = determineState(connectionStates);
-    bool resubscribe = (StateEnum::ConnectionFailure == m_serverState.state() && StateEnum::Good == newState);
+    // bool resubscribe = (StateEnum::ConnectionFailure == m_serverState.state() && StateEnum::Good == newState);
+    bool resubscribe = ( StateEnum::Good == newState );
     if (resubscribe)
     {
+        LogDebug( "[MqttClient::connectionStatusChanged]",
+                  std::string( m_clientId + " - Resubscribing..." ) );
         {
             OSDEV_COMPONENTS_LOCKGUARD(m_internalMutex);
             m_activeTokens.clear();
         }
-        for (auto* cl : clients) {
-            try {
+
+        for (auto* cl : clients)
+        {
+            try
+            {
+                LogDebug( "[MqttClient::connectionStatusChanged]", std::string( m_clientId + " - Client " + cl->clientId() + " has " + std::string( cl->hasPendingSubscriptions() ? "" : "no" ) + " pending subscriptions" ) );
                 cl->resubscribe();
             }
-            catch (const std::exception& e) {
+            catch (const std::exception& e)
+            {
                 LogError("[MqttClient::connectionStatusChanged]", std::string( m_clientId + " - resubscribe on wrapped client " + cl->clientId() + " in context of connection status change in wrapped client : " + id + " => FAILED : " + e.what() ) );
             }
         }
@@ -421,20 +441,32 @@ void MqttClient::connectionStatusChanged(const std::string&amp; id, ConnectionStatus
     // The server state change and a possible resubscription are done in the context of the MqttClient worker thread
     // The wrapper is free to pick up new work such as the acknowledment of the just recreated subscriptions.
     this->pushEvent([this, resubscribe, clients, principalClient, newState]() {
-        if (resubscribe) {
+        if (resubscribe)
+        {
             // Just wait for the subscription commands to complete. We do not use waitForCompletionInternal because that call will always timeout when there are active tokens.
             // Active tokens are removed typically by work done on the worker thread. The wait action is also performed on the worker thread.
             auto waitFor = std::chrono::milliseconds(1000);
-            if (!waitForCompletionInternalClients(clients, waitFor, std::set<Token>{})) {
-                if (std::accumulate(clients.begin(), clients.end(), false, [](bool hasPending, IMqttClientImpl* client) { return hasPending || client->hasPendingSubscriptions(); })) {
+            if (!waitForCompletionInternalClients(clients, waitFor, std::set<Token>{}))
+            {
+                if (std::accumulate(clients.begin(),
+                                    clients.end(),
+                                    false,
+                                    [](bool hasPending, IMqttClientImpl* client)
+                                    {
+                                        return hasPending || client->hasPendingSubscriptions();
+                                    }))
+                {
                     LogWarning("[MqttClient::connectionStatusChanged]", std::string( m_clientId + " - subscriptions are not recovered within timeout." ) );
                 }
             }
-            if (principalClient) {
-                try {
+            if (principalClient)
+            {
+                try
+                {
                     principalClient->publishPending();
                 }
-                catch (const std::exception& e) {
+                catch (const std::exception& e)
+                {
                     LogError( "[MqttClient::connectionStatusChanged]", std::string( m_clientId + " - publishPending on wrapped client " + principalClient->clientId() + " => FAILED " + e.what() ) );
                 }
             }