feat: implement last-will support

Patric Stout
1 parent 584030be
Showing 5 changed files with 49 additions and 35 deletions
example/pubsub/main.cpp
include/TrueMQTT.h
src/Client.cpp
src/ClientImpl.h
src/Packet.cpp
@@ -19,6 +19,7 @@ int main()
     client.setPublishQueue(TrueMQTT::Client::PublishQueueType::FIFO, 10);
     client.setErrorCallback([](TrueMQTT::Client::Error error, std::string message)
                             { std::cout << "Error " << error << ": " << message << std::endl; });
+    client.setLastWill("test/lastwill", "example pubsub finished", true);
     client.connect();
     std::this_thread::sleep_for(std::chrono::milliseconds(100));
@@ -146,12 +146,12 @@ namespace TrueMQTT
          * @brief Set the last will message on the connection.
          *
          * @param topic The topic to publish the last will message to.
-         * @param payload The payload of the last will message.
+         * @param message The message of the last will message.
          * @param retain Whether to retain the last will message.
          *
          * @note Cannot be called after \ref connect.
          */
-        void setLastWill(const std::string &topic, const std::string &payload, bool retain) const;
+        void setLastWill(const std::string &topic, const std::string &message, bool retain) const;
         /**
          * @brief Set the error callback, called when any error occurs.
@@ -198,14 +198,14 @@ namespace TrueMQTT
         void disconnect() const;
         /**
-         * @brief Publish a payload on a topic.
+         * @brief Publish a message on a topic.
          *
          * After \ref connect is called, this function will either publish the message
          * immediately (if connected) or queue it for later (if still connecting).
          * In the latter case, it will be published as soon as the connection is established.
          *
-         * @param topic The topic to publish the payload on.
-         * @param payload The payload to publish.
+         * @param topic The topic to publish the message on.
+         * @param message The message to publish.
          * @param retain Whether to retain the message on the broker.
          *
          * @note All messages are always published under QoS 0, and this library supports no
@@ -218,7 +218,7 @@ namespace TrueMQTT
          * moment the connection to the broker is established, and there are messages in the
          * publish queue and/or subscriptions.
          */
