Merge branch 'develop'

Author: hevake

examples/mqtt/conn/conn.cpp

@@ -42,7 +42,10 @@ int main(int argc, char **argv)
 
     mqtt::Client::Config conf;
     conf.auto_reconnect_enable = true;
-    conf.auto_reconnect_wait_sec = 5;
+    //conf.auto_reconnect_wait_sec = 5;
+    conf.auto_reconnect_wait_sec_gen_func = [](int fail_count) {
+        return 1 << std::min(fail_count, 4);
+    };
 #if 0
     conf.base.broker.domain = "cppmain.cpp";
     conf.base.broker.port = 1883;

modules/eventx/async.cpp

@@ -24,6 +24,9 @@
 #include <tbox/util/fs.h>
 #include <tbox/util/execute_cmd.h>
 
+#undef  MODULE_ID
+#define MODULE_ID "tbox.async"
+
 namespace tbox {
 namespace eventx {
 

modules/eventx/loop_thread.cpp

@@ -20,6 +20,9 @@
 #include "loop_thread.h"
 #include "loop_wdog.h"
 
+#undef  MODULE_ID
+#define MODULE_ID "tbox.loop_thread"
+
 namespace tbox {
 namespace eventx {
 

modules/eventx/loop_wdog.cpp

@@ -32,6 +32,9 @@
 #include <tbox/base/defines.h>
 #include <tbox/base/wrapped_recorder.h>
 
+#undef  MODULE_ID
+#define MODULE_ID "tbox.loop_wdog"
+
 namespace tbox {
 namespace eventx {
 

modules/eventx/thread_pool.cpp

@@ -38,6 +38,9 @@
 #include <tbox/base/wrapped_recorder.h>
 #include <tbox/event/loop.h>
 
+#undef  MODULE_ID
+#define MODULE_ID "tbox.thread_pool"
+
 namespace tbox {
 namespace eventx {
 

modules/eventx/timer_fd.cpp

@@ -30,6 +30,9 @@
 #include <tbox/event/loop.h>
 #include <tbox/event/fd_event.h>
 
+#undef  MODULE_ID
+#define MODULE_ID "tbox.timer_fd"
+
 namespace tbox {
 namespace eventx {
 

modules/eventx/timer_pool.cpp

@@ -25,6 +25,9 @@
 #include <tbox/event/loop.h>
 #include <tbox/event/timer_event.h>
 
+#undef  MODULE_ID
+#define MODULE_ID "tbox.timer_pool"
+
 namespace tbox {
 namespace eventx {
 

modules/eventx/work_thread.cpp

@@ -37,6 +37,9 @@
 #include <tbox/base/wrapped_recorder.h>
 #include <tbox/event/loop.h>
 
+#undef  MODULE_ID
+#define MODULE_ID "tbox.work_thread"
+
 namespace tbox {
 namespace eventx {
 

modules/mqtt/client.cpp

@@ -82,6 +82,8 @@ struct Client::Data {
     /// 在回调函数执行之前，检查一下这个 std::weak_ptr 是否过期就可以得知 this 指针是否有效。
     /// 进而避免继续访问到无效 this 指向的内容。
 
+    int connect_fail_count = 0; //!< 单次连接失败
+
     static int _instance_count;
 };
 
@@ -289,6 +291,7 @@ bool Client::start()
 
     CHECK_DELETE_RESET_OBJ(d_->sp_thread);
     auto is_alive = d_->alive_tag.get();  //! 原理见Q1
+    d_->connect_fail_count = 0;
 
     //! 由于 mosquitto_connect() 是阻塞函数，为了避免阻塞其它事件，特交给子线程去做
     d_->sp_thread = new thread(
@@ -609,13 +612,17 @@ void Client::onTcpConnectDone(int ret)
     CHECK_DELETE_RESET_OBJ(d_->sp_thread);
 
     if (ret == MOSQ_ERR_SUCCESS) {
+        d_->connect_fail_count = 0;
         LogDbg("connect success");
         enableSocketRead();
         enableSocketWriteIfNeed();
         updateStateTo(State::kTcpConnected);
 
     } else {
-        LogNotice("connect fail, rc:%d, %s", ret, mosquitto_strerror(ret));
+        ++d_->connect_fail_count;
+        LogNotice("connect fail, rc:%d, %s, fail_count:%d",
+                  ret, mosquitto_strerror(ret), d_->connect_fail_count);
+
         tryReconnect();
 
         ++d_->cb_level;
@@ -679,17 +686,24 @@ void Client::tryReconnect()
 {
     //! 如果开启了自动重连
     if (d_->config.auto_reconnect_enable) {
-        if (d_->config.auto_reconnect_wait_sec > 0) {
-            LogDbg("reconnect after %d sec", d_->config.auto_reconnect_wait_sec);
-            d_->reconnect_wait_remain_sec = d_->config.auto_reconnect_wait_sec;
+        auto wait_remain_sec = d_->config.auto_reconnect_wait_sec;
+
+        //! 如果设置了 auto_reconnect_wait_sec_gen_func 函数
+        //! 就则使用 auto_reconnect_wait_sec_gen_func 函数生成 wait_remain_sec
+        if (d_->config.auto_reconnect_wait_sec_gen_func)
+            wait_remain_sec = d_->config.auto_reconnect_wait_sec_gen_func(d_->connect_fail_count);
+
+        if (wait_remain_sec > 0) {
+            LogDbg("reconnect after %d sec", wait_remain_sec);
             updateStateTo(State::kReconnWaiting);
 
         } else {
             LogDbg("reconnect now");
-            d_->reconnect_wait_remain_sec = 0;
             updateStateTo(State::kConnecting);
         }
 
+        d_->reconnect_wait_remain_sec = wait_remain_sec;
+
     } else {  //! 如果不需要自动重连
         LogDbg("no need reconnect, end");
         updateStateTo(State::kEnd);

modules/mqtt/client.h

@@ -87,6 +87,8 @@ class Client {
 
         bool auto_reconnect_enable = true;   //! 是否自动重连
         int  auto_reconnect_wait_sec = 0;    //! 自动重连等待时长，秒
+        //! 根据失败次数动态生成自动重连等待时长，秒
+        std::function<int(int)> auto_reconnect_wait_sec_gen_func;
 
         bool isValid() const;
     };