#571 Sync reconnect example crashes on first reconnect

Author: fpagliughi

src/async_client.cpp

@@ -92,7 +92,11 @@ async_client::~async_client() { MQTTAsync_destroy(&cli_); }
 // 'context' should be the address of the async_client object that
 // registered the callback.
 
-// Callback for MQTTAsync_setConnected()
+// Callback from the C library for when the client is initially
+// connected or reconnected.
+//
+// Installed via MQTTAsync_setConnected()
+//
 // This is installed with the normal callbacks and with a call to
 // reconnect() to indicate that it succeeded. It signals the client's
 // connect token then calls any registered callbacks.
@@ -120,14 +124,23 @@ void async_client::on_connected(void* context, char* cause)
         if (connHandler)
             connHandler(cause_str);
 
-        if (que)
-            que->put(connected_event{cause_str});
+        if (que) {
+            try {
+                que->put(connected_event{cause_str});
+            }
+            catch (const queue_closed&) {
+            }
+        }
     }
 }
 
-// Callback for when the connection is lost.
+// Callback for when the connection is (unecpectedly) lost. This
+// typically happens if the network goes down or if the server drops the
+// connection due to a protocol violation.
+//
 // This is called from the MQTTAsync_connectionLost registered via
 // MQTTAsync_setCallbacks().
+//
 // It calls the registered handlers then, if there's a consumer queue, it
 // places a null pointer in the queue to alert the consumer to a closed
 // connection.
@@ -151,8 +164,13 @@ void async_client::on_connection_lost(void* context, char* cause)
         if (connLostHandler)
             connLostHandler(cause_str);
 
-        if (que)
-            que->put(connection_lost_event{cause_str});
+        if (que) {
+            try {
+                que->put(connection_lost_event{cause_str});
+            }
+            catch (const queue_closed&) {
+            }
+        }
     }
 }
 
@@ -176,8 +194,13 @@ void async_client::on_disconnected(
         if (disconnectedHandler)
             disconnectedHandler(props, ReasonCode(reasonCode));
 
-        if (que)
-            que->put(disconnected_event{std::move(props), ReasonCode(reasonCode)});
+        if (que) {
+            try {
+                que->put(disconnected_event{std::move(props), ReasonCode(reasonCode)});
+            }
+            catch (const queue_closed&) {
+            }
+        }
     }
 }
 
@@ -208,8 +231,13 @@ int async_client::on_message_arrived(
         if (cb)
             cb->message_arrived(m);
 
-        if (que)
-            que->put(m);
+        if (que) {
+            try {
+                que->put(m);
+            }
+            catch (const queue_closed&) {
+            }
+        }
     }
 
     MQTTAsync_freeMessage(&msg);
@@ -854,6 +882,10 @@ token_ptr async_client::unsubscribe(
 
 void async_client::start_consuming()
 {
+    // Don't do anything if the consumer queue is already up.
+    if (que_ && !que_->closed())
+        return;
+
     // Make sure callbacks don't happen while we update the que, etc
     disable_callbacks();
 

src/client.cpp

@@ -170,14 +170,12 @@ unsubscribe_response client::unsubscribe(
 
 void client::disconnect()
 {
-    cli_.stop_consuming();
     if (!cli_.disconnect()->wait_for(timeout_))
         throw timeout_error();
 }
 
 void client::disconnect(int timeoutMS)
 {
-    cli_.stop_consuming();
     if (!cli_.disconnect(timeoutMS)->wait_for(timeout_))
         throw timeout_error();
 }

test/unit/test_async_client.cpp

@@ -578,7 +578,8 @@ TEST_CASE("async_client publish 5 args", "[client]")
 
     const void* payload{PAYLOAD.data()};
     const size_t payload_size{PAYLOAD.size()};
-    delivery_token_ptr token_pub{cli.publish(TOPIC, payload, payload_size, GOOD_QOS, RETAINED)
+    delivery_token_ptr token_pub{
+        cli.publish(TOPIC, payload, payload_size, GOOD_QOS, RETAINED)
     };
     REQUIRE(token_pub);
     token_pub->wait_for(TIMEOUT);
@@ -737,7 +738,8 @@ TEST_CASE("async_client subscribe many topics 2 args_single", "[client]")
     cli.connect()->wait();
     REQUIRE(cli.is_connected());
 
-    mqtt::const_string_collection_ptr TOPIC_1_COLL{mqtt::string_collection::create({"TOPIC0"})
+    mqtt::const_string_collection_ptr TOPIC_1_COLL{
+        mqtt::string_collection::create({"TOPIC0"})
     };
     iasync_client::qos_collection GOOD_QOS_1_COLL{0};
     try {

test/unit/test_async_client_v3.cpp

@@ -534,7 +534,8 @@ class async_client_v3_test : public CppUnit::TestFixture
 
         // NOTE: Only tokens for messages with QOS=1 and QOS=2 are kept. That's
         // why the vector's size does not account for QOS=0 message tokens
-        std::vector<mqtt::delivery_token_ptr> tokens_pending{cli.get_pending_delivery_tokens()
+        std::vector<mqtt::delivery_token_ptr> tokens_pending{
+            cli.get_pending_delivery_tokens()
         };
         CPPUNIT_ASSERT_EQUAL(2, static_cast<int>(tokens_pending.size()));
 
@@ -831,7 +832,8 @@ class async_client_v3_test : public CppUnit::TestFixture
         CPPUNIT_ASSERT(cli.is_connected());
 
         mqtt::test::dummy_action_listener listener;
-        mqtt::token_ptr token_sub{cli.subscribe(TOPIC_COLL, GOOD_QOS_COLL, &CONTEXT, listener)
+        mqtt::token_ptr token_sub{
+            cli.subscribe(TOPIC_COLL, GOOD_QOS_COLL, &CONTEXT, listener)
         };
         CPPUNIT_ASSERT(token_sub);
         token_sub->wait_for(TIMEOUT);

test/unit/test_create_options.cpp

@@ -65,9 +65,10 @@ TEST_CASE("create_options_builder default ctor", "[options]")
 
 TEST_CASE("create_options_builder sets", "[options]")
 {
-    const auto opts =
-        create_options_builder().send_while_disconnected().delete_oldest_messages().finalize(
-        );
+    const auto opts = create_options_builder()
+                          .send_while_disconnected()
+                          .delete_oldest_messages()
+                          .finalize();
 
     REQUIRE(opts.get_send_while_disconnected());
     REQUIRE(opts.get_delete_oldest_messages());