Merge branch 'feat/sridder/will_set' into 'development'

added LWT ( last will & testament ) option to mqtt connect. added support for a last will & testament message. LWT settings are not required for connection, but can be provided as optional parameter when calling the mqttclient connect function. ( see publisher test case ). The parameters to be added are the topic to publish on and the message to be published in case of a connection disruption. See merge request !8

Merge branch 'feat/sridder/will_set' into 'development'
added LWT ( last will & testament ) option to mqtt connect. added support for a last will & testament message. LWT settings are not required for connection, but can be provided as optional parameter when calling the mqttclient connect function. ( see publisher test case ). The parameters to be added are the topic to publish on and the message to be published in case of a connection disruption. See merge request !8
Peter M. Groen
2 parents a1927923 55d805d7
Showing 10 changed files with 106 additions and 19 deletions
examples/pub/main.cpp
examples/pub/publisher.cpp
examples/pub/publisher.h
include/clientpaho.h
include/imqttclient.h
include/imqttclientimpl.h
include/mqtt_lwt.h
include/mqttclient.h
src/clientpaho.cpp
src/mqttclient.cpp
@@ -78,7 +78,7 @@ int main( int argc, char* argv[] )
     {
         std::cout << "{OK}" << std::endl;
         std::cout << "Connecting to the broker : ";
-        pPublisher->connect( "localhost", 1883, "", "" );
+        pPublisher->connect( "localhost", 1883, "", "", "LWT-test", "connection disrupted.." );
         // Assume we are connected now, start publishing.
         while( 1 )
@@ -94,7 +94,7 @@ int main( int argc, char* argv[] )
                 }
                 messageNumber++;
             }
-            sleepcp( 1, T_MICRO );
+            sleepcp( 5, T_SECONDS );
         }
     }
     else
@@ -29,9 +29,9 @@ Publisher::Publisher()
 }
-void Publisher::connect( const std::string &hostname, int portnumber, const std::string &username, const std::string &password )
+void Publisher::connect(const std::string &hostname, int portnumber, const std::string &username, const std::string &password, const std::string &lwt_topic, const std::string &lwt_message)
 {
-    m_mqtt_client.connect( hostname, portnumber, osdev::components::mqtt::Credentials( username, password ) );
+    m_mqtt_client.connect( hostname, portnumber, osdev::components::mqtt::Credentials( username, password ), osdev::components::mqtt::mqtt_LWT( lwt_topic, lwt_message ) );
     std::cout << "Client state : " << m_mqtt_client.state() << std::endl;
 }
@@ -36,7 +36,8 @@ public:
     virtual ~Publisher() {}
-    void    connect( const std::string &hostname, int portnumber = 1883, const std::string &username = std::string(), const std::string &password = std::string() );
+    void    connect( const std::string &hostname, int portnumber = 1883, const std::string &username = std::string(), const std::string &password = std::string()
+            , const std::string &lwt_topic = std::string(), const std::string &lwt_message = std::string() );
     void    publish( const std::string &message_topic, const std::string &message_payload );
@@ -45,6 +45,7 @@
 #include "imqttclientimpl.h"
 #include "mqttfailure.h"
 #include "mqttsuccess.h"
