From d557d523819c9e3d5cce08fbe8fc46ea27ec7aa4 Mon Sep 17 00:00:00 2001 From: Peter M. Groen Date: Fri, 8 Jul 2022 01:51:29 +0200 Subject: [PATCH] Fix deferred subscriptions --- include/mqttclient.h | 22 ++++++++++++++++++++++ src/clientpaho.cpp | 8 ++++++-- src/mqttclient.cpp | 68 ++++++++++++++++++++++++++++++++++++++++++++++++++------------------ 3 files changed, 78 insertions(+), 20 deletions(-) diff --git a/include/mqttclient.h b/include/mqttclient.h index 391908a..0f1d7db 100644 --- a/include/mqttclient.h +++ b/include/mqttclient.h @@ -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& 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& getCallBack() const { return m_call_back; } + +private: + std::string m_topic; + int m_qos; + std::function& 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 m_activeTokens; ///< Set with active command tokens. Callbacks still need to be made for these tokens. @@ -225,6 +246,7 @@ private: std::vector> m_additionalClients; ///< A vector of additional wrapper clients. SynchronizedQueue> 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 m_deferredSubscriptions; ///< A buffer to store subscription requests until the principal client comes online }; } // End namespace mqtt diff --git a/src/clientpaho.cpp b/src/clientpaho.cpp index 5ec2436..1dcbf6a 100644 --- a/src/clientpaho.cpp +++ b/src/clientpaho.cpp @@ -680,7 +680,7 @@ std::int32_t ClientPaho::publishInternal( const MqttMessage& 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& 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& 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& 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()); diff --git a/src/mqttclient.cpp b/src/mqttclient.cpp index dbf0d79..b3276b9 100644 --- a/src/mqttclient.cpp +++ b/src/mqttclient.cpp @@ -57,6 +57,7 @@ std::string generateUniqueClientId(const std::string& clientId, std::size_t clie MqttClient::MqttClient(const std::string& _clientId, const std::function& deliveryCompleteCallback) : m_interfaceMutex() , m_internalMutex() + , m_subscriptionMutex() , m_endpoint() , m_clientId(_clientId) , m_activeTokens() @@ -67,6 +68,7 @@ MqttClient::MqttClient(const std::string& _clientId, const std::functionconnectionStatus() == 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& 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& 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& id, ConnectionStatus std::vector 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& 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{})) { - if (std::accumulate(clients.begin(), clients.end(), false, [](bool hasPending, IMqttClientImpl* client) { return hasPending || client->hasPendingSubscriptions(); })) { + if (!waitForCompletionInternalClients(clients, waitFor, std::set{})) + { + 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() ) ); } } -- libgit2 0.21.4