|
|
@@ -20,11 +20,13 @@ CMqttClientWorker::CMqttClientWorker()
|
|
|
|
|
|
CMqttClientWorker::~CMqttClientWorker()
|
|
|
{
|
|
|
-
|
|
|
+ m_is_closing = true;
|
|
|
+ m_async_client.reset();
|
|
|
}
|
|
|
|
|
|
void CMqttClientWorker::Start()
|
|
|
{
|
|
|
+ m_is_closing = false;
|
|
|
m_is_running = true;
|
|
|
|
|
|
//处理Mqtt消息接收
|
|
|
@@ -34,13 +36,14 @@ void CMqttClientWorker::Start()
|
|
|
|
|
|
void CMqttClientWorker::Stop()
|
|
|
{
|
|
|
+ m_is_closing = true;
|
|
|
m_is_running = false;
|
|
|
}
|
|
|
|
|
|
void CMqttClientWorker::Run()
|
|
|
{
|
|
|
m_client_id = "GID_ZHIPUZI_WINDOWS_POS@@@" + CSetting::GetInstance()->getUsername();
|
|
|
- m_async_client = new mqtt::async_client(m_server_address, m_client_id);
|
|
|
+ m_async_client = std::make_shared<mqtt::async_client>(m_server_address, m_client_id);
|
|
|
|
|
|
// A subscriber often wants the server to remember its messages when its
|
|
|
// disconnected. In that case, it needs a unique ClientID and a
|
|
|
@@ -78,22 +81,34 @@ void CMqttClientWorker::Run()
|
|
|
Sleep(100);
|
|
|
}
|
|
|
|
|
|
- //代码走到这里,说明退出登录了,要暂停推送了
|
|
|
+ //准备退出,禁止回调里继续重连/处理
|
|
|
+ m_is_closing = true;
|
|
|
|
|
|
try
|
|
|
{
|
|
|
LOG_INFO("Disconnecting from the MQTT server...");
|
|
|
- m_async_client->disconnect()->wait();
|
|
|
- LOG_INFO("OK...");
|
|
|
+ auto client = m_async_client;
|
|
|
+ if (client)
|
|
|
+ {
|
|
|
+ auto tok = client->disconnect();
|
|
|
+ if (!tok->wait_for(std::chrono::seconds(2)))
|
|
|
+ {
|
|
|
+ LOG_INFO("MQTT disconnect timeout, wait for pending callbacks to finish.");
|
|
|
+ }
|
|
|
+ else
|
|
|
+ {
|
|
|
+ LOG_INFO("OK...");
|
|
|
+ }
|
|
|
+ }
|
|
|
}
|
|
|
- catch (const mqtt::exception & exc)
|
|
|
+ catch (const mqtt::exception& exc)
|
|
|
{
|
|
|
LOG_INFO(("disconnect error, exc:" + exc.get_message()).c_str());
|
|
|
CLewaimaiLog::OutputDebugMessage(("disconnect error, exc:" + exc.get_message()).c_str());
|
|
|
}
|
|
|
|
|
|
- //销毁客户端
|
|
|
- delete m_async_client;
|
|
|
+ //不要 delete,改成释放智能指针
|
|
|
+ m_async_client.reset();
|
|
|
|
|
|
CAppEnv::GetInstance()->DelWorkerNum();
|
|
|
|
|
|
@@ -102,8 +117,19 @@ void CMqttClientWorker::Run()
|
|
|
|
|
|
void CMqttClientWorker::reconnect()
|
|
|
{
|
|
|
+ if (m_is_closing)
|
|
|
+ return;
|
|
|
+
|
|
|
+ auto client = m_async_client;
|
|
|
+ if (!client)
|
|
|
+ return;
|
|
|
+
|
|
|
//暂停10秒后重连
|
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(10000));
|
|
|
+
|
|
|
+ if (m_is_closing)
|
|
|
+ return;
|
|
|
+
|
|
|
try
|
|
|
{
|
|
|
m_async_client->connect(m_connOpts, nullptr, *this);
|
|
|
@@ -120,6 +146,9 @@ void CMqttClientWorker::reconnect()
|
|
|
// Re-connection failure
|
|
|
void CMqttClientWorker::on_failure(const mqtt::token & tok)
|
|
|
{
|
|
|
+ if (m_is_closing)
|
|
|
+ return;
|
|
|
+
|
|
|
if (m_is_mqtt_connected == false)
|
|
|
{
|
|
|
//连接失败了
|
|
|
@@ -146,6 +175,9 @@ void CMqttClientWorker::on_failure(const mqtt::token & tok)
|
|
|
|
|
|
void CMqttClientWorker::on_success(const mqtt::token & tok)
|
|
|
{
|
|
|
+ if (m_is_closing)
|
|
|
+ return;
|
|
|
+
|
|
|
if (m_is_mqtt_connected == false)
|
|
|
{
|
|
|
//连接成功了,再connected里面处理
|
|
|
@@ -170,17 +202,27 @@ void CMqttClientWorker::on_success(const mqtt::token & tok)
|
|
|
|
|
|
void CMqttClientWorker::connected(const std::string & cause)
|
|
|
{
|
|
|
+ if (m_is_closing)
|
|
|
+ return;
|
|
|
+
|
|
|
m_is_mqtt_connected = true;
|
|
|
|
|
|
std::string info = "Connection success, nSubscribing to topic " + m_topic + " for client " + m_client_id + " using QoS" + std::to_string(MQTT_QOS);
|
|
|
LOG_INFO(info.c_str());
|
|
|
|
|
|
//不管是第一次连接,还是重连,都订阅一次
|
|
|
- m_async_client->subscribe(m_topic, MQTT_QOS, nullptr, *this);
|
|
|
+ auto client = m_async_client;
|
|
|
+ if (client)
|
|
|
+ {
|
|
|
+ client->subscribe(m_topic, MQTT_QOS, nullptr, *this);
|
|
|
+ }
|
|
|
}
|
|
|
|
|
|
void CMqttClientWorker::connection_lost(const std::string & cause)
|
|
|
{
|
|
|
+ if (m_is_closing)
|
|
|
+ return;
|
|
|
+
|
|
|
m_is_mqtt_connected = false;
|
|
|
|
|
|
std::string info = "Connection lost";
|
|
|
@@ -200,6 +242,9 @@ void CMqttClientWorker::connection_lost(const std::string & cause)
|
|
|
// Callback for when a message arrives.
|
|
|
void CMqttClientWorker::message_arrived(mqtt::const_message_ptr msg)
|
|
|
{
|
|
|
+ if (m_is_closing)
|
|
|
+ return;
|
|
|
+
|
|
|
std::string info = "Message arrived, ttopic: '" + msg->get_topic() + "', payload: '" + msg->to_string();
|
|
|
|
|
|
LOG_INFO(info.c_str());
|