Fix publishing

Peter M. Groen
1 parent 3fef3f83
Showing 3 changed files with 19 additions and 10 deletions
examples/pub/main.cpp
examples/sub/main.cpp
src/mqttclient.cpp
@@ -94,7 +94,7 @@ int main( int argc, char* argv[] )
                 }
                 messageNumber++;
             }
-            sleepcp( 1, T_SECONDS );
+            sleepcp(100, T_MILLI);
         }
     }
     else
@@ -91,7 +91,6 @@ int main( int argc, char* argv[] )
         while (loop_counter < MAX_LOOP_COUNT)
         {
             sleepcp( 1, T_SECONDS );   // Sleep 1 Sec to give the scheduler the change to interfene.
-            std::cout << ".";
             loop_counter++;
         }
@@ -101,11 +100,20 @@ int main( int argc, char* argv[] )
         pSubscriber->unsubscribe( "test/publisher/TestPublisher_0" );
         pSubscriber->unsubscribe( "test/publisher/TestPublisher_1" );
         pSubscriber->unsubscribe( "test/publisher/TestPublisher_2" );
+        pSubscriber->unsubscribe( "test/publisher/TestPublisher_3" );
+
+        // Start a loop to give the subscriber the possibility to do its work.
+        while (loop_counter < MAX_LOOP_COUNT)
+        {
+            sleepcp( 1, T_SECONDS );   // Sleep 1 Sec to give the scheduler the change to interfene.
+            loop_counter++;
+        }
+
+        pSubscriber->subscribe("test/publisher/#");
         while (1)
         {
-            sleepcp(1, T_MILLI);   // Sleep 1 Sec to give the scheduler the change to interfene.
-            std::cout << ".";
+            sleepcp(1, T_MILLI);   // Sleep 1 mSec to give the scheduler the change to interfene.
         }
     }
@@ -431,8 +431,8 @@ void MqttClient::connectionStatusChanged(const std::string&amp; id, ConnectionStatus
     }
     auto newState = determineState(connectionStates);
-    LogDebug( "[MqttClient::connectionStatusChanged]", std::string("Old state : " + std::to_string(static_cast<int>(m_serverState.state())));
-    LogDebug( "[MqttClient::connectionStatusChanged]", std::string("New state : " + std::to_string(static_cast<int>(newState)));
+    LogDebug( "[MqttClient::connectionStatusChanged]", std::string("Old state : " + std::to_string(static_cast<int>(m_serverState.state()))));
+    LogDebug( "[MqttClient::connectionStatusChanged]", std::string("New state : " + std::to_string(static_cast<int>(newState))));
     bool resubscribe = (StateEnum::ConnectionFailure == m_serverState.state() && StateEnum::Good == newState);
     // First activate pending subscriptions
     {
@@ -472,9 +472,10 @@ void MqttClient::connectionStatusChanged(const std::string&amp; 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)
-        {
+    this->pushEvent([this, resubscribe, clients, principalClient, newState]() 
+    {
+        // 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);
@@ -502,7 +503,7 @@ void MqttClient::connectionStatusChanged(const std::string&amp; id, ConnectionStatus
                     LogError( "[MqttClient::connectionStatusChanged]", std::string( m_clientId + " - publishPending on wrapped client " + principalClient->clientId() + " => FAILED " + e.what() ) );
                 }
             }
-        }
+        // }
         m_serverState.emitStateChanged(newState);
     });
 }