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

generate ledger to main thread

Browse Source
This commit is contained in:
lnk
2025-06-26 16:44:21 +08:00
parent 4e6ba12b25
commit 747d6c757c
7 changed files with 93 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -31,6 +31,8 @@
#include "tinyxml2.h"
#include "rocketmq.h"
/////////////////////////////////////////////////////////////////////////////////////////////////
using namespace std;
@@ -58,8 +60,6 @@ extern std::list<queue_data_t> queue_data_list; //queue发送数据链表
extern int three_secs_enabled;
extern std::vector<terminal_dev> terminal_devlist;
extern std::map<std::string, Xmldata*> xmlinfo_list;//保存所有型号对应的icd映射文件解析数据
extern XmlConfig xmlcfg;//星形接线xml节点解析的数据-默认映射文件解析数据
extern std::list<CTopic *> topicList; //队列发送主题链表
@@ -2686,15 +2686,15 @@ void to_json(nlohmann::json& j, const FullObj& f) {
};
}
std::string generate_json(
int Mid,
int Did,
int Pri,
int Type,
int Cldid,
int DataType,
int DataAttr,
int DsNameIdx,
const std::vector<DataArrayItem>& dataArray //构造出array后调用这个函数
int Mid, //需应答的报文订阅者收到后需以此ID应答，无需应答填入“-1”
int Did, //设备唯一标识Ldid，填入0代表Ndid。
int Pri, //报文处理的优先级
int Type, //消息类型
int Cldid, //逻辑子设备ID，0-逻辑设备本身，无填-1
int DataType, //数据类型，0-表示以数据集方式上送
int DataAttr, //数据属性：无“0”、实时“1”、统计“2”等。
int DsNameIdx, //数据集序号（以数据集方式上送），无填-1
const std::vector<DataArrayItem>& dataArray //数据数组。
) {
FullObj fobj;
fobj.Mid = Mid;
@@ -2712,7 +2712,11 @@ std::string generate_json(
void upload_data_test(){
std::vector<DataArrayItem> arr;
arr.push_back({1, 1725477660, 0, 1, "xxxx"});
arr.push_back({1, 1725477660, 0, 1, "xxxx"}); //数据属性 -1-无， 0-“Rt”,1-“Max”,2-“Min”,3-“Avg”,4-“Cp95”
//数据时标，相对1970年的秒，无效填入“-1”
//数据时标，微秒钟，无效填入“-1”
//数据标识，1-标识数据异常
//数据序列（数据集上送时将二进制数据流转换成Base64字符串，其他数据为object）
arr.push_back({2, 1691741340, 0, 1, "yyyy"});
std::string js = generate_json(
@@ -2728,3 +2732,34 @@ void upload_data_test(){
std::lock_guard<std::mutex> lock(queue_data_list_mutex);
queue_data_list.push_back(data);
}
//////////////////////////////////////////////////////////////////////////////////////////////////台账赋值给通信
std::vector<DeviceInfo> GenerateDeviceInfoFromLedger(const std::vector<terminal_dev>& terminal_devlist) {
std::vector<DeviceInfo> devices;
for (const auto& terminal : terminal_devlist) {
DeviceInfo device;
device.device_id = terminal.terminal_id;
device.name = terminal.terminal_name;
device.model = terminal.dev_series;
device.mac = terminal.mac;
for (const auto& monitor : terminal.line) {
PointInfo point;
point.point_id = monitor.monitor_id;
point.name = monitor.monitor_name;
point.device_id = terminal.terminal_id;
point.PT1 = monitor.PT1;
point.PT2 = monitor.PT2;
point.CT1 = monitor.CT1;
point.CT2 = monitor.CT2;
device.points.push_back(point);
}
devices.push_back(device);
}
return devices;
}

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

@@ -12,6 +12,8 @@
#include "nlohmann/json.hpp"
#include "../../client2.h"
///////////////////////////////////////////////////////////////////////////////////////////
class Front;
@@ -308,6 +310,9 @@ int parse_model_cfg_web();
void qvvr_test();
void Fileupload_test();
extern std::vector<terminal_dev> terminal_devlist;
extern std::mutex ledgermtx;
//////////////////////////////////////////////////////////////////////////////////cfg_parse的函数声明
void init_config();
@@ -403,6 +408,10 @@ void to_json(nlohmann::json& j, const FullObj& f);
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////
std::vector<DeviceInfo> GenerateDeviceInfoFromLedger(const std::vector<terminal_dev>& terminal_devlist);
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////
#endif

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

@@ -47,15 +47,9 @@ extern int g_front_seg_index;
extern std::string FRONT_INST;
extern std::string subdir;
//mq
extern std::mutex queue_data_list_mutex; //queue发送数据锁
extern std::list<queue_data_t> queue_data_list; //queue发送数据链表
//日志主题
extern std::string G_LOG_TOPIC;
extern std::vector<terminal_dev> terminal_devlist;
////////////////////////////////////////////////////////辅助函数
std::string get_front_type_from_subdir() {
if (subdir == "cfg_3s_data")

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

@@ -78,10 +78,6 @@ extern int TEST_PORT; //测试端口号
extern std::string FRONT_INST;
extern std::mutex queue_data_list_mutex;
extern std::list<queue_data_t> queue_data_list;
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////// 功能函数
template<typename T, typename... Args>

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

@@ -58,10 +58,6 @@ static rocketmq::RocketMQProducer* g_producer = nullptr; //生产者
///////////////////////////////////////////////////////////////////////////////////////////////////////////
//台账
extern std::mutex ledgermtx;
extern std::vector<terminal_dev> terminal_devlist;
//前置进程
extern unsigned int g_node_id;
extern int g_front_seg_index;
@@ -1392,7 +1388,6 @@ bool shouldSkipTerminal(const std::string& terminal_id) {
}
void rocketmq_test_300(int mpnum, int front_index, int type, Front* front) {
if (!INITFLAG) {
std::cout << "前置未初始化完成\n";
return;
@@ -1451,41 +1446,34 @@ void rocketmq_test_300(int mpnum, int front_index, int type,Front* front) {
data.mp_id = dev.line[j].monitor_id;
data.monitor_no = static_cast<int>(i + j);
std::string modified_time = std::to_string(current_time_ms);
std::string modified_time = std::to_string(current_time_ms / 1000);
std::string modified_strText = base_strText;
// 替换 Monitor
size_t monitor_pos = modified_strText.find("\"Monitor\"");
if (monitor_pos != std::string::npos) {
size_t colon_pos = modified_strText.find(":", monitor_pos);
size_t quote_pos = modified_strText.find("\"", colon_pos);
size_t end_quote_pos = modified_strText.find("\"", quote_pos + 1);
if (colon_pos != std::string::npos && quote_pos != std::string::npos && end_quote_pos != std::string::npos) {
modified_strText.replace(quote_pos + 1, end_quote_pos - quote_pos - 1, data.mp_id);
try {
auto j = nlohmann::json::parse(modified_strText);
j["Did"] = i;
if (j.contains("Msg") && j["Msg"].is_object()) {
j["Msg"]["Cldid"] = j;
if (j["Msg"].contains("DataArray") && j["Msg"]["DataArray"].is_array()) {
for (auto& item : j["Msg"]["DataArray"]) {
if (item.is_object()) {
item["DataTimeSec"] = std::stoll(modified_time);
}
}
// 替换 TIME
size_t time_pos = modified_strText.find("\"TIME\"");
if (time_pos != std::string::npos) {
size_t colon_pos = modified_strText.find(":", time_pos);
size_t quote_pos = colon_pos;
size_t end_quote_pos = modified_strText.find(",", quote_pos + 1);
if (colon_pos != std::string::npos && quote_pos != std::string::npos && end_quote_pos != std::string::npos) {
modified_strText.replace(quote_pos + 1, end_quote_pos - quote_pos - 1, modified_time);
}
}
modified_strText = j.dump();
} catch (...) {
// 保持原始文本
}
data.strText = modified_strText;
//my_rocketmq_send(data,front->m_producer);
std::lock_guard<std::mutex> lock(queue_data_list_mutex);
queue_data_list.push_back(data);
std::cout << "Sent message " << (i + 1)
<< " with Monitor " << data.monitor_no
<< " and TIME " << modified_time << std::endl;
}
}
} else {
@@ -1493,43 +1481,36 @@ void rocketmq_test_300(int mpnum, int front_index, int type,Front* front) {
for (int i = 0; (total_messages > 0 && g_front_seg_index == 1 && g_node_id == 100) && i < total_messages; ++i) {
std::string monitor_id = "testmonitor" + std::to_string(i);
data.mp_id = monitor_id;
data.monitor_no = i;
std::string modified_time = std::to_string(current_time_ms);
std::string modified_time = std::to_string(current_time_ms / 1000);
std::string modified_strText = base_strText;
// 替换 Monitor
size_t monitor_pos = modified_strText.find("\"Monitor\"");
if (monitor_pos != std::string::npos) {
size_t colon_pos = modified_strText.find(":", monitor_pos);
size_t quote_pos = modified_strText.find("\"", colon_pos);
size_t end_quote_pos = modified_strText.find("\"", quote_pos + 1);
if (colon_pos != std::string::npos && quote_pos != std::string::npos && end_quote_pos != std::string::npos) {
modified_strText.replace(quote_pos + 1, end_quote_pos - quote_pos - 1, data.mp_id);
try {
auto j = nlohmann::json::parse(modified_strText);
j["Did"] = 0;
if (j.contains("Msg") && j["Msg"].is_object()) {
j["Msg"]["Cldid"] = data.mp_id;
if (j["Msg"].contains("DataArray") && j["Msg"]["DataArray"].is_array()) {
for (auto& item : j["Msg"]["DataArray"]) {
if (item.is_object()) {
item["DataTimeSec"] = std::stoll(modified_time);
}
}
// 替换 TIME
size_t time_pos = modified_strText.find("\"TIME\"");
if (time_pos != std::string::npos) {
size_t colon_pos = modified_strText.find(":", time_pos);
size_t quote_pos = colon_pos;
size_t end_quote_pos = modified_strText.find(",", quote_pos + 1);
if (colon_pos != std::string::npos && quote_pos != std::string::npos && end_quote_pos != std::string::npos) {
modified_strText.replace(quote_pos + 1, end_quote_pos - quote_pos - 1, modified_time);
}
}
modified_strText = j.dump();
} catch (...) {
// 保持原始文本
}
data.strText = modified_strText;
//my_rocketmq_send(data,front->m_producer);
std::lock_guard<std::mutex> lock(queue_data_list_mutex);
queue_data_list.push_back(data);
std::cout << "Sent message " << (i + 1)
<< " with Monitor " << data.monitor_no
<< " and TIME " << modified_time << std::endl;
}
}

7
LFtid1056/cloudfront/code/worker.cpp
View File

@@ -48,11 +48,6 @@ bool showinshellflag =false;
extern std::list<std::string> errorList, warnList, normalList;
extern std::mutex errorListMutex, warnListMutex, normalListMutex;
extern std::vector<terminal_dev> terminal_devlist;
extern std::mutex ledgermtx;
extern std::list<queue_data_t> queue_data_list;
extern int IED_COUNT;
extern int INITFLAG;
extern int g_front_seg_index;
@@ -281,7 +276,7 @@ extern bool normalOutputEnabled;
if (G_TEST_NUM != 0) {
std::cout << "[PeriodicTask] Executing rocketmq_test_300()\n";
rocketmq_test_300(G_TEST_NUM, g_front_seg_index, G_TEST_TYPE,m_front);
upload_data_test();
//upload_data_test();
}
}

2
LFtid1056/main_thread.cpp
View File

@@ -152,6 +152,8 @@ void* client_manager_thread(void* arg) {
// <20><><EFBFBD><EFBFBD>100<30><30><EFBFBD><EFBFBD><EFBFBD><EFBFBD>װ<EFBFBD><D7B0>
std::vector<DeviceInfo> test_devices = generate_test_devices(100);
//std::vector<DeviceInfo> devices = GenerateDeviceInfoFromLedger(terminal_devlist);//lnk<6E><6B><EFBFBD><EFBFBD>
// <20><><EFBFBD><EFBFBD><EFBFBD>ͻ<EFBFBD><CDBB><EFBFBD><EFBFBD><EFBFBD><EFBFBD><EFBFBD>
start_client_connect(devices);