feat: callback to be informed about state changes

Useful if you want to take action when a connection to the broker is established or lost. For example, after a reconnection you can expect to receive all retained messages again on the topics you are subscribed too.

feat: callback to be informed about state changes
Useful if you want to take action when a connection to the broker is established or lost. For example, after a reconnection you can expect to receive all retained messages again on the topics you are subscribed too.
Patric Stout
1 parent 6f0b3228
Showing 6 changed files with 61 additions and 34 deletions
example/pubsub/main.cpp
example/stress/main.cpp
include/TrueMQTT.h
src/Client.cpp
src/ClientImpl.h
src/Packet.cpp
@@ -6,7 +6,9 @@
  */
 #include <TrueMQTT.h>
+
 #include <iostream>
+#include <magic_enum.hpp>
 #include <thread>
 int main()
@@ -15,10 +17,10 @@ int main()
     TrueMQTT::Client client("localhost", 1883, "test");
     client.setLogger(TrueMQTT::Client::LogLevel::WARNING, [](TrueMQTT::Client::LogLevel level, std::string_view message)
-                     { std::cout << "Log " << level << ": " << message << std::endl; });
+                     { std::cout << "Log " << std::string(magic_enum::enum_name(level)) << ": " << message << std::endl; });
     client.setPublishQueue(TrueMQTT::Client::PublishQueueType::FIFO, 10);
     client.setErrorCallback([](TrueMQTT::Client::Error error, std::string_view message)
-                            { std::cout << "Error " << error << ": " << message << std::endl; });
+                            { std::cout << "Error " << std::string(magic_enum::enum_name(error)) << ": " << message << std::endl; });
     client.setLastWill("example/pubsub/lastwill", "example pubsub finished", true);
     client.connect();
@@ -6,7 +6,9 @@
  */
 #include <TrueMQTT.h>
+
 #include <iostream>
+#include <magic_enum.hpp>
 #include <thread>
 int main()
@@ -15,10 +17,10 @@ int main()
     TrueMQTT::Client client("localhost", 1883, "test");
     client.setLogger(TrueMQTT::Client::LogLevel::WARNING, [](TrueMQTT::Client::LogLevel level, std::string_view message)
-                     { std::cout << "Log " << level << ": " << message << std::endl; });
+                     { std::cout << "Log " << std::string(magic_enum::enum_name(level)) << ": " << message << std::endl; });
     client.setPublishQueue(TrueMQTT::Client::PublishQueueType::FIFO, 100);
     client.setErrorCallback([](TrueMQTT::Client::Error error, std::string_view message)
-                            { std::cout << "Error " << error << ": " << message << std::endl; });
+                            { std::cout << "Error " << std::string(magic_enum::enum_name(error)) << ": " << message << std::endl; });
     client.setLastWill("test/lastwill", "example pubsub finished", true);
     client.connect();
@@ -24,7 +24,7 @@ namespace TrueMQTT
         /**
          * @brief Error codes that can be returned in the callback set by \ref setErrorCallback.
          */