+#include "mqtt_lwt.h"
 namespace osdev {
 namespace components {
@@ -94,7 +95,7 @@ public:
     /**
      * @see IMqttClientImpl
      */
-    virtual std::int32_t connect(bool wait) override;
+    virtual std::int32_t connect(bool wait, const mqtt_LWT &lwt = mqtt_LWT() ) override;
     /**
      * @see IMqttClientImpl
@@ -39,6 +39,7 @@
 #include "credentials.h"
 #include "mqttmessage.h"
 #include "token.h"
+#include "mqtt_lwt.h"
 namespace osdev {
 namespace components {
@@ -59,13 +60,13 @@ public:
      * @param port The port to use.
      * @param credentials The credentials to use.
      */
-    virtual void connect(const std::string& host, int port, const Credentials& credentials) = 0;
+    virtual void connect(const std::string& host, int port, const Credentials& credentials, const mqtt_LWT &lwt = mqtt_LWT() ) = 0;
     /**
      * @brief Connect to the endpoint
      * @param endpoint an uri endpoint description.
      */
-    virtual void connect(const std::string& endpoint) = 0;
+    virtual void connect(const std::string& endpoint, const mqtt_LWT &lwt = mqtt_LWT() ) = 0;
     /**
      * @brief Disconnect the client from the broker
@@ -36,6 +36,7 @@
 // mlogic::mqtt
 #include "connectionstatus.h"
 #include "mqttmessage.h"
+#include "mqtt_lwt.h"
 namespace osdev {
 namespace components {
@@ -65,7 +66,7 @@ public:
      * @param wait A flag that indicates if the method should wait for a succesful connection.
      * @return the operation token.
      */
-    virtual std::int32_t connect(bool wait) = 0;
+    virtual std::int32_t connect( bool wait, const mqtt_LWT &lwt = mqtt_LWT() ) = 0;
     /**
      * @brief Disconnect the wrapper.
+/* ****************************************************************************
+ * Copyright 2019 Open Systems Development BV                                 *
+ *                                                                            *
+ * Permission is hereby granted, free of charge, to any person obtaining a    *
+ * copy of this software and associated documentation files (the "Software"), *
+ * to deal in the Software without restriction, including without limitation  *
+ * the rights to use, copy, modify, merge, publish, distribute, sublicense,   *
+ * and/or sell copies of the Software, and to permit persons to whom the      *
+ * Software is furnished to do so, subject to the following conditions:       *
+ *                                                                            *
+ * The above copyright notice and this permission notice shall be included in *
+ * all copies or substantial portions of the Software.                        *
+ *                                                                            *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR *
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,   *
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL    *
+ * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER *
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING    *
+ * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER        *
+ * DEALINGS IN THE SOFTWARE.                                                  *
+ * ***************************************************************************/
+#pragma once
+
+// std
+#include <string>
+
+namespace osdev {
+namespace components {
+namespace mqtt {
+
+/*!
+ *  \brief  Class that holds LWT ( Last Will & Testament ) context.
+ */
+class mqtt_LWT
+{
+public:
+    /*!
+     *  \brief Default CTor, empty LWT
+     */
+    mqtt_LWT()
+        : m_topic()
+        , m_message()
+    {}
+
+    /*!
+     *  \brief  Constructor for LWT topic/message
+     *  \param  topic    - The topic on which the LWT is published.
+     *  \param  message  - The message published on broker disconnection.
+     */
+    mqtt_LWT( const std::string &topic, const std::string &message )
+        : m_topic( topic )
+        , m_message( message )
+    {}
+
+    const std::string& topic()   const { return m_topic; }
+    const std::string& message() const { return m_message; }
+
+private:
+    std::string m_topic;
+    std::string m_message;
+};
+
+}   // End namespace mqtt
+}   // End namespace components
+}   // End namespace osdev
@@ -36,6 +36,7 @@
 #include "serverstate.h"
 #include "imqttclient.h"
+#include "mqtt_lwt.h"
 namespace osdev {
 namespace components {
@@ -91,12 +92,12 @@ public:
     /**
      * @see IMqttClient
      */
-    virtual void connect(const std::string& host, int port, const Credentials& credentials) override;
+    virtual void connect( const std::string& host, int port, const Credentials &credentials, const mqtt_LWT &lwt = mqtt_LWT() ) override;
     /**
      * @see IMqttClient
      */
-    virtual void connect(const std::string& endpoint) override;
+    virtual void connect( const std::string &endpoint, const mqtt_LWT &lwt = mqtt_LWT() ) override;
     /**
      * @see IMqttClient
@@ -182,7 +182,7 @@ ConnectionStatus ClientPaho::connectionStatus() const
     return m_connectionStatus;
 }
-std::int32_t ClientPaho::connect(bool wait)
+std::int32_t ClientPaho::connect( bool wait, const mqtt_LWT &lwt )
 {
     {
         OSDEV_COMPONENTS_LOCKGUARD(m_mutex);
@@ -200,6 +200,21 @@ std::int32_t ClientPaho::connect(bool wait)
     conn_opts.onFailure = &ClientPaho::onConnectFailure;
     conn_opts.context = this;
     conn_opts.automaticReconnect = 1;
+
+    if( !lwt.topic().empty() )
+    {
+        MQTTAsync_willOptions will_opts = MQTTAsync_willOptions_initializer;
+        will_opts.message = lwt.message().c_str();
+        will_opts.topicName = lwt.topic().c_str();
+
+        conn_opts.will = &will_opts;
+    }
+    else
+    {
+        conn_opts.will = nullptr;
+    }
+
+
     if (!m_username.empty())
     {
         conn_opts.username = m_username.c_str();
@@ -451,7 +466,7 @@ void ClientPaho::resubscribe()
     }
 }
-std::int32_t ClientPaho::unsubscribe(const std::string& topic, int qos)
+std::int32_t ClientPaho::unsubscribe( const std::string& topic, int qos )
 {
     {
         OSDEV_COMPONENTS_LOCKGUARD(m_mutex);
@@ -125,9 +125,8 @@ StateEnum MqttClient::state() const
     return m_serverState.state();
 }
-void MqttClient::connect(const std::string& host, int port, const Credentials& credentials)
+void MqttClient::connect(const std::string& host, int port, const Credentials &credentials, const mqtt_LWT &lwt )
 {
-
     osdev::components::mqtt::ParsedUri _endpoint = {
         { "scheme", "tcp" },
         { "user", credentials.username() },
@@ -135,10 +134,11 @@ void MqttClient::connect(const std::string&amp; host, int port, const Credentials&amp; c
         { "host", host },
         { "port", std::to_string(port) }
     };
-    this->connect(UriParser::toString(_endpoint));
+
+    this->connect( UriParser::toString( _endpoint ), lwt );
 }
-void MqttClient::connect(const std::string& _endpoint)
+void MqttClient::connect( const std::string &_endpoint, const mqtt_LWT &lwt )
 {
     LogInfo( "MqttClient", std::string( m_clientId + " - Request connect" ) );
@@ -159,7 +159,8 @@ void MqttClient::connect(const std::string&amp; _endpoint)
             }
         }
         m_endpoint = _endpoint;
-        if (!m_principalClient) {
+        if (!m_principalClient)
+        {
             std::string derivedClientId(generateUniqueClientId(m_clientId, 1));
             m_principalClient = std::make_unique<ClientPaho>(
                 m_endpoint,
@@ -169,7 +170,8 @@ void MqttClient::connect(const std::string&amp; _endpoint)
         }
         client = m_principalClient.get();
     }
-    client->connect(true);
+
+    client->connect( true, lwt );
 }
 void MqttClient::disconnect()