Rework for subscription async

Peter M. Groen
1 parent 4b9da1a8
Showing 5 changed files with 39 additions and 9 deletions
examples/sub/main.cpp
examples/sub/subscriber.cpp
examples/sub/subscriber.h
src/clientpaho.cpp
src/mqttclient.cpp
@@ -76,7 +76,8 @@ int main( int argc, char* argv[] )
     {
         std::cout << "[OK]" << std::endl;
         std::cout << "Connecting to the test-broker : " << std::endl;
-        pSubscriber->connect( "localhost", 1883, "", "" );
+        pSubscriber->connect( "localhost", 1883, "", "", "test/subscriber/LWT", "Subscriber disconnected." );
+
         std::cout << "Subscribing to the test-topic....." << std::endl;
         pSubscriber->subscribe( "test/publisher/#" );
@@ -84,6 +85,7 @@ int main( int argc, char* argv[] )
         while( 1 )
         {
             sleepcp( 1, T_SECONDS );   // Sleep 1 Sec to give the scheduler the change to interfene.
+            std::cout << ".";
         }
     }
     else
@@ -24,11 +24,25 @@
 #include <iostream>
 Subscriber::Subscriber( const std::string &client_id )
-    : MqttSubscriberBase( client_id )
+    : m_mqtt_client( client_id )
 {
 }
+void Subscriber::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 ), osdev::components::mqtt::mqtt_LWT( lwt_topic, lwt_message ) );
+    std::cout << "Client state : " << m_mqtt_client.state() << std::endl;
+}
+
+void Subscriber::subscribe( const std::string &message_topic )
+{
+    m_mqtt_client.subscribe( message_topic, 1, [this](const osdev::components::mqtt::MqttMessage &message )
+    {
+        this->receive_data( message.topic(), message.payload() );
+    });
+}
+
 void Subscriber::receive_data( const std::string &message_topic, const std::string &message_payload )
 {
     std::cout << "[Subscriber::receive_data] - Received message : " << message_payload << " from topic : " << message_topic << std::endl;
@@ -22,19 +22,29 @@
 #pragma once
 // std
+#include <memory>
 #include <string>
 // mqtt-cpp
-#include "mqttsubscriberbase.h"
+#include "mqttclient.h"
+#include "compat-c++14.h"
-class Subscriber : public MqttSubscriberBase
+class Subscriber
 {
 public:
-    Subscriber( const std::string &client_id );
+    Subscriber(const std::string &client_id);
     virtual ~Subscriber() {}
+    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    subscribe( const std::string &message_topic );
+
 protected:
-    void receive_data( const std::string &message_topic, const std::string &message_payload );
+    void    receive_data( const std::string &message_topic, const std::string &message_payload );
+
+private:
+    osdev::components::mqtt::MqttClient m_mqtt_client;
 };
@@ -425,12 +425,14 @@ std::int32_t ClientPaho::subscribe( const std::string&amp; topic, int qos, const std
 {
     if( ConnectionStatus::Connected != m_connectionStatus )
     {
-        // MqttException, "Not connected"
+        LogDebug( "[ClientPaho::subscribe]", std::string( m_clientId + " - Client not connected..." ) );
     }
     if( !isValidTopic( topic ) )
     {
         // ("ClientPaho", "%1 - topic %2 is invalid", m_clientId, topic);
+        LogDebug( "[ClientPaho::subscribe]", std::string( m_clientId + " - Topic " + topic + " is invalid." ) );
+        return -1;
     }
     if( qos > 2 )
@@ -477,6 +479,7 @@ std::int32_t ClientPaho::subscribe( const std::string&amp; topic, int qos, const std
         if( isOverlappingInternal( topic, existingTopic ) )
         {
             // (OverlappingTopicException, "overlapping topic", existingTopic, topic);
+            LogDebug( "[ClientPaho::subscribe]", std::string( m_clientId + " - Overlapping topic : Existing Topic : " + existingTopic + " => New Topic : " + topic ) );
         }
         LogDebug( "[ClientPaho::subscribe]", std::string( m_clientId + " -adding subscription on topic " + topic + " to the pending subscriptions" ) );
@@ -284,7 +284,7 @@ Token MqttClient::subscribe(const std::string&amp; topic, int qos, const std::functi
     }
     if (!clientFound)
     {
-        client->connect(true);
+        client->connect( false );
     }
     return Token{ client->clientId(), client->subscribe(topic, qos, cb) };
 }
@@ -401,7 +401,8 @@ void MqttClient::connectionStatusChanged(const std::string&amp; id, ConnectionStatus
     }
     auto newState = determineState(connectionStates);
     bool resubscribe = (StateEnum::ConnectionFailure == m_serverState.state() && StateEnum::Good == newState);
-    if (resubscribe) {
+    if (resubscribe)
+    {
         {
             OSDEV_COMPONENTS_LOCKGUARD(m_internalMutex);
             m_activeTokens.clear();