-        enum Error
+        enum class Error
         {
             /**
              * @brief The hostname could not be resolved into either an IPv4 or IPv6 address.
@@ -62,7 +62,7 @@ namespace TrueMQTT
          * After all, memory is finite, so this allows you to configure what scenario
          * works best for you.
          */
-        enum PublishQueueType
+        enum class PublishQueueType
         {
             DROP, ///< Do not queue.
@@ -96,7 +96,7 @@ namespace TrueMQTT
         /**
          * @brief The log levels used by this library.
          */
-        enum LogLevel
+        enum class LogLevel
         {
             NONE,    ///< Do not log anything (default).
             ERROR,   ///< Something went wrong and the library cannot recover.
@@ -107,6 +107,16 @@ namespace TrueMQTT
         };
         /**
+         * @brief The state of the connection.
+         */
+        enum class State
+        {
+            DISCONNECTED, ///< The client is not connected to a broker.
+            CONNECTING,   ///< The client is (re)connecting to a broker.
+            CONNECTED,    ///< The client is connected to a broker.
+        };
+
+        /**
          * @brief Constructor for the MQTT client.
          *
          * @param host The hostname of the MQTT broker. Can be either an IP or a domain name.
@@ -195,6 +205,13 @@ namespace TrueMQTT
         void setSendQueue(size_t size) const;
         /**
+         * @brief Set the state-change callback.
+         *
+         * @param callback The callback to call when the state changes.
+         */
+        void setStateChangeCallback(const std::function<void(State)> &callback) const;
+
+        /**
          * @brief Connect to the broker.
          *
          * After calling this function, the library will try a connection to the broker.
@@ -64,7 +64,7 @@ void TrueMQTT::Client::setLogger(Client::LogLevel log_level, const std::function
 void TrueMQTT::Client::setLastWill(const std::string_view topic, const std::string_view message, bool retain) const
 {
-    if (m_impl->m_state != Client::Impl::State::DISCONNECTED)
+    if (m_impl->m_state != Client::State::DISCONNECTED)
     {
         LOG_ERROR(m_impl, "Cannot set last will when not disconnected");
         return;
@@ -86,7 +86,7 @@ void TrueMQTT::Client::setErrorCallback(const std::function&lt;void(Error, std::str
 void TrueMQTT::Client::setPublishQueue(Client::PublishQueueType queue_type, size_t size) const
 {
-    if (m_impl->m_state != Client::Impl::State::DISCONNECTED)
+    if (m_impl->m_state != Client::State::DISCONNECTED)
     {
         LOG_ERROR(m_impl, "Cannot set publish queue when not disconnected");
         return;
@@ -100,7 +100,7 @@ void TrueMQTT::Client::setPublishQueue(Client::PublishQueueType queue_type, size
 void TrueMQTT::Client::setSendQueue(size_t size) const
 {
-    if (m_impl->m_state != Client::Impl::State::DISCONNECTED)
+    if (m_impl->m_state != Client::State::DISCONNECTED)
     {
         LOG_ERROR(m_impl, "Cannot set send queue when not disconnected");
         return;
@@ -111,18 +111,27 @@ void TrueMQTT::Client::setSendQueue(size_t size) const
     m_impl->m_send_queue_size = size;
 }
+void TrueMQTT::Client::setStateChangeCallback(const std::function<void(State)> &callback) const
+{
+    LOG_TRACE(m_impl, "Setting state callback");
+
+    m_impl->m_state_change_callback = callback;
+}
+
 void TrueMQTT::Client::connect() const
 {
     std::scoped_lock lock(m_impl->m_state_mutex);
-    if (m_impl->m_state != Client::Impl::State::DISCONNECTED)
+    if (m_impl->m_state != Client::State::DISCONNECTED)
     {
+        LOG_ERROR(m_impl, "Can't connect when already connecting / connected");
         return;
     }
     LOG_INFO(m_impl, "Connecting to " + m_impl->m_host + ":" + std::to_string(m_impl->m_port));
-    m_impl->m_state = Client::Impl::State::CONNECTING;
+    m_impl->m_state = Client::State::CONNECTING;
+    m_impl->m_state_change_callback(m_impl->m_state);
     m_impl->connect();
 }
@@ -130,15 +139,16 @@ void TrueMQTT::Client::disconnect() const
 {
     std::scoped_lock lock(m_impl->m_state_mutex);
-    if (m_impl->m_state == Client::Impl::State::DISCONNECTED)
+    if (m_impl->m_state == Client::State::DISCONNECTED)
     {
-        LOG_TRACE(m_impl, "Already disconnected");
+        LOG_ERROR(m_impl, "Can't disconnect when already disconnected");
         return;
     }
     LOG_INFO(m_impl, "Disconnecting from broker");
-    m_impl->m_state = Client::Impl::State::DISCONNECTED;
+    m_impl->m_state = Client::State::DISCONNECTED;
+    m_impl->m_state_change_callback(m_impl->m_state);
     m_impl->disconnect();
 }
@@ -150,12 +160,12 @@ bool TrueMQTT::Client::publish(const std::string_view topic, const std::string_v
     switch (m_impl->m_state)
     {
-    case Client::Impl::State::DISCONNECTED:
+    case Client::State::DISCONNECTED:
         LOG_ERROR(m_impl, "Cannot publish when disconnected");
         return false;
-    case Client::Impl::State::CONNECTING:
+    case Client::State::CONNECTING:
         return m_impl->toPublishQueue(topic, message, retain);
-    case Client::Impl::State::CONNECTED:
+    case Client::State::CONNECTED:
         return m_impl->sendPublish(topic, message, retain);
     }
@@ -166,7 +176,7 @@ void TrueMQTT::Client::subscribe(const std::string_view topic, const std::functi
 {
     std::scoped_lock lock(m_impl->m_state_mutex);
-    if (m_impl->m_state == Client::Impl::State::DISCONNECTED)
+    if (m_impl->m_state == Client::State::DISCONNECTED)
     {
         LOG_ERROR(m_impl, "Cannot subscribe when disconnected");
         return;
@@ -210,7 +220,7 @@ void TrueMQTT::Client::subscribe(const std::string_view topic, const std::functi
     subscriptions->callbacks.push_back(callback);
     m_impl->m_subscription_topics.emplace(topic);
-    if (m_impl->m_state == Client::Impl::State::CONNECTED)
+    if (m_impl->m_state == Client::State::CONNECTED)
     {
         if (!m_impl->sendSubscribe(topic))
         {
@@ -225,7 +235,7 @@ void TrueMQTT::Client::unsubscribe(const std::string_view topic) const
 {
     std::scoped_lock lock(m_impl->m_state_mutex);
-    if (m_impl->m_state == Client::Impl::State::DISCONNECTED)
+    if (m_impl->m_state == Client::State::DISCONNECTED)
     {
         LOG_ERROR(m_impl, "Cannot unsubscribe when disconnected");
         return;
@@ -302,7 +312,7 @@ void TrueMQTT::Client::unsubscribe(const std::string_view topic) const
     }
     m_impl->m_subscription_topics.erase(m_impl->m_subscription_topics.find(topic));
-    if (m_impl->m_state == Client::Impl::State::CONNECTED)
+    if (m_impl->m_state == Client::State::CONNECTED)
     {
         if (!m_impl->sendUnsubscribe(topic))
         {
@@ -321,7 +331,8 @@ void TrueMQTT::Client::Impl::connectionStateChange(bool connected)
     {
         LOG_INFO(this, "Connected to broker");
-        m_state = Client::Impl::State::CONNECTED;
+        m_state = Client::State::CONNECTED;
+        m_state_change_callback(m_state);
         // Restoring subscriptions and flushing the queue is done while still under
         // the lock. This to prevent \ref disconnect from being called while we are
@@ -356,13 +367,14 @@ void TrueMQTT::Client::Impl::connectionStateChange(bool connected)
     else
     {
         LOG_INFO(this, "Disconnected from broker");
-        m_state = Client::Impl::State::CONNECTING;
+        m_state = Client::State::CONNECTING;
+        m_state_change_callback(m_state);
     }
 }
 bool TrueMQTT::Client::Impl::toPublishQueue(const std::string_view topic, const std::string_view message, bool retain)
 {
-    if (m_state != Client::Impl::State::CONNECTING)
+    if (m_state != Client::State::CONNECTING)
     {
         LOG_ERROR(this, "Cannot queue publish message when not connecting");
         return false;
@@ -31,13 +31,6 @@ public:
          std::chrono::milliseconds keep_alive_interval);
     ~Impl();
-    enum State
-    {
-        DISCONNECTED, ///< The client is not connected to the broker, nor is it trying to connect.
-        CONNECTING,   ///< The client is either connecting or reconnecting to the broker. This can be in any stage of the connection process.
-        CONNECTED,    ///< The client is connected to the broker.
-    };
-
     class SubscriptionPart
     {
     public:
@@ -70,6 +63,8 @@ public:
     Client::LogLevel m_log_level = Client::LogLevel::NONE;                                                                     ///< The log level to use.
     std::function<void(Client::LogLevel, std::string_view)> m_logger = [](Client::LogLevel, std::string_view) { /* empty */ }; ///< Logger callback.
+    std::function<void(State)> m_state_change_callback = [](State) { /* empty */ }; ///< Callback when the connection state changes.
+
     std::string m_last_will_topic = "";   ///< Topic to publish the last will message to.
     std::string m_last_will_message = ""; ///< Message to publish on the last will topic.
     bool m_last_will_retain = false;      ///< Whether to retain the last will message.
@@ -10,8 +10,7 @@
 #include "Log.h"
 #include "Packet.h"
-#include "magic_enum.hpp"
-
+#include <magic_enum.hpp>
 #include <string.h>
 ssize_t TrueMQTT::Client::Impl::Connection::recv(char *buffer, size_t length)