zhangwen/front_linux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

modify ledger update and rtdata

Browse Source
This commit is contained in:
lnk
2025-09-30 16:17:04 +08:00
parent caf3c308c1
commit 35231baae7
6 changed files with 233 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files

193
LFtid1056/cloudfront/code/cfg_parser.cpp
View File

@@ -72,6 +72,10 @@ extern std::map<std::string, Xmldata*> xmlinfo_list2;//保存所有型号角形
extern time_t ConvertToTimestamp(const tagTime& time);
////////////////////////////////////////////////////////////////////////////////////////////////////
//实时数据时间记录
std::mutex g_last_ts_mtx;
std::unordered_map<std::string, time_t> g_last_ts_by_devid;
static std::mutex g_filemenu_cache_mtx;
std::map<std::string, std::vector<tag_dir_info>> 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<std::string>();
send_reply_to_queue(guid, "1", "收到补招指令");
send_reply_to_queue(guid, static_cast<int>(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<size_t>(IED_COUNT)) {
send_reply_to_queue(new_dev.guid, "2",
send_reply_to_queue(new_dev.guid, static_cast<int>(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<int>(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<int>(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<int>(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<int>(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<int>(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<int>(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<int>(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<int>(RecallStatus::RUNNING)) {
has_running = true; //存在测点正在补招
}
}
if (!lm.recall_list_static.empty()) any_non_empty = true;
}
// 无条目时的装置态收尾
if (!any_non_empty && dev.busytype == static_cast<int>(DeviceState::READING_EVENTLOG)) {
// （保持你原注释）处于“暂态补招”的状态且无条目 -> 清空运行态
if (!any_non_empty && dev.busytype == static_cast<int>(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<int>(DeviceState::IDLE)) {
} else if (!any_non_empty && dev.busytype != static_cast<int>(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<int>(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<int>(RecallStatus::RUNNING)) continue;
if (lm.recall_list_static.empty()) continue;//跳过没有记录的测点
RecallFile& front = lm.recall_list_static.front();//有记录测点的第一个补招
if (front.recall_status != static_cast<int>(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<int>(RecallStatus::NOT_STARTED)) {
RecallFile& front = lm.recall_list_static.front(); //取测点第一条记录
if (front.recall_status == static_cast<int>(RecallStatus::NOT_STARTED)) { //补招未开始
// 标记为 RUNNING，并设置终端忙状态
front.recall_status = static_cast<int>(RecallStatus::RUNNING);
dev.isbusy = 1;
dev.busytype = static_cast<int>(DeviceState::READING_STATSFILE);
dev.busytimecount = 0;
front.recall_status = static_cast<int>(RecallStatus::RUNNING); //该补招记录刷新为补招中
dev.isbusy = 1; //装置由idle标记为忙
dev.busytype = static_cast<int>(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<int>(RecallStatus::FAILED);
front.recall_status = static_cast<int>(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<tag_dir_info> names;
// 2) 按该终端当前 busytype 分支处理
const int bt = dev->busytype;
if (bt == static_cast<int>(DeviceState::READING_FILEMENU)) {
// ====== 分支 A：当前业务就是“读取文件目录” ======
if (ok) {
std::vector<tag_dir_info> names;
if (filemenu_cache_take(id, names)) {
//发送目录
@@ -4786,41 +4738,42 @@ void on_device_response_minimal(int response_code,
bt == static_cast<int>(DeviceState::READING_EVENTFILE)
|| bt == static_cast<int>(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<int>(RecallStatus::RUNNING)
&& f.phase == RecallPhase::DOWNLOADING) {
running_monitor = &lm;
running_front = &f;
if (f.recall_status == static_cast<int>(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<std::mutex> 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<DataArrayItem> 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;
// 基本信息

72
LFtid1056/cloudfront/code/interface.h
View File

@@ -8,6 +8,7 @@
#include <map>
#include <mutex>
#include <sstream>
#include <iostream>
///////////////////////////////////////////////////////////////////////////////////////////
@@ -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<ledger_monitor> line;
std::vector<update_monitor> 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<std::string, terminal_dev>& 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 <typename Mon>
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 <typename Dev>
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<size_t>(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;

4
LFtid1056/cloudfront/code/log4.cpp
View File

@@ -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 // 需要时间时再打开

4
LFtid1056/cloudfront/code/log4.h
View File

@@ -61,7 +61,7 @@ struct DebugSwitch {
extern std::map<std::string, TypedLogger> 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 ======================

2
LFtid1056/cloudfront/code/main.cpp
View File

@@ -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));
}

121
LFtid1056/cloudfront/code/rocketmq.cpp
View File

@@ -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<int>(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<int>(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<int>(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<int>(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<int>(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);
std::lock_guard<std::mutex> 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()) {
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(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);
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<int>(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<int>(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<int>(ResponseCode::OK),
"终端 id: " + json_data.terminal_id + " 台账修改成功");
}
else{
send_reply_to_queue(json_data.guid, static_cast<int>(ResponseCode::BAD_REQUEST),
"终端 id: " + json_data.terminal_id + " 台账修改失败");
}
}
}
else if(code_str == "ledger_modify"){
std::lock_guard<std::mutex> 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<int>(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<int>(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<int>(ResponseCode::OK),
"终端 id: " + json_data.terminal_id + " 台账修改成功");
}
else{
send_reply_to_queue(json_data.guid, static_cast<int>(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<std::mutex> 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<int>(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<int>(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 消息