generated from guanyuankai/bonus-edge-proxy
322 lines
10 KiB
C++
322 lines
10 KiB
C++
#include "config_manager.h"
|
||
|
||
#include <cstdio> // for std::rename, std::remove
|
||
#include <fstream>
|
||
|
||
ConfigManager& ConfigManager::getInstance() {
|
||
static ConfigManager instance;
|
||
return instance;
|
||
}
|
||
|
||
// 注册监听器
|
||
void ConfigManager::monitorKey(const std::string& key, KeyUpdateCallback cb) {
|
||
std::lock_guard<std::mutex> lock(m_callbackMutex);
|
||
m_key_callbacks[key].push_back(cb);
|
||
spdlog::debug("Registered observer for config key: '{}'", key);
|
||
}
|
||
|
||
// 核心逻辑：对比新旧配置，触发回调
|
||
void ConfigManager::checkDiffAndNotify(const json& old_config, const json& new_config) {
|
||
// 复制一份监听列表，避免在遍历时加锁时间过长，或者回调中再次注册导致死锁
|
||
std::map<std::string, std::vector<KeyUpdateCallback>> callbacks_snapshot;
|
||
{
|
||
std::lock_guard<std::mutex> lock(m_callbackMutex);
|
||
callbacks_snapshot = m_key_callbacks;
|
||
}
|
||
|
||
int trigger_count = 0;
|
||
|
||
for (const auto& [key, callbacks] : callbacks_snapshot) {
|
||
if (!new_config.contains(key))
|
||
continue;
|
||
|
||
// 获取新旧值 (如果旧配置没这个key，视为 null)
|
||
json old_val = old_config.contains(key) ? old_config[key] : json();
|
||
json new_val = new_config[key];
|
||
|
||
if (old_val != new_val) {
|
||
spdlog::info("Config Hot-Reload: Key '{}' changed. Triggering {} callbacks.", key,
|
||
callbacks.size());
|
||
|
||
for (const auto& cb : callbacks) {
|
||
try {
|
||
cb(new_val); // 执行回调
|
||
trigger_count++;
|
||
} catch (const std::exception& e) {
|
||
spdlog::error("Exception inside config callback for key '{}': {}", key,
|
||
e.what());
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
if (trigger_count > 0) {
|
||
spdlog::info("Config reload finished. {} callbacks triggered.", trigger_count);
|
||
}
|
||
}
|
||
|
||
json ConfigManager::createDefaultConfig() {
|
||
return json{
|
||
{"device_id", "default-edge-proxy-01"},
|
||
{"config_base_path", "/app/config/"},
|
||
{"mqtt_broker", "tcp://localhost:1883"},
|
||
{"mqtt_client_id_prefix", "edge-proxy-"},
|
||
{"data_storage_db_path", "edge_proxy_data.db"},
|
||
{"data_cache_db_path", "edge_data_cache.db"},
|
||
{"tcp_server_ports", {12345}},
|
||
{"web_server_port", 8080},
|
||
{"log_level", "debug"},
|
||
{"alarm_rules_path", "alarms.json"},
|
||
{"piper_executable_path", "/usr/bin/piper"},
|
||
{"piper_model_path", "/app/models/model.onnx"},
|
||
{"video_config_path", "/app/config/video_config.json"},
|
||
};
|
||
}
|
||
|
||
// 原子写入实现
|
||
bool ConfigManager::save_unlocked() {
|
||
std::string tmpPath = m_configFilePath + ".tmp";
|
||
|
||
// 1. 写入临时文件
|
||
{
|
||
std::ofstream ofs(tmpPath);
|
||
if (!ofs.is_open()) {
|
||
spdlog::error("Failed to open temp config file '{}' for writing.", tmpPath);
|
||
return false;
|
||
}
|
||
try {
|
||
ofs << m_config_json.dump(4);
|
||
ofs.flush(); // 确保缓冲区刷入内核
|
||
// 在某些关键系统上，可能还需要 fsync
|
||
} catch (const json::exception& e) {
|
||
spdlog::error("Failed to serialize config. Error: {}", e.what());
|
||
return false;
|
||
}
|
||
} // ofs 关闭
|
||
|
||
// 2. 原子重命名 (Atomic Rename)
|
||
// 如果重命名失败，原文件不会损坏
|
||
if (std::rename(tmpPath.c_str(), m_configFilePath.c_str()) != 0) {
|
||
spdlog::error("Failed to rename temp config file to '{}'. Errno: {}", m_configFilePath,
|
||
errno);
|
||
return false;
|
||
}
|
||
|
||
return true;
|
||
}
|
||
|
||
bool ConfigManager::load(const std::string& configFilePath) {
|
||
// 准备两个容器，用于在锁外进行对比
|
||
json old_config_copy;
|
||
json new_config_copy;
|
||
|
||
{
|
||
std::unique_lock<std::shared_mutex> lock(m_mutex); // 获取写锁
|
||
m_configFilePath = configFilePath;
|
||
|
||
// 备份当前配置（如果是第一次加载，这里是空的）
|
||
old_config_copy = m_config_json;
|
||
|
||
std::ifstream ifs(m_configFilePath);
|
||
if (!ifs.is_open()) {
|
||
spdlog::warn("Config file '{}' not found. Creating default.", m_configFilePath);
|
||
m_config_json = createDefaultConfig();
|
||
// 保存默认文件
|
||
if (!save_unlocked()) {
|
||
spdlog::error("Failed to create default config file.");
|
||
return false;
|
||
}
|
||
// 第一次加载，不需要触发“变更”回调
|
||
return true;
|
||
}
|
||
|
||
try {
|
||
json loaded_json;
|
||
ifs >> loaded_json;
|
||
|
||
// 合并默认值
|
||
json defaults = createDefaultConfig();
|
||
defaults.merge_patch(loaded_json);
|
||
m_config_json = defaults;
|
||
|
||
// 复制新配置用于后续对比
|
||
new_config_copy = m_config_json;
|
||
|
||
spdlog::info("Config loaded from '{}'. DeviceID: {}", m_configFilePath,
|
||
m_config_json.value("device_id", "N/A"));
|
||
|
||
// 可选：如果缺少key，回写文件
|
||
save_unlocked();
|
||
|
||
} catch (const json::exception& e) {
|
||
spdlog::error("Failed to parse config file. Error: {}. Using defaults.", e.what());
|
||
m_config_json = createDefaultConfig();
|
||
return false;
|
||
}
|
||
} // !!! 写锁在这里释放 !!!
|
||
|
||
// 锁释放后，执行回调通知
|
||
// 这样做非常安全，不会因为回调里调用了 get() 而导致死锁
|
||
checkDiffAndNotify(old_config_copy, new_config_copy);
|
||
|
||
return true;
|
||
}
|
||
|
||
bool ConfigManager::save() {
|
||
std::unique_lock<std::shared_mutex> lock(m_mutex);
|
||
return save_unlocked();
|
||
}
|
||
|
||
std::string ConfigManager::getConfigFilePath() const {
|
||
std::shared_lock<std::shared_mutex> lock(m_mutex);
|
||
return m_configFilePath;
|
||
}
|
||
|
||
std::string ConfigManager::getDeviceID() {
|
||
return get<std::string>("device_id", "default-edge-proxy-01");
|
||
}
|
||
std::string ConfigManager::getConfigBasePath() {
|
||
return get<std::string>("config_base_path", "/app/config/");
|
||
}
|
||
std::string ConfigManager::getMqttBroker() {
|
||
return get<std::string>("mqtt_broker", "tcp://localhost:1883");
|
||
}
|
||
std::string ConfigManager::getMqttClientID() {
|
||
return get<std::string>("mqtt_client_id_prefix", "edge-proxy-") + getDeviceID();
|
||
}
|
||
std::string ConfigManager::getDataStorageDbPath() {
|
||
return getConfigBasePath() + get<std::string>("data_storage_db_path", "edge_proxy_data.db");
|
||
}
|
||
std::string ConfigManager::getDataCacheDbPath() {
|
||
return getConfigBasePath() + get<std::string>("data_cache_db_path", "edge_data_cache.db");
|
||
}
|
||
std::string ConfigManager::getDevicesConfigPath() {
|
||
return getConfigBasePath() + "devices.json";
|
||
}
|
||
int ConfigManager::getWebServerPort() {
|
||
return get<int>("web_server_port", 8080);
|
||
}
|
||
std::vector<uint16_t> ConfigManager::getTcpServerPorts() {
|
||
return get<std::vector<uint16_t>>("tcp_server_ports", {12345});
|
||
}
|
||
std::string ConfigManager::getLogLevel() {
|
||
return get<std::string>("log_level", "debug");
|
||
}
|
||
std::string ConfigManager::getAlarmRulesPath() {
|
||
return getConfigBasePath() + get<std::string>("alarm_rules_path", "alarms.json");
|
||
}
|
||
std::string ConfigManager::getPiperExecutablePath() {
|
||
return get<std::string>("piper_executable_path", "/usr/bin/piper");
|
||
}
|
||
std::string ConfigManager::getPiperModelPath() {
|
||
return get<std::string>("piper_model_path", "/app/models/model.onnx");
|
||
}
|
||
std::string ConfigManager::getVideoConfigPath() {
|
||
return getConfigBasePath() + get<std::string>("video_config_path", "video_config.json");
|
||
}
|
||
|
||
std::string ConfigManager::getDbUser() {
|
||
return get<std::string>("db_user", "forlinx");
|
||
}
|
||
std::string ConfigManager::getDbPwd() {
|
||
return get<std::string>("db_pwd", "forlinx");
|
||
}
|
||
std::string ConfigManager::getDbHost() {
|
||
return get<std::string>("db_host", "127.0.0.1");
|
||
}
|
||
std::string ConfigManager::getDbName() {
|
||
return get<std::string>("db_database", "smart-car-dev");
|
||
}
|
||
|
||
TripwireConfig ConfigManager::getTripwireConfig() {
|
||
TripwireConfig config;
|
||
|
||
std::shared_lock<std::shared_mutex> lock(m_mutex);
|
||
|
||
if (!m_config_json.contains("tripwire") || !m_config_json["tripwire"].is_object()) {
|
||
spdlog::warn("Config key 'tripwire' not found or not an object. Using defaults.");
|
||
return config;
|
||
}
|
||
|
||
try {
|
||
const auto& j = m_config_json.at("tripwire");
|
||
|
||
// 注意：JSON 中是 "enable"，结构体中是 "enabled"
|
||
config.enabled = j.value("enable", false);
|
||
config.name = j.value("name", "Unnamed_Line");
|
||
|
||
// 5. 解析嵌套的 line 坐标
|
||
if (j.contains("line") && j["line"].is_object()) {
|
||
const auto& line = j.at("line");
|
||
|
||
// 解析 p1 (起点)
|
||
if (line.contains("p1")) {
|
||
config.p1_norm.x = line["p1"].value("x", 0.0f);
|
||
config.p1_norm.y = line["p1"].value("y", 0.0f);
|
||
}
|
||
|
||
// 解析 p2 (终点)
|
||
if (line.contains("p2")) {
|
||
config.p2_norm.x = line["p2"].value("x", 1.0f);
|
||
config.p2_norm.y = line["p2"].value("y", 1.0f);
|
||
}
|
||
}
|
||
} catch (const json::exception& e) {
|
||
spdlog::error("Failed to parse TripwireConfig: {}", e.what());
|
||
}
|
||
|
||
return config;
|
||
}
|
||
bool ConfigManager::updateTripwireLine(float p1_x, float p1_y, float p2_x, float p2_y) {
|
||
json new_tripwire_value; // 用于存储更新后的片段
|
||
|
||
// 1. 更新数据并保存 (使用作用域限制锁的范围)
|
||
{
|
||
std::unique_lock<std::shared_mutex> lock(m_mutex); // 获取写锁
|
||
|
||
if (!m_config_json.contains("tripwire") || !m_config_json["tripwire"].is_object()) {
|
||
m_config_json["tripwire"] = json::object();
|
||
}
|
||
|
||
if (!m_config_json["tripwire"].contains("line") || !m_config_json["tripwire"]["line"].is_object()) {
|
||
m_config_json["tripwire"]["line"] = json::object();
|
||
}
|
||
|
||
m_config_json["tripwire"]["line"]["p1"] = { {"x", p1_x}, {"y", p1_y} };
|
||
m_config_json["tripwire"]["line"]["p2"] = { {"x", p2_x}, {"y", p2_y} };
|
||
|
||
// 捕获最新的 tripwire 值，用于发送给回调
|
||
new_tripwire_value = m_config_json["tripwire"];
|
||
|
||
spdlog::info("Updating Tripwire: P1({:.2f}, {:.2f}), P2({:.2f}, {:.2f})",
|
||
p1_x, p1_y, p2_x, p2_y);
|
||
|
||
if (!save_unlocked()) {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
// 2. === [修复核心] 手动触发观察者回调 ===
|
||
std::vector<KeyUpdateCallback> callbacks;
|
||
{
|
||
std::lock_guard<std::mutex> cb_lock(m_callbackMutex);
|
||
if (m_key_callbacks.count("tripwire")) {
|
||
callbacks = m_key_callbacks["tripwire"];
|
||
}
|
||
}
|
||
|
||
// 执行回调
|
||
int trigger_count = 0;
|
||
for (const auto& cb : callbacks) {
|
||
try {
|
||
cb(new_tripwire_value); // 这里的 new_tripwire_value 就是传给 VideoPipeline 的新配置
|
||
trigger_count++;
|
||
} catch (const std::exception& e) {
|
||
spdlog::error("Exception inside manual config update callback: {}", e.what());
|
||
}
|
||
}
|
||
|
||
spdlog::info("Manual update triggered {} callbacks.", trigger_count);
|
||
|
||
return true;
|
||
} |