|
|
@@ -21,6 +21,8 @@ CMqttClientWorker::CMqttClientWorker()
|
|
|
CMqttClientWorker::~CMqttClientWorker()
|
|
|
{
|
|
|
m_is_closing = true;
|
|
|
+ m_is_running = false;
|
|
|
+ m_stopCv.notify_all();
|
|
|
m_async_client.reset();
|
|
|
}
|
|
|
|
|
|
@@ -38,6 +40,7 @@ void CMqttClientWorker::Stop()
|
|
|
{
|
|
|
m_is_closing = true;
|
|
|
m_is_running = false;
|
|
|
+ m_stopCv.notify_all();
|
|
|
}
|
|
|
|
|
|
void CMqttClientWorker::Run()
|
|
|
@@ -75,36 +78,42 @@ void CMqttClientWorker::Run()
|
|
|
return;
|
|
|
}
|
|
|
|
|
|
- while (m_is_running)
|
|
|
+ while (m_is_running && !m_is_closing)
|
|
|
{
|
|
|
- //一直循环,一直工作,等待接收消息
|
|
|
- Sleep(100);
|
|
|
+ std::unique_lock<std::mutex> lock(m_stopMutex);
|
|
|
+ m_stopCv.wait_for(lock, std::chrono::milliseconds(100), [this]()
|
|
|
+ {
|
|
|
+ return !m_is_running || m_is_closing;
|
|
|
+ });
|
|
|
}
|
|
|
|
|
|
+ CLewaimaiLog::OutputDebugMessage("准备退出");
|
|
|
+
|
|
|
//准备退出,禁止回调里继续重连/处理
|
|
|
m_is_closing = true;
|
|
|
+ m_stopCv.notify_all();
|
|
|
|
|
|
try
|
|
|
{
|
|
|
- LOG_INFO("Disconnecting from the MQTT server...");
|
|
|
+ CLewaimaiLog::OutputDebugMessage("Disconnecting from the MQTT server...");
|
|
|
|
|
|
auto client = m_async_client;
|
|
|
if (client && client->is_connected())
|
|
|
{
|
|
|
auto tok = client->disconnect();
|
|
|
|
|
|
- if (!tok->wait_for(std::chrono::seconds(2)))
|
|
|
+ if (!tok->wait_for(std::chrono::milliseconds(200)))
|
|
|
{
|
|
|
- LOG_INFO("MQTT disconnect timeout, wait for pending callbacks to finish.");
|
|
|
+ CLewaimaiLog::OutputDebugMessage("MQTT disconnect timeout, wait for pending callbacks to finish.");
|
|
|
}
|
|
|
else
|
|
|
{
|
|
|
- LOG_INFO("MQTT disconnected.");
|
|
|
+ CLewaimaiLog::OutputDebugMessage("MQTT disconnected.");
|
|
|
}
|
|
|
}
|
|
|
else
|
|
|
{
|
|
|
- LOG_INFO("MQTT client is already disconnected.");
|
|
|
+ CLewaimaiLog::OutputDebugMessage("MQTT client is already disconnected.");
|
|
|
}
|
|
|
}
|
|
|
catch (const mqtt::exception& exc)
|
|
|
@@ -112,7 +121,7 @@ void CMqttClientWorker::Run()
|
|
|
// is_connected() 与 disconnect() 之间连接仍可能被远端断开。
|
|
|
if (exc.get_return_code() == MQTTASYNC_DISCONNECTED)
|
|
|
{
|
|
|
- LOG_INFO("MQTT client was already disconnected.");
|
|
|
+ CLewaimaiLog::OutputDebugMessage("MQTT client was already disconnected.");
|
|
|
}
|
|
|
else
|
|
|
{
|
|
|
@@ -121,15 +130,18 @@ void CMqttClientWorker::Run()
|
|
|
std::to_string(exc.get_return_code()) +
|
|
|
"]: " + exc.get_message();
|
|
|
|
|
|
- LOG_INFO(msg.c_str());
|
|
|
CLewaimaiLog::OutputDebugMessage(msg.c_str());
|
|
|
}
|
|
|
}
|
|
|
|
|
|
+ CLewaimaiLog::OutputDebugMessage("MQTT client cleanup complete.");
|
|
|
+
|
|
|
//不要 delete,改成释放智能指针
|
|
|
m_async_client.reset();
|
|
|
|
|
|
- CAppEnv::GetInstance()->DelWorkerNum();
|
|
|
+ CLewaimaiLog::OutputDebugMessage("MQTT client worker thread exiting.");
|
|
|
+
|
|
|
+ CAppEnv::GetInstance()->DelWorkerNum(L"CMqttClientWorker::Run");
|
|
|
|
|
|
return;
|
|
|
}
|
|
|
@@ -144,7 +156,16 @@ void CMqttClientWorker::reconnect()
|
|
|
return;
|
|
|
|
|
|
//暂停10秒后重连
|
|
|
- std::this_thread::sleep_for(std::chrono::milliseconds(10000));
|
|
|
+ {
|
|
|
+ std::unique_lock<std::mutex> lock(m_stopMutex);
|
|
|
+ if (m_stopCv.wait_for(lock, std::chrono::milliseconds(10000), [this]()
|
|
|
+ {
|
|
|
+ return m_is_closing.load();
|
|
|
+ }))
|
|
|
+ {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ }
|
|
|
|
|
|
if (m_is_closing)
|
|
|
return;
|