From 35231baae7b3f4371a68fca70d2b0b37353c730a Mon Sep 17 00:00:00 2001 From: lnk Date: Tue, 30 Sep 2025 16:17:04 +0800 Subject: [PATCH] modify ledger update and rtdata --- LFtid1056/cloudfront/code/cfg_parser.cpp | 193 +++++++++-------------- LFtid1056/cloudfront/code/interface.h | 72 ++++++++- LFtid1056/cloudfront/code/log4.cpp | 4 - LFtid1056/cloudfront/code/log4.h | 4 +- LFtid1056/cloudfront/code/main.cpp | 2 +- LFtid1056/cloudfront/code/rocketmq.cpp | 117 ++++++++++---- 6 files changed, 233 insertions(+), 159 deletions(-) diff --git a/LFtid1056/cloudfront/code/cfg_parser.cpp b/LFtid1056/cloudfront/code/cfg_parser.cpp index 77c6e86..b7520c7 100644 --- a/LFtid1056/cloudfront/code/cfg_parser.cpp +++ b/LFtid1056/cloudfront/code/cfg_parser.cpp @@ -72,6 +72,10 @@ extern std::map xmlinfo_list2;//保存所有型号角形 extern time_t ConvertToTimestamp(const tagTime& time); //////////////////////////////////////////////////////////////////////////////////////////////////// +//实时数据时间记录 +std::mutex g_last_ts_mtx; +std::unordered_map g_last_ts_by_devid; + static std::mutex g_filemenu_cache_mtx; std::map> g_filemenu_cache; @@ -1152,7 +1156,7 @@ int recall_json_handle_from_mq(const std::string& body) return 10004; } std::string guid = messageBody["guid"].get(); - send_reply_to_queue(guid, "1", "收到补招指令"); + send_reply_to_queue(guid, static_cast(ResponseCode::OK), "收到补招指令"); // 提取 data 数组 if (!messageBody.contains("data") || !messageBody["data"].is_array()) { @@ -1816,7 +1820,7 @@ void process_ledger_update(trigger_update_xml_t& ledger_update_xml) } if (terminal_devlist.size() >= static_cast(IED_COUNT)) { - send_reply_to_queue(new_dev.guid, "2", + send_reply_to_queue(new_dev.guid, static_cast(ResponseCode::BAD_REQUEST), "终端 id: " + new_dev.terminal_id + " 台账更新失败，配置台账数量已满"); ++it; continue; @@ -1824,27 +1828,19 @@ void process_ledger_update(trigger_update_xml_t& ledger_update_xml) terminal_dev target_dev; if (update_one_terminal_ledger(new_dev, target_dev) != 0) { - send_reply_to_queue(new_dev.guid, "2", + send_reply_to_queue(new_dev.guid, static_cast(ResponseCode::BAD_REQUEST), "终端 id: " + new_dev.terminal_id + " 台账更新失败，无法写入台账"); ++it; continue; } - /*if (parse_model_cfg_web_one(target_dev.dev_type).empty()) { - send_reply_to_queue(new_dev.guid, "2", - "终端 id: " + new_dev.terminal_id + " 台账更新失败，未找到装置型号"); - ++it; - continue; - }*/ - - //Set_xml_nodeinfo_one(target_dev.dev_type); init_loggers_bydevid(target_dev.terminal_id); terminal_devlist.push_back(target_dev); DeviceInfo device = make_device_from_terminal(target_dev); ClientManager::instance().add_device(device); - send_reply_to_queue(new_dev.guid, "2", + send_reply_to_queue(new_dev.guid, static_cast(ResponseCode::OK), "终端 id: " + new_dev.terminal_id + " 台账添加成功"); it = ledger_update_xml.new_updates.erase(it); @@ -1860,27 +1856,20 @@ void process_ledger_update(trigger_update_xml_t& ledger_update_xml) if (it != terminal_devlist.end()) { erase_one_terminals_by_id(mod_dev.terminal_id); if (update_one_terminal_ledger(mod_dev, *it) != 0) { - send_reply_to_queue(mod_dev.guid, "2", + send_reply_to_queue(mod_dev.guid, static_cast(ResponseCode::BAD_REQUEST), "终端 id: " + mod_dev.terminal_id + " 台账更新失败，写入失败"); continue; } - /*if (parse_model_cfg_web_one(it->dev_type).empty()) { - send_reply_to_queue(mod_dev.guid, "2", - "终端 id: " + mod_dev.terminal_id + " 台账更新失败，未找到装置型号"); - continue; - }*/ - - //Set_xml_nodeinfo_one(it->dev_type); init_loggers_bydevid(mod_dev.terminal_id); - DeviceInfo device = make_device_from_terminal(target_dev); + DeviceInfo device = make_device_from_terminal(*it); ClientManager::instance().add_device(device); - send_reply_to_queue(mod_dev.guid, "2", + send_reply_to_queue(mod_dev.guid, static_cast(ResponseCode::OK), "终端 id: " + mod_dev.terminal_id + " 台账修改成功"); } else { - send_reply_to_queue(mod_dev.guid, "2", + send_reply_to_queue(mod_dev.guid, static_cast(ResponseCode::NOT_FOUND), "终端 id: " + mod_dev.terminal_id + " 台账修改失败，未找到终端"); } } @@ -1895,10 +1884,10 @@ void process_ledger_update(trigger_update_xml_t& ledger_update_xml) if (it != terminal_devlist.end()) { erase_one_terminals_by_id(del_dev.terminal_id); ClientManager::instance().remove_device(del_dev.terminal_id); - send_reply_to_queue(del_dev.guid, "2", + send_reply_to_queue(del_dev.guid, static_cast(ResponseCode::OK), "终端 id: " + del_dev.terminal_id + " 台账删除成功"); } else { - send_reply_to_queue(del_dev.guid, "2", + send_reply_to_queue(del_dev.guid, static_cast(ResponseCode::NOT_FOUND), "终端 id: " + del_dev.terminal_id + " 台账删除失败，未找到终端"); } } @@ -2261,50 +2250,6 @@ void execute_bash(string fun,int process_num,string type) ////////////////////////////////////////////////////////////////////////////////////////////////////////////////打印台账更新 -void print_monitor(const ledger_monitor& mon) { - auto safe = [](const std::string& s) { return s.empty() ? "N/A" : s; }; - - std::cout << "Monitor ID: " << safe(mon.monitor_id) << "\n"; - std::cout << "Terminal ID: " << safe(mon.terminal_id) << "\n"; - std::cout << "Monitor Name: " << safe(mon.monitor_name) << "\n"; - std::cout << "Logical Device Sequence: " << safe(mon.logical_device_seq) << "\n"; - std::cout << "Voltage Level: " << safe(mon.voltage_level) << "\n"; - std::cout << "Terminal Connect: " << safe(mon.terminal_connect) << "\n"; - std::cout << "Timestamp: " << safe(mon.timestamp) << "\n"; - std::cout << "Status: " << safe(mon.status) << "\n"; - - std::cout << "CT1: " << mon.CT1 << "\n"; - std::cout << "CT2: " << mon.CT2 << "\n"; - std::cout << "PT1: " << mon.PT1 << "\n"; - std::cout << "PT2: " << mon.PT2 << "\n"; -} - -void print_terminal(const terminal_dev& tmnl) { - auto safe = [](const std::string& s) { return s.empty() ? "N/A" : s; }; - - std::cout << "GUID: " << safe(tmnl.guid) << "\n"; - std::cout << "Terminal ID: " << safe(tmnl.terminal_id) << "\n"; - std::cout << "Terminal Code: " << safe(tmnl.terminal_name)<< "\n"; - std::cout << "Organization Name: "<< safe(tmnl.org_name) << "\n"; - std::cout << "Maintenance Name: " << safe(tmnl.maint_name) << "\n"; - std::cout << "Station Name: " << safe(tmnl.station_name) << "\n"; - std::cout << "Factory Name: " << safe(tmnl.tmnl_factory) << "\n"; - std::cout << "Terminal Status: " << safe(tmnl.tmnl_status) << "\n"; - std::cout << "Device Type: " << safe(tmnl.dev_type) << "\n"; - std::cout << "Device Key: " << safe(tmnl.dev_key) << "\n"; - std::cout << "Device Series: " << safe(tmnl.dev_series) << "\n"; - std::cout << "Address: " << safe(tmnl.addr_str) << "\n"; - std::cout << "Port: " << safe(tmnl.port) << "\n"; - std::cout << "Timestamp: " << safe(tmnl.timestamp) << "\n"; - std::cout << "Righttime: " << safe(tmnl.Righttime) << "\n"; - std::cout << "mac: " << safe(tmnl.mac) << "\n"; - - for (size_t i = 0; i < 10 && !tmnl.line[i].monitor_id.empty(); ++i) { - std::cout << " Monitor " << (i + 1) << ":\n"; - print_monitor(tmnl.line[i]); - } -} - void print_trigger_update_xml(const trigger_update_xml_t& trigger_update) { std::cout << "Work Updates Count: " << trigger_update.work_updates.size() << "\n"; std::cout << "New Updates Count: " << trigger_update.new_updates.size() << "\n"; @@ -4285,6 +4230,7 @@ void check_recall_file() { // 1) 清理首条 DONE/FAILED bool any_non_empty = false; + bool has_running = false; for (auto& lm : dev.line) { while (!lm.recall_list_static.empty()) { const RecallFile& front = lm.recall_list_static.front(); @@ -4299,43 +4245,48 @@ void check_recall_file() { << " " << front.StartTime << " ~ " << front.EndTime << std::endl; lm.recall_list_static.pop_front(); } else { - break; + break;//找到第一条不是成功或失败的记录就退出循环 + } + } + if (!lm.recall_list_static.empty()) { + any_non_empty = true;//弹出后是否为空 + if (lm.recall_list_static.front().recall_status == static_cast(RecallStatus::RUNNING)) { + has_running = true; //存在测点正在补招 } } - if (!lm.recall_list_static.empty()) any_non_empty = true; } // 无条目时的装置态收尾 - if (!any_non_empty && dev.busytype == static_cast(DeviceState::READING_EVENTLOG)) { - // （保持你原注释）处于“暂态补招”的状态且无条目 -> 清空运行态 + if (!any_non_empty && dev.busytype == static_cast(DeviceState::READING_STATSFILE)) { + // 处于“文件补招”的状态且无条目 -> 清空运行态 dev.guid.clear(); dev.busytype = 0; dev.isbusy = 0; dev.busytimecount = 0; continue; - } else if (!any_non_empty && dev.busytype == static_cast(DeviceState::IDLE)) { + } else if (!any_non_empty && dev.busytype != static_cast(DeviceState::IDLE)) {//其他运行态不处理，idle往下执行 continue; } // 2) 若任一 monitor 的首条为 RUNNING，则该终端正在补招中 -> 不下发新的任务（但需要推进状态机！） - bool has_running = false; + /*bool has_running = false; for (auto& lm : dev.line) { if (!lm.recall_list_static.empty() && lm.recall_list_static.front().recall_status == static_cast(RecallStatus::RUNNING)) { - has_running = true; + has_running = true; //存在测点正在补招 break; } - } + }*/ // ★新增：当存在 RUNNING 时，推进“该终端的首条补招记录”的两步状态机 if (has_running) { for (auto& lm : dev.line) { - if (lm.recall_list_static.empty()) continue; - RecallFile& front = lm.recall_list_static.front(); - if (front.recall_status != static_cast(RecallStatus::RUNNING)) continue; + if (lm.recall_list_static.empty()) continue;//跳过没有记录的测点 + RecallFile& front = lm.recall_list_static.front();//有记录测点的第一个补招 + if (front.recall_status != static_cast(RecallStatus::RUNNING)) continue;//跳过不是正在补招的记录 - // 初始化阶段：从 NOT_STARTED->RUNNING 已完成，此处保证 phase 就绪 - if (front.phase == RecallPhase::IDLE) { + // 初始化阶段：补招分为两个阶段，读文件列表和下载文件，如果是刚进入 RUNNING 状态则初始化 + if (front.phase == RecallPhase::IDLE) { //暂态补招 front.phase = RecallPhase::LISTING; front.cur_dir_index = 0; front.cur_dir.clear(); @@ -4522,31 +4473,31 @@ void check_recall_file() { // 3) 没有 RUNNING：挑选第一条 NOT_STARTED，并发起“首个目录”的请求 for (auto& lm : dev.line) { - if (lm.recall_list_static.empty()) continue; + if (lm.recall_list_static.empty()) continue; //跳过补招列表为空的测点 - RecallFile& front = lm.recall_list_static.front(); - if (front.recall_status == static_cast(RecallStatus::NOT_STARTED)) { + RecallFile& front = lm.recall_list_static.front(); //取测点第一条记录 + if (front.recall_status == static_cast(RecallStatus::NOT_STARTED)) { //补招未开始 // 标记为 RUNNING，并设置终端忙状态 - front.recall_status = static_cast(RecallStatus::RUNNING); - dev.isbusy = 1; - dev.busytype = static_cast(DeviceState::READING_STATSFILE); - dev.busytimecount = 0; + front.recall_status = static_cast(RecallStatus::RUNNING); //该补招记录刷新为补招中 + dev.isbusy = 1; //装置由idle标记为忙 + dev.busytype = static_cast(DeviceState::READING_STATSFILE);//装置状态刷新为正在补招文件 + dev.busytimecount = 0; //清空业务超时计数 // 初始化状态机并发出第一个目录请求 front.reset_runtime(true);//保留直下文件信息 - front.phase = RecallPhase::LISTING; - if (!front.dir_candidates.empty()) { - front.cur_dir_index = 0; - front.cur_dir = front.dir_candidates[0]; - front.list_result = ActionResult::PENDING; - ClientManager::instance().add_file_menu_action_to_device(dev.terminal_id, front.cur_dir); + front.phase = RecallPhase::LISTING; //正在请求并等待“目录文件名列表” + if (!front.dir_candidates.empty()) {//目录列表非空 + front.cur_dir_index = 0; //正在尝试的目录下标 + front.cur_dir = front.dir_candidates[0]; //第一个目录 + front.list_result = ActionResult::PENDING; //目录状态：等待返回 + ClientManager::instance().add_file_menu_action_to_device(dev.terminal_id, front.cur_dir);//调用目录请求接口 std::cout << "[check_recall_stat] LIST start dev=" << dev.terminal_id << " monitor=" << lm.monitor_id << " dir=" << front.cur_dir << " start=" << front.StartTime << " end=" << front.EndTime << std::endl; } else { - front.recall_status = static_cast(RecallStatus::FAILED); + front.recall_status = static_cast(RecallStatus::FAILED); //目录列表空，失败 std::cout << "[check_recall_stat] empty dir_candidates, FAIL dev=" << dev.terminal_id << " monitor=" << lm.monitor_id << std::endl; } @@ -4746,12 +4697,13 @@ void on_device_response_minimal(int response_code, break; } + std::vector names; // 2) 按该终端当前 busytype 分支处理 const int bt = dev->busytype; if (bt == static_cast(DeviceState::READING_FILEMENU)) { // ====== 分支 A：当前业务就是“读取文件目录” ====== if (ok) { - std::vector names; + if (filemenu_cache_take(id, names)) { //发送目录 @@ -4786,41 +4738,42 @@ void on_device_response_minimal(int response_code, bt == static_cast(DeviceState::READING_EVENTFILE) || bt == static_cast(DeviceState::READING_STATSFILE) ) { - // ====== 分支 B：当前业务为“下载事件文件/统计文件”（两者处理相同） ====== - // 一个装置同一时刻只会有一个监测点在下载 + // ====== 分支 B：当前业务为“事件文件/统计文件”（两者处理相同） ====== + // 一个装置同一时刻只会有一个监测点在补招 ledger_monitor* running_monitor = nullptr; RecallFile* running_front = nullptr; - // 在该终端下，找到“首条 recall_list_static.front() 正在运行且处于 DOWNLOADING 的监测点” + // 在该终端下，找到“首条 recall_list_static.front() 正在运行且处于 LISTING 的监测点” for (auto& lm : dev->line) { if (lm.recall_list_static.empty()) continue; RecallFile& f = lm.recall_list_static.front(); - if (f.recall_status == static_cast(RecallStatus::RUNNING) - && f.phase == RecallPhase::DOWNLOADING) { - running_monitor = &lm; - running_front = &f; + if (f.recall_status == static_cast(RecallStatus::RUNNING)//补招中 + && f.phase == RecallPhase::LISTING) {//正在请求目录 + running_monitor = &lm; //补招的测点 + running_front = &f; //补招记录 break; } } - if (!running_monitor || !running_front) { - std::cout << "[RESP][FILEDATA->(EVENT/STATS)FILE][WARN] dev=" << id - << " no RUNNING/DOWNLOADING recall on this device, ignore resp" + if (!running_monitor || !running_front) { //该装置没有正在补招的测点和补招记录，退出处理 + std::cout << "[RESP][FILEMENU->(EVENT/STATS)FILE][WARN] dev=" << id + << " no RUNNING/LISTING recall on this device, ignore resp" << " rc=" << response_code << std::endl; break; } - // 根据回执结果，回写下载结果；状态机会在下一轮推进到下一个文件/结束 + // 根据回执结果，回写目录结果；状态机会在下一轮推进到下一个目录/结束 if (ok) { - running_front->download_result = ActionResult::OK; - std::cout << "[RESP][FILEDATA->(EVENT/STATS)FILE][OK] dev=" << id + running_front->dir_files[running_front->cur_dir] = std::move(names); + running_front->list_result = ActionResult::OK; + std::cout << "[RESP][FILEMENU->(EVENT/STATS)FILE][OK] dev=" << id << " monitor=" << running_monitor->monitor_id - << " file=" << running_front->downloading_file << std::endl; + << " dir=" << running_front->cur_dir << std::endl; } else { - running_front->download_result = ActionResult::FAIL; - std::cout << "[RESP][FILEDATA->(EVENT/STATS)FILE][FAIL] dev=" << id + running_front->list_result = ActionResult::FAIL; + std::cout << "[RESP][FILEMENU->(EVENT/STATS)FILE][FAIL] dev=" << id << " monitor=" << running_monitor->monitor_id - << " file=" << running_front->downloading_file + << " dir=" << running_front->cur_dir << " rc=" << response_code << std::endl; } break; @@ -5162,6 +5115,16 @@ void enqueue_realtime_pq(const RealtagPqDate_float& realdata, //lnk实时数据使用接口发送20250711 time_t data_time = ConvertToTimestamp(realdata.time); + { + std::lock_guard lk(g_last_ts_mtx); + auto it = g_last_ts_by_devid.find(devid); + if (it != g_last_ts_by_devid.end() && it->second == data_time) { + // 同一设备与上次时间相同 → 丢弃本次 + return; + } + // 记录本次时间 + g_last_ts_by_devid[devid] = data_time; + } std::vector arr; arr.push_back({1, //数据属性 -1-无， 0-“Rt”,1-“Max”,2-“Min”,3-“Avg”,4-“Cp95” @@ -5262,7 +5225,7 @@ size_t erase_one_terminals_by_id(const std::string& terminal_id) { } /////////////////////////////////////////////////////////////////////////////////////////////// -static DeviceInfo make_device_from_terminal(const terminal_dev& t) { +DeviceInfo make_device_from_terminal(const terminal_dev& t) { DeviceInfo d; // 基本信息 diff --git a/LFtid1056/cloudfront/code/interface.h b/LFtid1056/cloudfront/code/interface.h index d9aaa8a..bdace14 100644 --- a/LFtid1056/cloudfront/code/interface.h +++ b/LFtid1056/cloudfront/code/interface.h @@ -8,6 +8,7 @@ #include #include #include +#include /////////////////////////////////////////////////////////////////////////////////////////// @@ -203,6 +204,7 @@ public: class update_dev { public: + std::string guid; // ★新增：供发送回复使用 std::string terminal_id; std::string terminal_name; @@ -222,7 +224,7 @@ public: std::string maxProcessNum; std::string mac; // 装置MAC地址，接口中从addr_str获取，因为ip和mac放同一位置 - std::vector line; + std::vector line; }; ////////////////////////////////////////////////////////////////////台账更新用 @@ -529,7 +531,6 @@ std::string get_current_time(); bool is_blank(const std::string& str); -void print_terminal(const terminal_dev& tmnl); void printTerminalDevMap(const std::map& terminal_dev_map); void upload_data_test(); @@ -728,7 +729,72 @@ size_t erase_one_terminals_by_id(const std::string& terminal_id); //转换结构 DeviceInfo make_device_from_terminal(const terminal_dev& t); -//小工具 +////////////////////////////////////////////////////////////////////////////////////////////////////////////////模板打印 +// ===== 通用打印实现（模板） ===== + +template +inline void print_monitor_common(const Mon& mon) { + auto safe = [](const std::string& s) -> const std::string& { + static const std::string NA = "N/A"; + return s.empty() ? NA : s; + }; + + std::cout << "Monitor ID: " << safe(mon.monitor_id) << "\n"; + std::cout << "Terminal ID: " << safe(mon.terminal_id) << "\n"; + std::cout << "Monitor Name: " << safe(mon.monitor_name) << "\n"; + std::cout << "Logical Device Sequence: " << safe(mon.logical_device_seq) << "\n"; + std::cout << "Voltage Level: " << safe(mon.voltage_level) << "\n"; + std::cout << "Terminal Connect: " << safe(mon.terminal_connect) << "\n"; + std::cout << "Timestamp: " << safe(mon.timestamp) << "\n"; + std::cout << "Status: " << safe(mon.status) << "\n"; + std::cout << "CT1: " << mon.CT1 << "\n"; + std::cout << "CT2: " << mon.CT2 << "\n"; + std::cout << "PT1: " << mon.PT1 << "\n"; + std::cout << "PT2: " << mon.PT2 << "\n"; +} + +template +inline void print_terminal_common(const Dev& tmnl) { + auto safe = [](const std::string& s) -> const std::string& { + static const std::string NA = "N/A"; + return s.empty() ? NA : s; + }; + + std::cout << "GUID: " << safe(tmnl.guid) << "\n"; + std::cout << "Terminal ID: " << safe(tmnl.terminal_id) << "\n"; + std::cout << "Terminal Code: " << safe(tmnl.terminal_name) << "\n"; + std::cout << "Organization Name: "<< safe(tmnl.org_name) << "\n"; + std::cout << "Maintenance Name: " << safe(tmnl.maint_name) << "\n"; + std::cout << "Station Name: " << safe(tmnl.station_name) << "\n"; + std::cout << "Factory Name: " << safe(tmnl.tmnl_factory) << "\n"; + std::cout << "Terminal Status: " << safe(tmnl.tmnl_status) << "\n"; + std::cout << "Device Type: " << safe(tmnl.dev_type) << "\n"; + std::cout << "Device Key: " << safe(tmnl.dev_key) << "\n"; + std::cout << "Device Series: " << safe(tmnl.dev_series) << "\n"; + std::cout << "Address: " << safe(tmnl.addr_str) << "\n"; + std::cout << "Port: " << safe(tmnl.port) << "\n"; + std::cout << "Timestamp: " << safe(tmnl.timestamp) << "\n"; + std::cout << "Righttime: " << safe(tmnl.Righttime) << "\n"; + std::cout << "mac: " << safe(tmnl.mac) << "\n"; + + // 安全遍历最多前 10 个测点（不越界），且跳过空 monitor_id 的项 + const size_t n = std::min(tmnl.line.size(), 10); + for (size_t i = 0; i < n; ++i) { + if (tmnl.line[i].monitor_id.empty()) continue; + std::cout << " Monitor " << (i + 1) << ":\n"; + print_monitor_common(tmnl.line[i]); // 模板会自动匹配 update/ledger 的 monitor + } +} + +// ===== 薄包装重载（保持原有函数名不变）===== + +inline void print_monitor(const update_monitor& mon) { print_monitor_common(mon); } +inline void print_monitor(const ledger_monitor& mon) { print_monitor_common(mon); } + +inline void print_terminal(const update_dev& tmnl) { print_terminal_common(tmnl); } +inline void print_terminal(const terminal_dev& tmnl) { print_terminal_common(tmnl); } + +///////////////////////////////////////////////////////////////////////////////////////////////////////////////小工具 inline std::string trim_cstr(const char* s, size_t n) { if (!s) return {}; size_t end = 0; diff --git a/LFtid1056/cloudfront/code/log4.cpp b/LFtid1056/cloudfront/code/log4.cpp index bd4567d..0e2df6e 100644 --- a/LFtid1056/cloudfront/code/log4.cpp +++ b/LFtid1056/cloudfront/code/log4.cpp @@ -481,10 +481,6 @@ extern "C" { void log_warn (const char* key, const char* msg) { log4_log_with_level(key, msg, 2); } void log_error(const char* key, const char* msg) { log4_log_with_level(key, msg, 3); } - void send_reply_to_queue_c(const char* guid, const char* step, const char* result) { - send_reply_to_queue(std::string(guid), std::string(step), std::string(result)); - } - //标准化日志接口 // #define LOGMSG_WITH_TS // 需要时间时再打开 diff --git a/LFtid1056/cloudfront/code/log4.h b/LFtid1056/cloudfront/code/log4.h index b86b126..b9a1546 100644 --- a/LFtid1056/cloudfront/code/log4.h +++ b/LFtid1056/cloudfront/code/log4.h @@ -61,7 +61,7 @@ struct DebugSwitch { extern std::map logger_map; extern DebugSwitch g_debug_switch; -extern void send_reply_to_queue(const std::string& guid, const std::string& step, const std::string& result); +extern void send_reply_to_queue(const std::string& guid, const int code, const std::string& result); //std::string get_front_type_from_subdir(); @@ -94,8 +94,6 @@ void log_debug(const char* key, const char* msg); void log_info(const char* key, const char* msg); void log_warn(const char* key, const char* msg); void log_error(const char* key, const char* msg); - -void send_reply_to_queue_c(const char* guid, const char* step, const char* result); void format_log_msg(char* buf, size_t buf_size, const char* fmt, ...); // ====================== ★新增：线程局部变量透传 code ====================== diff --git a/LFtid1056/cloudfront/code/main.cpp b/LFtid1056/cloudfront/code/main.cpp index 8adc8cf..f4f9957 100644 --- a/LFtid1056/cloudfront/code/main.cpp +++ b/LFtid1056/cloudfront/code/main.cpp @@ -320,7 +320,7 @@ void Front::FrontThread() { check_recall_file(); //处理补招文件-稳态和暂态 check_recall_event(); // 处理补招事件，从list中读取然后直接调用接口,每一条可能都不同测点，每个测点自己做好记录 - check_ledger_update(); // 触发台账更新 + //check_ledger_update(); // 触发台账更新 std::this_thread::sleep_for(std::chrono::milliseconds(100)); } diff --git a/LFtid1056/cloudfront/code/rocketmq.cpp b/LFtid1056/cloudfront/code/rocketmq.cpp index 721c368..cefafe7 100644 --- a/LFtid1056/cloudfront/code/rocketmq.cpp +++ b/LFtid1056/cloudfront/code/rocketmq.cpp @@ -408,7 +408,7 @@ bool parseJsonMessageRT(const std::string& body,std::string& devSeries,ushort& l // 回复：执行结果直接看实时数据，不需要再回复，1是收到消息 if (!guid.empty()) { - send_reply_to_queue(guid, "1", "收到三秒数据指令"); + send_reply_to_queue(guid, static_cast(ResponseCode::ACCEPTED), "收到三秒数据指令"); } return true; @@ -500,7 +500,7 @@ bool parseJsonMessageSET(const std::string& json_str) { DIY_WARNLOG("process", "【WARN】前置的%d号进程执行指令:%s,reset表示重启所有进程,add表示添加进程", g_front_seg_index, fun.c_str()); - send_reply_to_queue(guid, "1", "收到重置进程指令,重启所有进程!"); + send_reply_to_queue(guid, static_cast(ResponseCode::ACCEPTED), "收到重置进程指令,重启所有进程!"); std::cout << "this msg should only execute once" << std::endl; } else { std::cout << "only cfg_stat_data index 1 can control process, this process not handle this msg" << std::endl; @@ -508,7 +508,7 @@ bool parseJsonMessageSET(const std::string& json_str) { } else if (fun == "delete") { - send_reply_to_queue(guid, "1", "收到删除进程指令,这个进程将会重启 "); + send_reply_to_queue(guid, static_cast(ResponseCode::ACCEPTED), "收到删除进程指令,这个进程将会重启 "); DIY_WARNLOG("process", "【WARN】前置的%d号进程执行指令:%s,即将重启", g_front_seg_index, fun.c_str()); @@ -602,7 +602,7 @@ bool parseJsonMessageLOG(const std::string& json_str) { /*std::cout << "msg frontType: " << frontType << " self frontType: " << subdir << std::endl;*/ // 回复消息 - send_reply_to_queue(guid, "1", "收到实时日志指令"); + send_reply_to_queue(guid, static_cast(ResponseCode::ACCEPTED), "收到实时日志指令"); if (code_str == "set_log") { // 校验数据合法性 @@ -673,7 +673,7 @@ bool parseJsonMessageUD(const std::string& json_str, const std::string& output_d DIY_INFOLOG("process", "【NORMAL】前置的%d号进程处理topic:%s_%s的台账更新消息", g_front_seg_index, FRONT_INST.c_str(), G_MQCONSUMER_TOPIC_UD.c_str()); - send_reply_to_queue(guid, "1", "收到台账更新指令"); + send_reply_to_queue(guid, static_cast(ResponseCode::ACCEPTED), "收到台账更新指令"); if (code_str == "add_terminal" || code_str == "ledger_modify") { std::cout << "add or modify ledger" << std::endl; @@ -717,10 +717,10 @@ bool parseJsonMessageUD(const std::string& json_str, const std::string& output_d m.terminal_connect = monitor_item.value("ptType", ""); //m.timestamp = json_data.timestamp; m.terminal_id = monitor_item.value("deviceId", json_data.terminal_id); - m.CT1 = monitor_item.value("CT1", ""); - m.CT2 = monitor_item.value("CT2", ""); - m.PT1 = monitor_item.value("PT1", ""); - m.PT2 = monitor_item.value("PT2", ""); + m.CT1 = monitor_item.value("CT1", 0.0); + m.CT2 = monitor_item.value("CT2", 0.0); + m.PT1 = monitor_item.value("PT1", 0.0); + m.PT2 = monitor_item.value("PT2", 0.0); } } @@ -734,32 +734,84 @@ bool parseJsonMessageUD(const std::string& json_str, const std::string& output_d writeToFile(file_name, xmlContent); }*/ if(code_str == "add_terminal"){ - init_loggers_bydevid(target_dev.terminal_id); - terminal_devlist.push_back(json_data); + std::lock_guard lock(ledgermtx); + // ① 先判断 json_data.terminal_id 是否已在当前进程维护的终端列表中 + const std::string& tid = json_data.terminal_id; + auto it = std::find_if(terminal_devlist.begin(), terminal_devlist.end(), + [&](const terminal_dev& d){ return d.terminal_id == tid; }); - //调用接口添加到通讯列表 - DeviceInfo device = make_device_from_terminal(json_data); - ClientManager::instance().add_device(device); + if (it == terminal_devlist.end()) { - send_reply_to_queue(new_dev.guid, "2", - "终端 id: " + new_dev.terminal_id + " 台账添加成功"); - } - else if(code_str == "ledger_modify"){ - - if(erase_one_terminals_by_id(json_data.terminal_id) == 1){ - init_loggers_bydevid(target_dev.terminal_id); + init_loggers_bydevid(json_data.terminal_id); terminal_devlist.push_back(json_data); //调用接口添加到通讯列表 DeviceInfo device = make_device_from_terminal(json_data); ClientManager::instance().add_device(device); - send_reply_to_queue(del_dev.guid, "2", - "终端 id: " + del_dev.terminal_id + " 台账修改成功"); + send_reply_to_queue(json_data.guid, static_cast(ResponseCode::OK), + "终端 id: " + json_data.terminal_id + " 台账添加成功"); } else{ - send_reply_to_queue(del_dev.guid, "2", - "终端 id: " + del_dev.terminal_id + " 台账修改失败"); + send_reply_to_queue(json_data.guid, static_cast(ResponseCode::OK), + "终端 id: " + json_data.terminal_id + " 已存在该装置，修改这个装置的台账"); + + if(erase_one_terminals_by_id(json_data.terminal_id) == 1){ + init_loggers_bydevid(json_data.terminal_id); + terminal_devlist.push_back(json_data); + + //调用接口添加到通讯列表 + DeviceInfo device = make_device_from_terminal(json_data); + ClientManager::instance().add_device(device); + + send_reply_to_queue(json_data.guid, static_cast(ResponseCode::OK), + "终端 id: " + json_data.terminal_id + " 台账修改成功"); + } + else{ + send_reply_to_queue(json_data.guid, static_cast(ResponseCode::BAD_REQUEST), + "终端 id: " + json_data.terminal_id + " 台账修改失败"); + } + } + } + else if(code_str == "ledger_modify"){ + std::lock_guard lock(ledgermtx); + + // ① 先判断 json_data.terminal_id 是否已在当前进程维护的终端列表中 + const std::string& tid = json_data.terminal_id; + auto it = std::find_if(terminal_devlist.begin(), terminal_devlist.end(), + [&](const terminal_dev& d){ return d.terminal_id == tid; }); + + if (it == terminal_devlist.end()) { + send_reply_to_queue(json_data.guid, static_cast(ResponseCode::NOT_FOUND), + "终端 id: " + tid + " 无法修改台账，未找到指定装置,改为添加这个装置"); + + DIY_WARNLOG("process", "【WARN】无法修改台账，未找到指定装置: %s ,改为添加这个装置", tid.c_str()); + + init_loggers_bydevid(json_data.terminal_id); + terminal_devlist.push_back(json_data); + + //调用接口添加到通讯列表 + DeviceInfo device = make_device_from_terminal(json_data); + ClientManager::instance().add_device(device); + + send_reply_to_queue(json_data.guid, static_cast(ResponseCode::OK), + "终端 id: " + json_data.terminal_id + " 台账添加成功"); + } + + if(erase_one_terminals_by_id(json_data.terminal_id) == 1){ + init_loggers_bydevid(json_data.terminal_id); + terminal_devlist.push_back(json_data); + + //调用接口添加到通讯列表 + DeviceInfo device = make_device_from_terminal(json_data); + ClientManager::instance().add_device(device); + + send_reply_to_queue(json_data.guid, static_cast(ResponseCode::OK), + "终端 id: " + json_data.terminal_id + " 台账修改成功"); + } + else{ + send_reply_to_queue(json_data.guid, static_cast(ResponseCode::BAD_REQUEST), + "终端 id: " + json_data.terminal_id + " 台账修改失败"); } } @@ -784,13 +836,13 @@ bool parseJsonMessageUD(const std::string& json_str, const std::string& output_d //直接加锁删除 std::lock_guard lock(ledgermtx); if(erase_one_terminals_by_id(json_data.terminal_id) == 1){ - lientManager::instance().remove_device(json_data.terminal_id); - send_reply_to_queue(del_dev.guid, "2", - "终端 id: " + del_dev.terminal_id + " 台账删除成功"); + ClientManager::instance().remove_device(json_data.terminal_id); + send_reply_to_queue(json_data.guid, static_cast(ResponseCode::OK), + "终端 id: " + json_data.terminal_id + " 台账删除成功"); } else{ - send_reply_to_queue(del_dev.guid, "2", - "终端 id: " + del_dev.terminal_id + " 台账删除失败"); + send_reply_to_queue(json_data.guid, static_cast(ResponseCode::BAD_REQUEST), + "终端 id: " + json_data.terminal_id + " 台账删除失败"); } } } @@ -1307,15 +1359,14 @@ void connect_status_to_queue(const std::string& id, const std::string& datetime, ////////////////////////////////////////////////////////////////////////////////////////////////////////////////响应消息 -void send_reply_to_queue(const std::string& guid, const std::string& step, const std::string& result) { +void send_reply_to_queue(const std::string& guid, const int code, const std::string& result) { try { // 构造 JSON 对象 nlohmann::json obj; obj["guid"] = guid; - obj["step"] = step; + obj["code"] = code; obj["result"] = result; obj["processNo"] = g_front_seg_index; - obj["frontType"] = "cloudfront"; obj["nodeId"] = FRONT_INST; // 构造 queue 消息