-        void publish(const std::string &topic, const std::string &payload, bool retain) const;
+        void publish(const std::string &topic, const std::string &message, bool retain) const;
         /**
          * @brief Subscribe to a topic, and call the callback function when a message arrives.
@@ -64,7 +64,7 @@ void TrueMQTT::Client::setLogger(Client::LogLevel log_level, const std::function
     LOG_DEBUG(m_impl, "Log level now on " + std::to_string(m_impl->m_log_level));
 }
-void TrueMQTT::Client::setLastWill(const std::string &topic, const std::string &payload, bool retain) const
+void TrueMQTT::Client::setLastWill(const std::string &topic, const std::string &message, bool retain) const
 {
     if (m_impl->m_state != Client::Impl::State::DISCONNECTED)
     {
@@ -72,10 +72,10 @@ void TrueMQTT::Client::setLastWill(const std::string &amp;topic, const std::string &amp;
         return;
     }
-    LOG_TRACE(m_impl, "Setting last will to topic " + topic + " with payload " + payload + " and retain " + std::to_string(retain));
+    LOG_TRACE(m_impl, "Setting last will to topic " + topic + " with message " + message + " and retain " + std::to_string(retain));
     m_impl->m_last_will_topic = topic;
-    m_impl->m_last_will_payload = payload;
+    m_impl->m_last_will_message = message;
     m_impl->m_last_will_retain = retain;
 }
@@ -131,11 +131,11 @@ void TrueMQTT::Client::disconnect() const
     m_impl->disconnect();
 }
-void TrueMQTT::Client::publish(const std::string &topic, const std::string &payload, bool retain) const
+void TrueMQTT::Client::publish(const std::string &topic, const std::string &message, bool retain) const
 {
     std::scoped_lock lock(m_impl->m_state_mutex);
-    LOG_DEBUG(m_impl, "Publishing message on topic '" + topic + "': " + payload + " (" + (retain ? "retained" : "not retained") + ")");
+    LOG_DEBUG(m_impl, "Publishing message on topic '" + topic + "': " + message + " (" + (retain ? "retained" : "not retained") + ")");
     switch (m_impl->m_state)
     {
@@ -143,10 +143,10 @@ void TrueMQTT::Client::publish(const std::string &amp;topic, const std::string &amp;payl
         LOG_ERROR(m_impl, "Cannot publish when disconnected");
         return;
     case Client::Impl::State::CONNECTING:
-        m_impl->toPublishQueue(topic, payload, retain);
+        m_impl->toPublishQueue(topic, message, retain);
         return;
     case Client::Impl::State::CONNECTED:
-        m_impl->sendPublish(topic, payload, retain);
+        m_impl->sendPublish(topic, message, retain);
         return;
     }
 }
@@ -266,9 +266,9 @@ void TrueMQTT::Client::Impl::connectionStateChange(bool connected)
             sendSubscribe(subscription);
         }
         // Flush the publish queue.
-        for (const auto &[topic, payload, retain] : m_publish_queue)
+        for (const auto &[topic, message, retain] : m_publish_queue)
         {
-            sendPublish(topic, payload, retain);
+            sendPublish(topic, message, retain);
         }
         m_publish_queue.clear();
     }
@@ -279,7 +279,7 @@ void TrueMQTT::Client::Impl::connectionStateChange(bool connected)
     }
 }
-void TrueMQTT::Client::Impl::toPublishQueue(const std::string &topic, const std::string &payload, bool retain)
+void TrueMQTT::Client::Impl::toPublishQueue(const std::string &topic, const std::string &message, bool retain)
 {
     if (m_state != Client::Impl::State::CONNECTING)
     {
@@ -309,7 +309,7 @@ void TrueMQTT::Client::Impl::toPublishQueue(const std::string &amp;topic, const std:
     }
     LOG_TRACE(this, "Adding message to publish queue");
-    m_publish_queue.emplace_back(topic, payload, retain);
+    m_publish_queue.emplace_back(topic, message, retain);
 }
 void TrueMQTT::Client::Impl::findSubscriptionMatch(std::vector<std::function<void(std::string, std::string)>> &matching_callbacks, const std::map<std::string, Client::Impl::SubscriptionPart> &subscriptions, std::deque<std::string> &parts)
@@ -357,9 +357,9 @@ void TrueMQTT::Client::Impl::findSubscriptionMatch(std::vector&lt;std::function&lt;voi
     }
 }
-void TrueMQTT::Client::Impl::messageReceived(std::string topic, std::string payload)
+void TrueMQTT::Client::Impl::messageReceived(std::string topic, std::string message)
 {
-    LOG_TRACE(this, "Message received on topic '" + topic + "': " + payload);
+    LOG_TRACE(this, "Message received on topic '" + topic + "': " + message);
     // Split the topic on the / in parts.
     std::string part;
@@ -378,14 +378,14 @@ void TrueMQTT::Client::Impl::messageReceived(std::string topic, std::string payl
     if (matching_callbacks.size() == 1)
     {
-        // For a single callback there is no need to copy the topic/payload.
-        matching_callbacks[0](std::move(topic), std::move(payload));
+        // For a single callback there is no need to copy the topic/message.
+        matching_callbacks[0](std::move(topic), std::move(message));
     }
     else
     {
         for (const auto &callback : matching_callbacks)
         {
-            callback(topic, payload);
+            callback(topic, message);
         }
     }
 }
@@ -47,12 +47,12 @@ public:
     void connect();                                                                         ///< Connect to the broker.
     void disconnect();                                                                      ///< Disconnect from the broker.
-    void sendPublish(const std::string &topic, const std::string &payload, bool retain);    ///< Send a publish message to the broker.
+    void sendPublish(const std::string &topic, const std::string &message, bool retain);    ///< Send a publish message to the broker.
     void sendSubscribe(const std::string &topic);                                           ///< Send a subscribe message to the broker.
     void sendUnsubscribe(const std::string &topic);                                         ///< Send an unsubscribe message to the broker.
     void connectionStateChange(bool connected);                                             ///< Called when a connection goes from CONNECTING state to CONNECTED state or visa versa.
-    void toPublishQueue(const std::string &topic, const std::string &payload, bool retain); ///< Add a publish message to the publish queue.
-    void messageReceived(std::string topic, std::string payload);                           ///< Called when a message is received from the broker.
+    void toPublishQueue(const std::string &topic, const std::string &message, bool retain); ///< Add a publish message to the publish queue.
+    void messageReceived(std::string topic, std::string message);                           ///< Called when a message is received from the broker.
     void findSubscriptionMatch(std::vector<std::function<void(std::string, std::string)>> &callbacks, const std::map<std::string, SubscriptionPart> &subscriptions, std::deque<std::string> &parts); ///< Recursive function to find any matching subscription based on parts.
@@ -71,7 +71,7 @@ public:
     std::function<void(Client::LogLevel, std::string)> m_logger = [](Client::LogLevel, std::string) { /* empty */ }; ///< Logger callback.
     std::string m_last_will_topic = "";   ///< Topic to publish the last will message to.
