添加取消订阅的逻辑
This commit is contained in:
parent
3279fcb859
commit
c6b3dedfdc
@ -1,12 +1,14 @@
|
|||||||
// mqtt_client.hpp
|
// mqtt_client.hpp
|
||||||
#pragma once
|
#pragma once
|
||||||
|
|
||||||
#include <string>
|
#include <mqtt/async_client.h>
|
||||||
|
|
||||||
|
#include <atomic>
|
||||||
#include <functional>
|
#include <functional>
|
||||||
#include <memory>
|
#include <memory>
|
||||||
#include <atomic>
|
|
||||||
#include <mutex>
|
#include <mutex>
|
||||||
#include <mqtt/async_client.h>
|
#include <string>
|
||||||
|
|
||||||
#include "app_config.hpp"
|
#include "app_config.hpp"
|
||||||
#include "logger.hpp"
|
#include "logger.hpp"
|
||||||
|
|
||||||
@ -29,6 +31,7 @@ public:
|
|||||||
void force_disconnect();
|
void force_disconnect();
|
||||||
void publish(const std::string& topic, const std::string& payload, int qos = -1);
|
void publish(const std::string& topic, const std::string& payload, int qos = -1);
|
||||||
void subscribe(const std::string& topic, int qos = -1);
|
void subscribe(const std::string& topic, int qos = -1);
|
||||||
|
void unsubscribe(const std::string& topic);
|
||||||
void switchServer(const MQTTConfig& newConfig);
|
void switchServer(const MQTTConfig& newConfig);
|
||||||
|
|
||||||
bool isConnected() const;
|
bool isConnected() const;
|
||||||
|
|||||||
@ -137,6 +137,21 @@ void MQTTClient::subscribe(const std::string& topic, int qos)
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
void MQTTClient::unsubscribe(const std::string& topic)
|
||||||
|
{
|
||||||
|
if (!connected_ || topic.empty()) return;
|
||||||
|
|
||||||
|
try
|
||||||
|
{
|
||||||
|
client_->unsubscribe(topic);
|
||||||
|
LOG_INFO("[MQTT] Unsubscribed from topic: " + topic);
|
||||||
|
}
|
||||||
|
catch (const mqtt::exception& e)
|
||||||
|
{
|
||||||
|
LOG_WARN(std::string("[MQTT] Unsubscribe failed: ") + e.what());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
void MQTTClient::switchServer(const MQTTConfig& newConfig)
|
void MQTTClient::switchServer(const MQTTConfig& newConfig)
|
||||||
{
|
{
|
||||||
std::lock_guard<std::mutex> lock(mutex_);
|
std::lock_guard<std::mutex> lock(mutex_);
|
||||||
|
|||||||
@ -25,15 +25,31 @@ static std::string g_last_vid;
|
|||||||
static bool try_activate_mqtt()
|
static bool try_activate_mqtt()
|
||||||
{
|
{
|
||||||
if (g_app_config.runtime_vid.empty()) return false;
|
if (g_app_config.runtime_vid.empty()) return false;
|
||||||
|
if (!mqtt_client || !mqtt_client->isConnected()) return false;
|
||||||
|
|
||||||
|
// 如果已激活且 VID 未变化,什么都不做
|
||||||
if (g_mqtt_activated.load() && g_last_vid == g_app_config.runtime_vid) return true;
|
if (g_mqtt_activated.load() && g_last_vid == g_app_config.runtime_vid) return true;
|
||||||
|
|
||||||
// VID 首次 or 发生变化
|
// ---------- 1. 如果是 VID 变化,先取消旧订阅 ----------
|
||||||
|
if (g_mqtt_activated.load())
|
||||||
|
{
|
||||||
|
LOG_INFO("[MQTT] VID changed, unsubscribe old topics: " + g_last_vid);
|
||||||
|
|
||||||
|
mqtt_client->unsubscribe(g_prev_topics.video_down);
|
||||||
|
mqtt_client->unsubscribe(g_prev_topics.record_query);
|
||||||
|
mqtt_client->unsubscribe(g_prev_topics.record_play);
|
||||||
|
mqtt_client->unsubscribe(g_prev_topics.vehicle_ctrl);
|
||||||
|
mqtt_client->unsubscribe(g_prev_topics.heartbeat_up);
|
||||||
|
}
|
||||||
|
|
||||||
|
// ---------- 2. 切换到新 VID ----------
|
||||||
g_app_config.mqtt.topics.fill_with_veh_id(g_app_config.runtime_vid);
|
g_app_config.mqtt.topics.fill_with_veh_id(g_app_config.runtime_vid);
|
||||||
|
g_prev_topics = g_app_config.mqtt.topics;
|
||||||
g_last_vid = g_app_config.runtime_vid;
|
g_last_vid = g_app_config.runtime_vid;
|
||||||
|
|
||||||
LOG_INFO("[MQTT] Activated with VID=" + g_last_vid);
|
LOG_INFO("[MQTT] Activated with VID=" + g_last_vid);
|
||||||
|
|
||||||
|
// ---------- 3. 订阅新 topic ----------
|
||||||
mqtt_client->subscribe(g_app_config.mqtt.topics.video_down);
|
mqtt_client->subscribe(g_app_config.mqtt.topics.video_down);
|
||||||
mqtt_client->subscribe(g_app_config.mqtt.topics.record_query);
|
mqtt_client->subscribe(g_app_config.mqtt.topics.record_query);
|
||||||
mqtt_client->subscribe(g_app_config.mqtt.topics.record_play);
|
mqtt_client->subscribe(g_app_config.mqtt.topics.record_play);
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user