-    std::string m_last_will_payload = ""; ///< Payload of the last will message.
+    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.
     std::function<void(Error, std::string)> m_error_callback = [](Error, std::string) { /* empty */ }; ///< Error callback.
@@ -265,12 +265,12 @@ bool TrueMQTT::Client::Impl::Connection::recvLoop()
             return false;
         }
-        std::string payload;
-        packet.read_remaining(payload);
+        std::string message;
+        packet.read_remaining(message);
-        LOG_DEBUG(&m_impl, "Received PUBLISH with topic " + topic + ": " + payload);
+        LOG_DEBUG(&m_impl, "Received PUBLISH with topic " + topic + ": " + message);
-        m_impl.messageReceived(std::move(topic), std::move(payload));
+        m_impl.messageReceived(std::move(topic), std::move(message));
         break;
     }
     case Packet::PacketType::SUBACK:
@@ -366,7 +366,16 @@ void TrueMQTT::Client::Impl::Connection::sendConnect()
     uint8_t flags = 0;
     flags |= 1 << 1; // Clean session
-    // TODO -- Support for last-will
+    if (!m_impl.m_last_will_topic.empty())
+    {
+        flags |= 1 << 2; // Last will
+        flags |= 0 << 3; // Last will QoS
+
+        if (m_impl.m_last_will_retain)
+        {
+            flags |= 1 << 5; // Last will retain
+        }
+    }
     Packet packet(Packet::PacketType::CONNECT, 0);
@@ -377,14 +386,18 @@ void TrueMQTT::Client::Impl::Connection::sendConnect()
     packet.write_uint16(30); // Keep-alive
     packet.write_string(client_id); // Client ID
-    // TODO -- Last will topic & message
+    if (!m_impl.m_last_will_topic.empty())
+    {
+        packet.write_string(m_impl.m_last_will_topic);
+        packet.write_string(m_impl.m_last_will_message);
+    }
     send(packet);
 }
-void TrueMQTT::Client::Impl::sendPublish(const std::string &topic, const std::string &payload, bool retain)
+void TrueMQTT::Client::Impl::sendPublish(const std::string &topic, const std::string &message, bool retain)
 {
-    LOG_TRACE(this, "Sending PUBLISH packet to topic '" + topic + "': " + payload + " (" + (retain ? "retained" : "not retained") + ")");
+    LOG_TRACE(this, "Sending PUBLISH packet to topic '" + topic + "': " + message + " (" + (retain ? "retained" : "not retained") + ")");
     uint8_t flags = 0;
     flags |= (retain ? 1 : 0) << 0; // Retain
@@ -394,7 +407,7 @@ void TrueMQTT::Client::Impl::sendPublish(const std::string &amp;topic, const std::st
     Packet packet(Packet::PacketType::PUBLISH, flags);
     packet.write_string(topic);
-    packet.write(payload.c_str(), payload.size());
+    packet.write(message.c_str(), message.size());
     m_connection->send(packet);
 }