1770 lines
72 KiB
C++
1770 lines
72 KiB
C++
#define NOMINMAX
|
||
#include "pq_app.h"
|
||
|
||
#include "json.hpp"
|
||
|
||
#include <windows.h>
|
||
#include <winhttp.h>
|
||
#include <objbase.h>
|
||
|
||
#include <algorithm>
|
||
#include <chrono>
|
||
#include <cctype>
|
||
#include <cwctype>
|
||
#include <filesystem>
|
||
#include <fstream>
|
||
#include <iomanip>
|
||
#include <map>
|
||
#include <mutex>
|
||
#include <set>
|
||
#include <sstream>
|
||
#include <stdexcept>
|
||
#include <thread>
|
||
#include <unordered_map>
|
||
#include <unordered_set>
|
||
|
||
/*
|
||
* mq.cpp
|
||
*
|
||
* 本文件负责“外部数据输入/输出”：
|
||
* - HTTP 请求台账、ICD 列表和 XML 文件下载；
|
||
* - 动态加载 RocketMQ DLL，初始化生产者/消费者；
|
||
* - 发送数据追踪请求；
|
||
* - 消费 DATATopic/TraceTopic，把 jsondata/origin 落盘到 trace_data；
|
||
* - RocketMQ 不可用时把待发消息写到 mq_outbox_*.jsonl，便于离线排查。
|
||
*
|
||
* 数据追踪落盘结构：
|
||
* runtime\trace_data\process_N\device_x\monitor_y\
|
||
* trace_context.json 本次追踪上下文，GUI 自动解析依赖它；
|
||
* sent_trace_request.json 已发送/待发送追踪请求；
|
||
* jsondata_01.txt DATA_TYPE=01 的 DATATopic 数据；
|
||
* jsondata_02.txt/03.txt 闪变等其它 DATA_TYPE；
|
||
* origin_*.txt TraceTopic 原始 MMS/报告数据；
|
||
* final_sorted.xls 自动解析生成的工作簿。
|
||
*/
|
||
|
||
using json = nlohmann::json;
|
||
namespace fs = std::filesystem;
|
||
|
||
namespace {
|
||
|
||
fs::path executableDir() {
|
||
wchar_t path[MAX_PATH]{};
|
||
DWORD n = GetModuleFileNameW(nullptr, path, (DWORD)std::size(path));
|
||
if (n == 0 || n >= std::size(path)) return fs::current_path();
|
||
return fs::path(path).parent_path();
|
||
}
|
||
|
||
fs::path appRootDir() {
|
||
fs::path dir = executableDir();
|
||
std::wstring name = dir.filename().wstring();
|
||
std::transform(name.begin(), name.end(), name.begin(), [](wchar_t ch) { return (wchar_t)towlower(ch); });
|
||
// exe 通常在 bin 下运行，runtime/source 与 bin 同级，所以从 bin 回到项目根。
|
||
if (name == L"bin" && dir.has_parent_path()) return dir.parent_path();
|
||
return dir;
|
||
}
|
||
|
||
fs::path runtimeDir() {
|
||
return appRootDir() / "runtime";
|
||
}
|
||
|
||
fs::path runtimeJsonDir() {
|
||
return runtimeDir() / "json";
|
||
}
|
||
|
||
fs::path runtimeLogsDir() {
|
||
return runtimeDir() / "logs";
|
||
}
|
||
|
||
fs::path runtimeJsonPath(const std::string& name) {
|
||
return runtimeJsonDir() / fs::u8path(name);
|
||
}
|
||
|
||
fs::path runtimeLogPath(const std::string& name) {
|
||
return runtimeLogsDir() / fs::u8path(name);
|
||
}
|
||
|
||
std::wstring utf8ToWide(const std::string& s) {
|
||
if (s.empty()) return L"";
|
||
int n = MultiByteToWideChar(CP_UTF8, 0, s.data(), (int)s.size(), nullptr, 0);
|
||
std::wstring out(n, L'\0');
|
||
MultiByteToWideChar(CP_UTF8, 0, s.data(), (int)s.size(), out.data(), n);
|
||
return out;
|
||
}
|
||
|
||
std::string wideToUtf8(const std::wstring& s) {
|
||
if (s.empty()) return "";
|
||
int n = WideCharToMultiByte(CP_UTF8, 0, s.data(), (int)s.size(), nullptr, 0, nullptr, nullptr);
|
||
std::string out(n, '\0');
|
||
WideCharToMultiByte(CP_UTF8, 0, s.data(), (int)s.size(), out.data(), n, nullptr, nullptr);
|
||
return out;
|
||
}
|
||
|
||
std::string jsonString(const json& obj, const char* key) {
|
||
auto it = obj.find(key);
|
||
if (it == obj.end() || it->is_null()) return "";
|
||
if (it->is_string()) return it->get<std::string>();
|
||
if (it->is_number_integer()) return std::to_string(it->get<long long>());
|
||
if (it->is_number_unsigned()) return std::to_string(it->get<unsigned long long>());
|
||
if (it->is_number_float()) return std::to_string(it->get<double>());
|
||
if (it->is_boolean()) return it->get<bool>() ? "true" : "false";
|
||
return it->dump();
|
||
}
|
||
|
||
bool setContains(const std::unordered_set<std::string>& allowed, const std::string& value) {
|
||
return allowed.empty() || allowed.count(value) > 0;
|
||
}
|
||
|
||
std::unordered_set<std::string> parseStatusSet(const std::string& text) {
|
||
std::unordered_set<std::string> out;
|
||
try {
|
||
// 支持 GUI 默认的 JSON 数组写法，例如 "[0]"、"[1,2]"。
|
||
json j = json::parse(text);
|
||
if (j.is_array()) {
|
||
for (const auto& v : j) out.insert(jsonString(json{{"v", v}}, "v"));
|
||
}
|
||
} catch (...) {
|
||
// 也兼容用户手写 "0,1,2" 或带方括号/引号的宽松格式。
|
||
std::stringstream ss(text);
|
||
std::string part;
|
||
while (std::getline(ss, part, ',')) {
|
||
part.erase(std::remove_if(part.begin(), part.end(), [](unsigned char ch) {
|
||
return std::isspace(ch) || ch == '[' || ch == ']' || ch == '"';
|
||
}), part.end());
|
||
if (!part.empty()) out.insert(part);
|
||
}
|
||
}
|
||
return out;
|
||
}
|
||
|
||
std::string sanitizeFilePart(std::string s) {
|
||
for (char& ch : s) {
|
||
if (!(std::isalnum((unsigned char)ch) || ch == '-' || ch == '_' || ch == '.')) ch = '_';
|
||
}
|
||
if (s.empty()) s = "unnamed";
|
||
return s;
|
||
}
|
||
|
||
std::string urlEncode(const std::string& value) {
|
||
static const char* hex = "0123456789ABCDEF";
|
||
std::string out;
|
||
for (unsigned char ch : value) {
|
||
if (std::isalnum(ch) || ch == '-' || ch == '_' || ch == '.' || ch == '~' || ch == '/' || ch == ':') {
|
||
out.push_back((char)ch);
|
||
} else {
|
||
out.push_back('%');
|
||
out.push_back(hex[ch >> 4]);
|
||
out.push_back(hex[ch & 15]);
|
||
}
|
||
}
|
||
return out;
|
||
}
|
||
|
||
std::string httpRequest(const std::string& url,
|
||
const std::string& query,
|
||
const std::string& body,
|
||
PqLogFn log) {
|
||
std::wstring full = utf8ToWide(url + (query.empty() ? "" : ("?" + query)));
|
||
URL_COMPONENTS uc{};
|
||
uc.dwStructSize = sizeof(uc);
|
||
wchar_t host[512]{};
|
||
wchar_t path[4096]{};
|
||
uc.lpszHostName = host;
|
||
uc.dwHostNameLength = (DWORD)std::size(host);
|
||
uc.lpszUrlPath = path;
|
||
uc.dwUrlPathLength = (DWORD)std::size(path);
|
||
uc.dwSchemeLength = (DWORD)-1;
|
||
uc.dwExtraInfoLength = (DWORD)-1;
|
||
|
||
if (!WinHttpCrackUrl(full.c_str(), 0, 0, &uc)) {
|
||
throw std::runtime_error("URL 解析失败: " + url);
|
||
}
|
||
|
||
std::wstring object(path, uc.dwUrlPathLength);
|
||
if (uc.lpszExtraInfo && uc.dwExtraInfoLength > 0) {
|
||
object.append(uc.lpszExtraInfo, uc.dwExtraInfoLength);
|
||
}
|
||
|
||
HINTERNET session = WinHttpOpen(L"pq_tool/1.0",
|
||
WINHTTP_ACCESS_TYPE_DEFAULT_PROXY,
|
||
WINHTTP_NO_PROXY_NAME,
|
||
WINHTTP_NO_PROXY_BYPASS,
|
||
0);
|
||
if (!session) throw std::runtime_error("WinHttpOpen 失败");
|
||
|
||
HINTERNET connect = nullptr;
|
||
HINTERNET request = nullptr;
|
||
std::string response;
|
||
try {
|
||
connect = WinHttpConnect(session, std::wstring(host, uc.dwHostNameLength).c_str(), uc.nPort, 0);
|
||
if (!connect) throw std::runtime_error("WinHttpConnect 失败");
|
||
|
||
DWORD flags = (uc.nScheme == INTERNET_SCHEME_HTTPS) ? WINHTTP_FLAG_SECURE : 0;
|
||
// 这个项目约定：body 为空走 GET，不为空走 POST JSON。
|
||
const wchar_t* method = body.empty() ? L"GET" : L"POST";
|
||
request = WinHttpOpenRequest(connect, method, object.c_str(),
|
||
nullptr, WINHTTP_NO_REFERER,
|
||
WINHTTP_DEFAULT_ACCEPT_TYPES, flags);
|
||
if (!request) throw std::runtime_error("WinHttpOpenRequest 失败");
|
||
|
||
WinHttpSetTimeouts(request, 10000, 10000, 10000, 10000);
|
||
std::wstring headers = L"Content-Type: application/json\r\n";
|
||
BOOL ok = WinHttpSendRequest(
|
||
request,
|
||
headers.c_str(),
|
||
(DWORD)-1,
|
||
body.empty() ? WINHTTP_NO_REQUEST_DATA : (LPVOID)body.data(),
|
||
(DWORD)body.size(),
|
||
(DWORD)body.size(),
|
||
0);
|
||
if (!ok || !WinHttpReceiveResponse(request, nullptr)) {
|
||
throw std::runtime_error("HTTP 请求失败，错误码: " + std::to_string(GetLastError()));
|
||
}
|
||
|
||
DWORD status = 0;
|
||
DWORD len = sizeof(status);
|
||
WinHttpQueryHeaders(request,
|
||
WINHTTP_QUERY_STATUS_CODE | WINHTTP_QUERY_FLAG_NUMBER,
|
||
WINHTTP_HEADER_NAME_BY_INDEX,
|
||
&status,
|
||
&len,
|
||
WINHTTP_NO_HEADER_INDEX);
|
||
if (status < 200 || status >= 300) {
|
||
throw std::runtime_error("HTTP 状态码异常: " + std::to_string(status));
|
||
}
|
||
|
||
DWORD available = 0;
|
||
do {
|
||
if (!WinHttpQueryDataAvailable(request, &available)) {
|
||
throw std::runtime_error("读取 HTTP 响应失败");
|
||
}
|
||
if (!available) break;
|
||
std::string chunk(available, '\0');
|
||
DWORD read = 0;
|
||
if (!WinHttpReadData(request, chunk.data(), available, &read)) {
|
||
throw std::runtime_error("读取 HTTP 响应内容失败");
|
||
}
|
||
chunk.resize(read);
|
||
response += chunk;
|
||
} while (available > 0);
|
||
} catch (...) {
|
||
if (request) WinHttpCloseHandle(request);
|
||
if (connect) WinHttpCloseHandle(connect);
|
||
WinHttpCloseHandle(session);
|
||
throw;
|
||
}
|
||
|
||
if (request) WinHttpCloseHandle(request);
|
||
if (connect) WinHttpCloseHandle(connect);
|
||
WinHttpCloseHandle(session);
|
||
if (log) log("HTTP " + std::string(body.empty() ? "GET " : "POST ") + url + " 返回 " + std::to_string(response.size()) + " 字节");
|
||
return response;
|
||
}
|
||
|
||
void writeTextFile(const fs::path& path, const std::string& text) {
|
||
fs::create_directories(path.parent_path().empty() ? fs::path(".") : path.parent_path());
|
||
std::ofstream ofs(path, std::ios::binary);
|
||
if (!ofs) throw std::runtime_error("无法写入文件: " + path.string());
|
||
ofs << text;
|
||
}
|
||
|
||
json parseResponseDataArray(const std::string& raw, const std::string& name) {
|
||
json root = json::parse(raw);
|
||
if (!root.contains("data") || !root["data"].is_array()) {
|
||
throw std::runtime_error(name + " 响应缺少 data 数组");
|
||
}
|
||
return root["data"];
|
||
}
|
||
|
||
std::vector<std::pair<std::string, std::string>> objectFields(
|
||
const json& obj,
|
||
const std::unordered_set<std::string>& skip = {}) {
|
||
std::vector<std::pair<std::string, std::string>> out;
|
||
if (!obj.is_object()) return out;
|
||
for (auto it = obj.begin(); it != obj.end(); ++it) {
|
||
if (skip.count(it.key())) continue;
|
||
json wrapper;
|
||
wrapper["v"] = it.value();
|
||
out.push_back({it.key(), jsonString(wrapper, "v")});
|
||
}
|
||
return out;
|
||
}
|
||
|
||
bool looksLikeXml(const std::string& text) {
|
||
std::string s = text;
|
||
s.erase(s.begin(), std::find_if(s.begin(), s.end(), [](unsigned char ch) { return !std::isspace(ch); }));
|
||
return s.rfind("<?xml", 0) == 0 || s.rfind("<", 0) == 0;
|
||
}
|
||
|
||
std::string makeGuid() {
|
||
GUID guid;
|
||
if (CoCreateGuid(&guid) != S_OK) {
|
||
auto now = std::chrono::high_resolution_clock::now().time_since_epoch().count();
|
||
return "guid_" + std::to_string(now);
|
||
}
|
||
wchar_t buf[64]{};
|
||
StringFromGUID2(guid, buf, 64);
|
||
std::string s = wideToUtf8(buf);
|
||
s.erase(std::remove(s.begin(), s.end(), '{'), s.end());
|
||
s.erase(std::remove(s.begin(), s.end(), '}'), s.end());
|
||
std::transform(s.begin(), s.end(), s.begin(), [](unsigned char ch) { return (char)std::tolower(ch); });
|
||
return s;
|
||
}
|
||
|
||
std::string nowIsoMillis() {
|
||
using namespace std::chrono;
|
||
auto now = system_clock::now();
|
||
auto ms = duration_cast<milliseconds>(now.time_since_epoch()) % 1000;
|
||
std::time_t tt = system_clock::to_time_t(now);
|
||
std::tm tm{};
|
||
localtime_s(&tm, &tt);
|
||
std::ostringstream oss;
|
||
oss << std::put_time(&tm, "%Y-%m-%dT%H:%M:%S")
|
||
<< "." << std::setw(3) << std::setfill('0') << ms.count();
|
||
return oss.str();
|
||
}
|
||
|
||
long long nowUnixMillis() {
|
||
return std::chrono::duration_cast<std::chrono::milliseconds>(
|
||
std::chrono::system_clock::now().time_since_epoch())
|
||
.count();
|
||
}
|
||
|
||
void appendTextFile(const fs::path& path, const std::string& text) {
|
||
fs::create_directories(path.parent_path().empty() ? fs::path(".") : path.parent_path());
|
||
std::ofstream ofs(path, std::ios::binary | std::ios::app);
|
||
if (!ofs) throw std::runtime_error("无法追加写入文件: " + path.string());
|
||
ofs << text;
|
||
}
|
||
|
||
const char* nonEmptyCStr(const std::string& value, const char* fallback) {
|
||
return value.empty() ? fallback : value.c_str();
|
||
}
|
||
|
||
std::string trimCopy(std::string value) {
|
||
value.erase(value.begin(), std::find_if(value.begin(), value.end(), [](unsigned char ch) {
|
||
return !std::isspace(ch);
|
||
}));
|
||
value.erase(std::find_if(value.rbegin(), value.rend(), [](unsigned char ch) {
|
||
return !std::isspace(ch);
|
||
}).base(), value.end());
|
||
return value;
|
||
}
|
||
|
||
bool isWildcardTag(const std::string& tag) {
|
||
std::string t = trimCopy(tag);
|
||
return t.empty() || t == "*";
|
||
}
|
||
|
||
std::string subscribeTagExpression(const std::string& tag) {
|
||
return isWildcardTag(tag) ? "*" : trimCopy(tag);
|
||
}
|
||
|
||
std::string mergeSubscribeTags(const std::string& left, const std::string& right) {
|
||
std::string a = subscribeTagExpression(left);
|
||
std::string b = subscribeTagExpression(right);
|
||
if (a == "*" || b == "*") return "*";
|
||
if (a == b) return a;
|
||
return a + " || " + b;
|
||
}
|
||
|
||
bool tagMatches(const std::string& expectedExpression, const std::string& actualTag) {
|
||
if (isWildcardTag(expectedExpression)) return true;
|
||
std::string actual = trimCopy(actualTag);
|
||
std::string expr = expectedExpression;
|
||
size_t start = 0;
|
||
while (start <= expr.size()) {
|
||
size_t pos = expr.find("||", start);
|
||
std::string part = trimCopy(expr.substr(start, pos == std::string::npos ? std::string::npos : pos - start));
|
||
// RocketMQ C 接口订阅表达式可能是 "tagA || tagB"，消费时也按同样规则过滤。
|
||
if (part == "*" || part == actual) return true;
|
||
if (pos == std::string::npos) break;
|
||
start = pos + 2;
|
||
}
|
||
return false;
|
||
}
|
||
|
||
long long parseLongLong(const json& value) {
|
||
try {
|
||
if (value.is_number_integer()) return value.get<long long>();
|
||
if (value.is_number_unsigned()) return static_cast<long long>(value.get<unsigned long long>());
|
||
if (value.is_number_float()) return static_cast<long long>(value.get<double>());
|
||
if (value.is_string()) return std::stoll(value.get<std::string>());
|
||
} catch (...) {
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
long long findTimestampValue(const json& value) {
|
||
static const std::unordered_set<std::string> keys = {
|
||
"Value_TIME", "valueTime", "timestamp", "timeStamp", "storeTimestamp", "time", "Time"
|
||
};
|
||
if (value.is_object()) {
|
||
// 不同消息体时间字段命名不统一，因此递归查找一组常见 key。
|
||
for (const auto& key : keys) {
|
||
auto it = value.find(key);
|
||
if (it != value.end()) {
|
||
long long v = parseLongLong(*it);
|
||
if (v > 0) return v;
|
||
}
|
||
}
|
||
for (auto it = value.begin(); it != value.end(); ++it) {
|
||
long long v = findTimestampValue(it.value());
|
||
if (v > 0) return v;
|
||
}
|
||
} else if (value.is_array()) {
|
||
for (const auto& item : value) {
|
||
long long v = findTimestampValue(item);
|
||
if (v > 0) return v;
|
||
}
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
long long timestampFromJsonText(const std::string& text) {
|
||
try {
|
||
return findTimestampValue(json::parse(text));
|
||
} catch (...) {
|
||
return 0;
|
||
}
|
||
}
|
||
|
||
std::string jsonScalarToString(const json& value) {
|
||
if (value.is_string()) return value.get<std::string>();
|
||
if (value.is_number_integer()) return std::to_string(value.get<long long>());
|
||
if (value.is_number_unsigned()) return std::to_string(value.get<unsigned long long>());
|
||
if (value.is_number_float()) {
|
||
std::ostringstream oss;
|
||
oss << value.get<double>();
|
||
return oss.str();
|
||
}
|
||
if (value.is_boolean()) return value.get<bool>() ? "true" : "false";
|
||
return value.is_null() ? "" : value.dump();
|
||
}
|
||
|
||
std::string normalizeDataType(std::string dataType) {
|
||
dataType.erase(std::remove_if(dataType.begin(), dataType.end(),
|
||
[](unsigned char ch) { return std::isspace(ch) != 0; }),
|
||
dataType.end());
|
||
if (dataType.size() == 1 && std::isdigit(static_cast<unsigned char>(dataType[0]))) {
|
||
dataType = "0" + dataType;
|
||
}
|
||
return dataType.empty() ? "unknown" : dataType;
|
||
}
|
||
|
||
bool findDataTypeValue(const json& value, std::string& out) {
|
||
if (value.is_object()) {
|
||
auto it = value.find("DATA_TYPE");
|
||
if (it != value.end()) {
|
||
out = normalizeDataType(jsonScalarToString(*it));
|
||
return true;
|
||
}
|
||
for (auto iter = value.begin(); iter != value.end(); ++iter) {
|
||
if (findDataTypeValue(iter.value(), out)) return true;
|
||
}
|
||
} else if (value.is_array()) {
|
||
for (const auto& item : value) {
|
||
if (findDataTypeValue(item, out)) return true;
|
||
}
|
||
}
|
||
return false;
|
||
}
|
||
|
||
std::string dataTypeFromJsonText(const std::string& text) {
|
||
try {
|
||
std::string dataType;
|
||
if (findDataTypeValue(json::parse(text), dataType)) return dataType;
|
||
} catch (...) {
|
||
}
|
||
return "unknown";
|
||
}
|
||
|
||
bool findReportNameValue(const json& value, std::string& out) {
|
||
static const char* keys[] = {"rpt_id", "rpt_name", "report_name", "rpt_no"};
|
||
if (value.is_object()) {
|
||
for (const char* key : keys) {
|
||
auto it = value.find(key);
|
||
if (it != value.end()) {
|
||
out = trimCopy(jsonScalarToString(*it));
|
||
if (!out.empty()) return true;
|
||
}
|
||
}
|
||
for (auto iter = value.begin(); iter != value.end(); ++iter) {
|
||
if (findReportNameValue(iter.value(), out)) return true;
|
||
}
|
||
} else if (value.is_array()) {
|
||
for (const auto& item : value) {
|
||
if (findReportNameValue(item, out)) return true;
|
||
}
|
||
}
|
||
return false;
|
||
}
|
||
|
||
std::string reportNameFromJsonText(const std::string& text) {
|
||
try {
|
||
std::string reportName;
|
||
if (findReportNameValue(json::parse(text), reportName)) return reportName;
|
||
} catch (...) {
|
||
}
|
||
return "";
|
||
}
|
||
|
||
std::string lowerAscii(std::string value);
|
||
|
||
bool isMonitorIdKey(const std::string& key) {
|
||
std::string k = lowerAscii(trimCopy(key));
|
||
return k == "monitor" ||
|
||
k == "monitorid" ||
|
||
k == "monitor_id" ||
|
||
k == "mp_id" ||
|
||
k == "mpid" ||
|
||
k == "measurepointid" ||
|
||
k == "measure_point_id" ||
|
||
k == "measureid" ||
|
||
k == "measure_id";
|
||
}
|
||
|
||
bool looksLikeJsonText(const std::string& text) {
|
||
std::string s = trimCopy(text);
|
||
return !s.empty() && (s.front() == '{' || s.front() == '[');
|
||
}
|
||
|
||
bool findMonitorIdValue(const json& value, std::string& out) {
|
||
if (value.is_object()) {
|
||
for (auto it = value.begin(); it != value.end(); ++it) {
|
||
if (isMonitorIdKey(it.key())) {
|
||
out = trimCopy(jsonScalarToString(it.value()));
|
||
if (!out.empty()) return true;
|
||
}
|
||
}
|
||
for (auto it = value.begin(); it != value.end(); ++it) {
|
||
if ((it.key() == "messageBody" || it.key() == "body") && it.value().is_string()) {
|
||
// RocketMQ 外层 JSON 的 messageBody 常是字符串化 JSON，需要再 parse 一次。
|
||
std::string embedded = it.value().get<std::string>();
|
||
if (looksLikeJsonText(embedded)) {
|
||
try {
|
||
if (findMonitorIdValue(json::parse(embedded), out)) return true;
|
||
} catch (...) {
|
||
}
|
||
}
|
||
}
|
||
if (findMonitorIdValue(it.value(), out)) return true;
|
||
}
|
||
} else if (value.is_array()) {
|
||
for (const auto& item : value) {
|
||
if (findMonitorIdValue(item, out)) return true;
|
||
}
|
||
}
|
||
return false;
|
||
}
|
||
|
||
std::string monitorIdFromJsonText(const std::string& text) {
|
||
try {
|
||
std::string monitorId;
|
||
if (findMonitorIdValue(json::parse(text), monitorId)) return monitorId;
|
||
} catch (...) {
|
||
}
|
||
return "";
|
||
}
|
||
|
||
bool findGuidValue(const json& value, std::string& out) {
|
||
if (value.is_object()) {
|
||
for (auto it = value.begin(); it != value.end(); ++it) {
|
||
std::string key = lowerAscii(trimCopy(it.key()));
|
||
if (key == "guid" || key == "traceguid" || key == "trace_guid") {
|
||
out = trimCopy(jsonScalarToString(it.value()));
|
||
if (!out.empty()) return true;
|
||
}
|
||
}
|
||
for (auto it = value.begin(); it != value.end(); ++it) {
|
||
if ((it.key() == "messageBody" || it.key() == "body") && it.value().is_string()) {
|
||
std::string embedded = it.value().get<std::string>();
|
||
if (looksLikeJsonText(embedded)) {
|
||
try {
|
||
if (findGuidValue(json::parse(embedded), out)) return true;
|
||
} catch (...) {
|
||
}
|
||
}
|
||
}
|
||
if (findGuidValue(it.value(), out)) return true;
|
||
}
|
||
} else if (value.is_array()) {
|
||
for (const auto& item : value) {
|
||
if (findGuidValue(item, out)) return true;
|
||
}
|
||
}
|
||
return false;
|
||
}
|
||
|
||
std::string guidFromJsonText(const std::string& text) {
|
||
try {
|
||
std::string guid;
|
||
if (findGuidValue(json::parse(text), guid)) return guid;
|
||
json root = json::parse(text);
|
||
if (root.is_object()) {
|
||
auto it = root.find("key");
|
||
if (it != root.end()) return trimCopy(jsonScalarToString(*it));
|
||
}
|
||
} catch (...) {
|
||
}
|
||
return "";
|
||
}
|
||
|
||
std::string normalizeMonitorId(std::string value) {
|
||
value = trimCopy(value);
|
||
value.erase(std::remove_if(value.begin(), value.end(), [](unsigned char ch) {
|
||
return std::isspace(ch) != 0 || ch == '"' || ch == '\'' || ch == '{' || ch == '}';
|
||
}), value.end());
|
||
return lowerAscii(value);
|
||
}
|
||
|
||
std::string safePathPart(std::string value, const std::string& fallback) {
|
||
if (value.empty()) value = fallback;
|
||
for (char& ch : value) {
|
||
unsigned char c = static_cast<unsigned char>(ch);
|
||
if (c < 32 || ch == '<' || ch == '>' || ch == ':' || ch == '"' ||
|
||
ch == '/' || ch == '\\' || ch == '|' || ch == '?' || ch == '*') {
|
||
ch = '_';
|
||
}
|
||
}
|
||
while (!value.empty() && (value.back() == '.' || value.back() == ' ')) value.pop_back();
|
||
if (value.empty()) value = fallback.empty() ? "unnamed" : fallback;
|
||
if (value.size() > 96) value.resize(96);
|
||
return value;
|
||
}
|
||
|
||
std::string firstNonEmpty(const std::string& a, const std::string& b) {
|
||
return !a.empty() ? a : b;
|
||
}
|
||
|
||
std::string prettyJsonOrRaw(const std::string& text) {
|
||
try {
|
||
return json::parse(text).dump(2);
|
||
} catch (...) {
|
||
return text;
|
||
}
|
||
}
|
||
|
||
std::string eventTitle(const std::string& prefix, const std::string& path) {
|
||
fs::path p = fs::u8path(path);
|
||
std::string name = p.filename().u8string();
|
||
return prefix + " " + nowIsoMillis() + (name.empty() ? "" : (" " + name));
|
||
}
|
||
|
||
int existingOriginCount(const fs::path& dir) {
|
||
int count = 0;
|
||
if (!fs::exists(dir)) return 0;
|
||
for (const auto& entry : fs::directory_iterator(dir)) {
|
||
if (!entry.is_regular_file()) continue;
|
||
std::string name = entry.path().filename().u8string();
|
||
if (name.rfind("origin", 0) == 0 && entry.path().extension() == ".txt") ++count;
|
||
}
|
||
return count;
|
||
}
|
||
|
||
std::string lowerAscii(std::string value) {
|
||
std::transform(value.begin(), value.end(), value.begin(),
|
||
[](unsigned char ch) { return static_cast<char>(std::tolower(ch)); });
|
||
return value;
|
||
}
|
||
|
||
bool isTraceTextRecord(const fs::path& path) {
|
||
std::string ext = lowerAscii(path.extension().u8string());
|
||
if (ext == ".txt" || ext == ".json" || ext == ".jsonl" || ext == ".log") return true;
|
||
return false;
|
||
}
|
||
|
||
void clearTraceTextRecords(const fs::path& dir) {
|
||
if (!fs::exists(dir)) return;
|
||
for (const auto& entry : fs::directory_iterator(dir)) {
|
||
if (!entry.is_regular_file()) continue;
|
||
if (isTraceTextRecord(entry.path())) {
|
||
std::error_code ec;
|
||
fs::remove(entry.path(), ec);
|
||
}
|
||
}
|
||
}
|
||
|
||
struct CProducer;
|
||
struct CPushConsumer;
|
||
struct CMessage;
|
||
struct CMessageExt;
|
||
|
||
struct CSendResult {
|
||
int sendStatus;
|
||
char msgId[256];
|
||
long long offset;
|
||
};
|
||
|
||
/*
|
||
* RocketMqRuntime 是进程内唯一的 RocketMQ 适配器。
|
||
*
|
||
* 设计原因：
|
||
* 1. RocketMQ C DLL 通过 LoadLibrary 动态加载，运行时可能不存在；
|
||
* 2. GUI 会多次点击“初始化 MQ”，需要避免重复启动/热重启导致 DLL 崩溃；
|
||
* 3. MQ 回调在 SDK 线程触发，必须用互斥锁保护会话状态，再通过 event_ 通知 GUI。
|
||
*/
|
||
class RocketMqRuntime {
|
||
struct DetachedHandles {
|
||
CProducer* producer = nullptr;
|
||
CPushConsumer* consumer = nullptr;
|
||
HMODULE module = nullptr;
|
||
int (*ShutdownProducer)(CProducer*) = nullptr;
|
||
int (*DestroyProducer)(CProducer*) = nullptr;
|
||
int (*ShutdownPushConsumer)(CPushConsumer*) = nullptr;
|
||
int (*DestroyPushConsumer)(CPushConsumer*) = nullptr;
|
||
};
|
||
|
||
struct TraceSession {
|
||
std::string monitorId;
|
||
std::string monitorName;
|
||
std::string traceDir;
|
||
std::string xmlPath;
|
||
std::string frontType;
|
||
std::string guid;
|
||
long long startedAtMs = 0; // 点击“数据追踪”的时间；更早的 MQ 存量消息不得进入本次会话。
|
||
std::unordered_set<std::string> receivedDataTypes; // 同一次追踪中每个 DATA_TYPE 只保留一次。
|
||
std::unordered_set<std::string> receivedTraceReports; // 同一次追踪中每个 rpt_id 只保留第一条。
|
||
long long dataTimestamp = 0;
|
||
int originCounter = 0;
|
||
};
|
||
|
||
public:
|
||
~RocketMqRuntime() {
|
||
DetachedHandles old;
|
||
{
|
||
std::lock_guard<std::mutex> lock(mu_);
|
||
old = detachLocked();
|
||
}
|
||
shutdownDetached(old);
|
||
}
|
||
|
||
bool initialize(const PqToolConfig& cfg, PqLogFn log, PqMqEventFn event) {
|
||
DetachedHandles old;
|
||
{
|
||
std::lock_guard<std::mutex> lock(mu_);
|
||
// 如果生产/消费连接参数未变，只刷新配置和回调；避免重复 StartProducer/StartPushConsumer。
|
||
bool sameRuntimeConfig = cfg.producer == cfg_.producer &&
|
||
cfg.ipport == cfg_.ipport &&
|
||
cfg.producerAccessKey == cfg_.producerAccessKey &&
|
||
cfg.producerSecretKey == cfg_.producerSecretKey &&
|
||
cfg.consumer == cfg_.consumer &&
|
||
cfg.consumerIpport == cfg_.consumerIpport &&
|
||
cfg.consumerAccessKey == cfg_.consumerAccessKey &&
|
||
cfg.consumerSecretKey == cfg_.consumerSecretKey &&
|
||
cfg.dataStatTopic == cfg_.dataStatTopic &&
|
||
cfg.dataRealtimeTopic == cfg_.dataRealtimeTopic &&
|
||
cfg.dataStatTag == cfg_.dataStatTag &&
|
||
cfg.dataRealtimeTag == cfg_.dataRealtimeTag &&
|
||
cfg.traceTopic == cfg_.traceTopic &&
|
||
cfg.traceTag == cfg_.traceTag;
|
||
if (active_ && producer_ && consumer_ && sameRuntimeConfig) {
|
||
cfg_ = cfg;
|
||
event_ = std::move(event);
|
||
if (log) log("RocketMQ 已初始化，已刷新界面回调和当前配置。");
|
||
return true;
|
||
}
|
||
if (active_ && producer_ && consumer_) {
|
||
// 连接参数变了也不在进程内重启底层 SDK，提示用户重启工具后生效。
|
||
cfg_ = cfg;
|
||
event_ = std::move(event);
|
||
if (log) {
|
||
log("MQ 配置已保存；当前 RocketMQ 连接已运行。");
|
||
log("为避免 RocketMQ SDK 热关闭/重建导致程序崩溃，本次不在进程内重启底层连接。");
|
||
log("如果修改了 nameserver、生产者/消费者组、Topic 或订阅 Tag，请关闭工具后重新打开并初始化 MQ 后生效。");
|
||
}
|
||
return true;
|
||
}
|
||
old = detachLocked();
|
||
}
|
||
|
||
if (old.producer || old.consumer || old.module) {
|
||
if (log) log("MQ 配置变更，正在重新初始化 RocketMQ...");
|
||
shutdownDetachedAsync(old);
|
||
}
|
||
|
||
std::lock_guard<std::mutex> lock(mu_);
|
||
cfg_ = cfg;
|
||
event_ = std::move(event);
|
||
dataTimestamp_ = 0;
|
||
originCounter_ = 0;
|
||
currentTraceDir_.clear();
|
||
currentMonitorId_.clear();
|
||
currentMonitorName_.clear();
|
||
sessions_.clear();
|
||
sessionsByGuid_.clear();
|
||
lastError_.clear();
|
||
|
||
if (!loadLibraryLocked()) {
|
||
if (log) log("RocketMQ DLL 未加载: " + lastError_);
|
||
return false;
|
||
}
|
||
|
||
producer_ = CreateProducer(nonEmptyCStr(cfg_.producer, "TRACE_PRODUCER"));
|
||
if (!producer_) return failLocked("CreateProducer 返回空指针", log);
|
||
|
||
if (!okLocked(SetProducerNameServerAddress(producer_, nonEmptyCStr(cfg_.ipport, "127.0.0.1:9876")),
|
||
"SetProducerNameServerAddress", log)) {
|
||
return false;
|
||
}
|
||
SetProducerSessionCredentials(producer_,
|
||
nonEmptyCStr(cfg_.producerAccessKey, ""),
|
||
nonEmptyCStr(cfg_.producerSecretKey, ""),
|
||
"");
|
||
fs::create_directories(runtimeLogsDir());
|
||
std::string mqLogDir = runtimeLogsDir().u8string();
|
||
SetProducerLogPath(producer_, mqLogDir.c_str());
|
||
if (!okLocked(StartProducer(producer_), "StartProducer", log)) return false;
|
||
|
||
consumer_ = CreatePushConsumer(nonEmptyCStr(cfg_.consumer, "TRACE_CONSUMER"));
|
||
if (!consumer_) return failLocked("CreatePushConsumer 返回空指针", log);
|
||
|
||
if (!okLocked(SetPushConsumerNameServerAddress(consumer_, nonEmptyCStr(cfg_.consumerIpport, "127.0.0.1:9876")),
|
||
"SetPushConsumerNameServerAddress", log)) {
|
||
return false;
|
||
}
|
||
SetPushConsumerSessionCredentials(consumer_,
|
||
nonEmptyCStr(cfg_.consumerAccessKey, ""),
|
||
nonEmptyCStr(cfg_.consumerSecretKey, ""),
|
||
"");
|
||
SetPushConsumerLogPath(consumer_, mqLogDir.c_str());
|
||
SetPushConsumerThreadCount(consumer_, 2);
|
||
|
||
// 同一个 Topic 可能同时配置统计/实时 Tag，这里合并成 RocketMQ 支持的 "A || B" 表达式。
|
||
std::map<std::string, std::string> subscriptions;
|
||
auto addSubscription = [&](const std::string& topic, const std::string& tag) {
|
||
if (topic.empty()) return;
|
||
auto it = subscriptions.find(topic);
|
||
if (it == subscriptions.end()) {
|
||
subscriptions[topic] = subscribeTagExpression(tag);
|
||
} else {
|
||
it->second = mergeSubscribeTags(it->second, tag);
|
||
}
|
||
};
|
||
addSubscription(cfg_.dataStatTopic, cfg_.dataStatTag);
|
||
addSubscription(cfg_.dataRealtimeTopic, cfg_.dataRealtimeTag);
|
||
addSubscription(cfg_.traceTopic, cfg_.traceTag);
|
||
|
||
bool subscribed = false;
|
||
for (const auto& sub : subscriptions) {
|
||
if (!okLocked(Subscribe(consumer_, sub.first.c_str(), sub.second.c_str()),
|
||
"Subscribe Topic", log)) {
|
||
return false;
|
||
}
|
||
subscribed = true;
|
||
if (log) log("Subscribed Topic by Tag only: " + sub.first + " [" + sub.second + "]");
|
||
}
|
||
if (!subscribed) return failLocked("DATATopic/TraceTopic 为空，未订阅任何 Topic", log);
|
||
|
||
if (!okLocked(RegisterMessageCallback(consumer_, &RocketMqRuntime::messageCallback),
|
||
"RegisterMessageCallback", log)) {
|
||
return false;
|
||
}
|
||
if (!okLocked(StartPushConsumer(consumer_), "StartPushConsumer", log)) return false;
|
||
|
||
active_ = true;
|
||
if (log) {
|
||
log("RocketMQ 已连接，生产者和消费者初始化完成。");
|
||
log("DATATopic -> " + (cfg_.jsonDataPath.empty() ? std::string("jsondata.txt") : cfg_.jsonDataPath)
|
||
+ "，TraceTopic -> " + originOutputPath().u8string());
|
||
}
|
||
return true;
|
||
}
|
||
|
||
void setTraceTarget(const PqToolConfig& cfg,
|
||
const PqLedgerDevice& device,
|
||
const PqLedgerMonitor& monitor,
|
||
const std::string& frontType,
|
||
const std::string& guid) {
|
||
const long long startedAtMs = nowUnixMillis();
|
||
std::lock_guard<std::mutex> lock(mu_);
|
||
cfg_ = cfg;
|
||
// 记录当前追踪目标。后续 MQ 回调会用 monitorId 或 guid 把消息路由到对应 TraceSession。
|
||
currentTraceDir_ = pqTraceDirectoryFor(cfg, device, monitor);
|
||
currentXmlPath_ = pqFindDeviceXmlPath(cfg, device);
|
||
currentMonitorId_ = monitor.id;
|
||
currentMonitorName_ = monitor.name;
|
||
fs::path dir = fs::u8path(currentTraceDir_);
|
||
fs::create_directories(dir);
|
||
clearTraceTextRecords(dir);
|
||
dataTimestamp_ = 0;
|
||
originCounter_ = 0;
|
||
std::string monitorKey = normalizeMonitorId(monitor.id);
|
||
// 一次按钮点击只允许一个活动会话。否则旧测点后续到达的消息仍会写入旧会话目录。
|
||
sessions_.clear();
|
||
sessionsByGuid_.clear();
|
||
TraceSession& session = sessions_[monitorKey];
|
||
session.monitorId = monitor.id;
|
||
session.monitorName = monitor.name;
|
||
session.traceDir = currentTraceDir_;
|
||
session.xmlPath = currentXmlPath_;
|
||
session.frontType = frontType;
|
||
session.guid = guid;
|
||
session.startedAtMs = startedAtMs;
|
||
session.receivedDataTypes.clear();
|
||
session.receivedTraceReports.clear();
|
||
session.dataTimestamp = 0;
|
||
session.originCounter = 0;
|
||
if (!guid.empty()) sessionsByGuid_[normalizeMonitorId(guid)] = monitorKey;
|
||
|
||
// GUI 自动解析优先读取这个上下文，避免用户手动维护 json/origin/output 路径。
|
||
json context = {
|
||
{"deviceId", device.id},
|
||
{"deviceName", device.name},
|
||
{"devType", device.devType},
|
||
{"processNo", device.processNo},
|
||
{"monitorId", monitor.id},
|
||
{"monitorName", monitor.name},
|
||
{"guid", guid},
|
||
{"frontType", frontType},
|
||
{"startedAt", nowIsoMillis()},
|
||
{"startedAtMs", startedAtMs},
|
||
{"xmlPath", currentXmlPath_},
|
||
{"jsonDataPath", (dir / "jsondata_01.txt").u8string()},
|
||
{"jsonDataPaths", json::array({
|
||
(dir / "jsondata_01.txt").u8string(),
|
||
(dir / "jsondata_02.txt").u8string(),
|
||
(dir / "jsondata_03.txt").u8string()
|
||
})},
|
||
{"outputPath", (dir / "final_sorted.xls").u8string()}
|
||
};
|
||
writeTextFile(dir / "trace_context.json", context.dump(2));
|
||
}
|
||
|
||
bool sendPayload(const std::string& topic,
|
||
const std::string& tag,
|
||
const std::string& messageKey,
|
||
const std::string& payload,
|
||
const std::string& label,
|
||
std::string& err,
|
||
PqLogFn log) {
|
||
std::lock_guard<std::mutex> lock(mu_);
|
||
err.clear();
|
||
if (!active_ || !producer_) {
|
||
err = lastError_.empty() ? "RocketMQ 未初始化" : lastError_;
|
||
return false;
|
||
}
|
||
|
||
CMessage* msg = CreateMessage(nonEmptyCStr(topic, ""));
|
||
if (!msg) {
|
||
err = "CreateMessage 返回空指针";
|
||
return false;
|
||
}
|
||
|
||
SetMessageTags(msg, nonEmptyCStr(tag, ""));
|
||
SetMessageKeys(msg, nonEmptyCStr(messageKey, ""));
|
||
SetMessageBody(msg, payload.c_str());
|
||
|
||
CSendResult result{};
|
||
int rc = SendMessageSync(producer_, msg, &result);
|
||
DestroyMessage(msg);
|
||
if (rc != 0) {
|
||
err = "SendMessageSync 返回错误码 " + std::to_string(rc);
|
||
return false;
|
||
}
|
||
|
||
if (log) {
|
||
log("RocketMQ " + label + "消息已发送，Topic=" + topic
|
||
+ "，MsgId=" + std::string(result.msgId));
|
||
}
|
||
return true;
|
||
}
|
||
|
||
private:
|
||
using MessageCallback = int (*)(CPushConsumer*, CMessageExt*);
|
||
|
||
static int messageCallback(CPushConsumer*, CMessageExt* msg) {
|
||
return instance().onMessage(msg);
|
||
}
|
||
|
||
TraceSession* findSessionForMessageLocked(const std::string& body,
|
||
const std::string& key,
|
||
std::string& actualMonitor,
|
||
std::string& reason) {
|
||
actualMonitor.clear();
|
||
reason.clear();
|
||
|
||
// 优先用消息体中的测点 ID 路由；没有测点 ID 时再用 guid/key 路由。
|
||
actualMonitor = monitorIdFromJsonText(body);
|
||
std::string normalized = normalizeMonitorId(actualMonitor);
|
||
if (!normalized.empty()) {
|
||
auto it = sessions_.find(normalized);
|
||
if (it != sessions_.end()) return &it->second;
|
||
reason = "monitor not being traced, actual=" + actualMonitor;
|
||
}
|
||
|
||
std::string guid = guidFromJsonText(body);
|
||
if (guid.empty()) guid = trimCopy(key);
|
||
std::string normalizedGuid = normalizeMonitorId(guid);
|
||
if (!normalizedGuid.empty()) {
|
||
auto git = sessionsByGuid_.find(normalizedGuid);
|
||
if (git != sessionsByGuid_.end()) {
|
||
auto sit = sessions_.find(git->second);
|
||
if (sit != sessions_.end()) return &sit->second;
|
||
}
|
||
if (reason.empty()) reason = "guid not being traced, actual=" + guid;
|
||
}
|
||
|
||
if (reason.empty()) reason = "missing monitor id/guid";
|
||
return nullptr;
|
||
}
|
||
|
||
int onMessage(CMessageExt* msg) {
|
||
std::string topic = GetMessageTopic && GetMessageTopic(msg) ? GetMessageTopic(msg) : "";
|
||
std::string tag = GetMessageTags && GetMessageTags(msg) ? GetMessageTags(msg) : "";
|
||
std::string key = GetMessageKeys && GetMessageKeys(msg) ? GetMessageKeys(msg) : "";
|
||
std::string body = GetMessageBody && GetMessageBody(msg) ? GetMessageBody(msg) : "";
|
||
long long storeTs = GetMessageStoreTimestamp ? GetMessageStoreTimestamp(msg) : 0;
|
||
|
||
std::lock_guard<std::mutex> lock(mu_);
|
||
try {
|
||
appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis()
|
||
+ " received topic=" + topic
|
||
+ " tag=" + tag
|
||
+ " key=" + key
|
||
+ " bodyLen=" + std::to_string(body.size()) + "\n");
|
||
// 先按 Topic/Tag 判断消息类型，再按 monitorId/guid 过滤到当前追踪会话。
|
||
std::string dataFrontType = dataFrontTypeForMessageLocked(topic, tag);
|
||
bool isData = !dataFrontType.empty();
|
||
bool isTrace = topic == cfg_.traceTopic && tagMatches(cfg_.traceTag, tag);
|
||
TraceSession* session = nullptr;
|
||
if (isData || isTrace) {
|
||
std::string actualMonitor;
|
||
std::string reason;
|
||
session = findSessionForMessageLocked(body, key, actualMonitor, reason);
|
||
if (!session) {
|
||
appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis()
|
||
+ " skipped " + (isData ? "DATATopic" : "TraceTopic")
|
||
+ " by monitor router: " + reason
|
||
+ " topic=" + topic
|
||
+ " tag=" + tag + "\n");
|
||
return 0;
|
||
}
|
||
if (isData && !session->frontType.empty() && session->frontType != dataFrontType) {
|
||
appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis()
|
||
+ " skipped DATATopic by frontType router: session="
|
||
+ session->frontType + ", message=" + dataFrontType
|
||
+ " topic=" + topic + " tag=" + tag + "\n");
|
||
return 0;
|
||
}
|
||
// PushConsumer 可能在点击后继续投递积压消息。必须在任何去重、落盘和 GUI 事件之前
|
||
// 使用 Broker 存储时间做会话边界判断，保证 jsondata 与 origin 都来自本次追踪。
|
||
if (storeTs <= 0 || storeTs < session->startedAtMs) {
|
||
appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis()
|
||
+ " skipped " + (isData ? "DATATopic" : "TraceTopic")
|
||
+ " before trace session: storeTs=" + std::to_string(storeTs)
|
||
+ ", startedAtMs=" + std::to_string(session->startedAtMs)
|
||
+ ", monitor=" + session->monitorId
|
||
+ " topic=" + topic + " tag=" + tag + "\n");
|
||
return 0;
|
||
}
|
||
}
|
||
if (isData) {
|
||
std::string dataType = dataTypeFromJsonText(body);
|
||
std::string normalizedType = normalizeDataType(dataType);
|
||
// 同一 DATA_TYPE 重复到达时跳过，避免后来的旧数据覆盖已解析文件。
|
||
if (!session->receivedDataTypes.insert(normalizedType).second) {
|
||
appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis()
|
||
+ " skipped DATATopic duplicate DATA_TYPE=" + normalizedType
|
||
+ " monitor=" + session->monitorId
|
||
+ " topic=" + topic + " tag=" + tag + "\n");
|
||
return 0;
|
||
}
|
||
fs::path out = jsonDataOutputPathLocked(*session, dataType);
|
||
writeTextFile(out, body);
|
||
long long bodyTs = timestampFromJsonText(body);
|
||
session->dataTimestamp = bodyTs > 0 ? bodyTs : storeTs;
|
||
if (normalizeMonitorId(session->monitorId) == normalizeMonitorId(currentMonitorId_)) {
|
||
dataTimestamp_ = session->dataTimestamp;
|
||
}
|
||
appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis() + " DATATopic DATA_TYPE=" + dataType + " tag=" + tag + " -> " + out.u8string() + "\n");
|
||
if (event_) event_({"data", eventTitle("DATATopic", out.u8string()), body, out.u8string(), session->monitorId, session->traceDir});
|
||
} else if (isTrace) {
|
||
long long bodyTs = timestampFromJsonText(body);
|
||
// 如果 TraceTopic 的时间早于已经收到的 DATATopic，认为是旧追踪结果，避免串入本次追踪。
|
||
if (session->dataTimestamp > 0 && bodyTs > 0 && bodyTs < session->dataTimestamp) {
|
||
appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis() + " TraceTopic tag=" + tag + " skipped by timestamp\n");
|
||
} else {
|
||
std::string reportName = reportNameFromJsonText(body);
|
||
std::string reportKey = lowerAscii(trimCopy(reportName));
|
||
if (reportKey.empty()) reportKey = "__unknown_report__";
|
||
// TopicAsk 的 limit=20 可能返回同一 rpt_id 的多份历史采样；本工具每次点击只取第一份。
|
||
if (!session->receivedTraceReports.insert(reportKey).second) {
|
||
appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis()
|
||
+ " skipped TraceTopic duplicate report=" + reportKey
|
||
+ " monitor=" + session->monitorId
|
||
+ " topic=" + topic + " tag=" + tag + "\n");
|
||
return 0;
|
||
}
|
||
fs::path out = nextOriginOutputPathLocked(*session, reportName);
|
||
writeTextFile(out, body);
|
||
appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis() + " TraceTopic tag=" + tag + " -> " + out.u8string() + "\n");
|
||
if (event_) event_({"trace", eventTitle("TraceTopic", out.u8string()), body, out.u8string(), session->monitorId, session->traceDir});
|
||
}
|
||
} else {
|
||
appendTextFile(runtimeLogPath("mq_messages.log"), nowIsoMillis() + " skipped topic=" + topic + " tag=" + tag + "\n");
|
||
}
|
||
} catch (...) {
|
||
return 1;
|
||
}
|
||
return 0;
|
||
}
|
||
|
||
static RocketMqRuntime& instance() {
|
||
static RocketMqRuntime runtime;
|
||
return runtime;
|
||
}
|
||
|
||
friend RocketMqRuntime& rocketRuntime();
|
||
|
||
fs::path originOutputPath() const {
|
||
if (!cfg_.originPaths.empty() && !cfg_.originPaths.front().empty()) {
|
||
return fs::u8path(cfg_.originPaths.front());
|
||
}
|
||
return fs::u8path("origin.txt");
|
||
}
|
||
|
||
std::string dataFrontTypeForMessageLocked(const std::string& topic, const std::string& tag) const {
|
||
if (topic == cfg_.dataStatTopic && tagMatches(cfg_.dataStatTag, tag)) return "cfg_stat_data";
|
||
if (topic == cfg_.dataRealtimeTopic && tagMatches(cfg_.dataRealtimeTag, tag)) return "cfg_3s_data";
|
||
return "";
|
||
}
|
||
|
||
fs::path jsonDataOutputPathLocked(const TraceSession& session, const std::string& dataType) const {
|
||
std::string suffix = normalizeDataType(dataType);
|
||
std::string filename = "jsondata_" + safePathPart(suffix, "unknown") + ".txt";
|
||
if (!session.traceDir.empty()) return fs::u8path(session.traceDir) / fs::u8path(filename);
|
||
if (!currentTraceDir_.empty()) return fs::u8path(currentTraceDir_) / fs::u8path(filename);
|
||
if (!cfg_.jsonDataPath.empty()) {
|
||
std::vector<std::string> paths;
|
||
std::stringstream ss(cfg_.jsonDataPath);
|
||
std::string part;
|
||
while (std::getline(ss, part)) {
|
||
part = trimCopy(part);
|
||
if (!part.empty()) paths.push_back(part);
|
||
}
|
||
fs::path base = paths.empty() ? fs::u8path(cfg_.jsonDataPath) : fs::u8path(paths.front());
|
||
fs::path dir = base.has_parent_path() ? base.parent_path() : fs::path(".");
|
||
return dir / fs::u8path(filename);
|
||
}
|
||
return runtimeJsonPath(filename);
|
||
}
|
||
|
||
fs::path nextOriginOutputPathLocked(TraceSession& session, const std::string& reportName) {
|
||
if (session.traceDir.empty() && currentTraceDir_.empty()) return originOutputPath();
|
||
std::string safeReport = reportName.empty() ? "" : safePathPart(reportName, "report");
|
||
std::string baseName;
|
||
if (safeReport.empty()) {
|
||
++session.originCounter;
|
||
baseName = "origin" + std::to_string(session.originCounter);
|
||
} else {
|
||
baseName = "origin_" + safeReport;
|
||
}
|
||
|
||
fs::path dir = session.traceDir.empty() ? fs::u8path(currentTraceDir_) : fs::u8path(session.traceDir);
|
||
fs::path out = dir / fs::u8path(baseName + ".txt");
|
||
int suffix = 2;
|
||
while (fs::exists(out)) {
|
||
out = dir / fs::u8path(baseName + "_" + std::to_string(suffix++) + ".txt");
|
||
}
|
||
return out;
|
||
}
|
||
|
||
bool failLocked(const std::string& message, PqLogFn log) {
|
||
lastError_ = message;
|
||
shutdownLocked();
|
||
if (log) log("RocketMQ 初始化失败: " + message);
|
||
return false;
|
||
}
|
||
|
||
bool okLocked(int rc, const char* op, PqLogFn log) {
|
||
if (rc == 0) return true;
|
||
return failLocked(std::string(op) + " 返回错误码 " + std::to_string(rc), log);
|
||
}
|
||
|
||
bool loadLibraryLocked() {
|
||
if (module_) return true;
|
||
|
||
// 兼容不同部署目录和 DLL 命名；LoadLibraryEx 用于带路径 DLL 时顺带搜索其依赖。
|
||
std::vector<std::string> candidates = {
|
||
"rocketmq-client-cpp.dll",
|
||
"rocketmq.dll",
|
||
"librocketmq.dll",
|
||
"bin\\rocketmq-client-cpp.dll",
|
||
"bin\\rocketmq.dll",
|
||
"bin\\librocketmq.dll",
|
||
"rocketmq-client-cpp-2.2.0-source-release\\bin\\rocketmq-client-cpp.dll",
|
||
"rocketmq-client-cpp-2.2.0-source-release\\bin\\rocketmq.dll",
|
||
"rocketmq-client-cpp-2.2.0-source-release\\bin\\librocketmq.dll",
|
||
"Win32\\Debug\\rocketmq-client-cpp.dll",
|
||
"Win32\\Release\\rocketmq-client-cpp.dll",
|
||
"Win32\\x86\\Debug\\rocketmq-client-cpp.dll",
|
||
"Win32\\x86\\Release\\rocketmq-client-cpp.dll",
|
||
"rocketmq-client-cpp-2.2.0-source-release\\Win32\\Debug\\rocketmq-client-cpp.dll",
|
||
"rocketmq-client-cpp-2.2.0-source-release\\Win32\\Release\\rocketmq-client-cpp.dll",
|
||
"rocketmq-client-cpp-2.2.0-source-release\\Win32\\x86\\Debug\\rocketmq-client-cpp.dll",
|
||
"rocketmq-client-cpp-2.2.0-source-release\\Win32\\x86\\Release\\rocketmq-client-cpp.dll",
|
||
"Win32\\x64\\Debug\\rocketmq-client-cpp.dll",
|
||
"Win32\\x64\\Release\\rocketmq-client-cpp.dll",
|
||
"rocketmq-client-cpp-2.2.0-source-release\\Win32\\x64\\Debug\\rocketmq-client-cpp.dll",
|
||
"rocketmq-client-cpp-2.2.0-source-release\\Win32\\x64\\Release\\rocketmq-client-cpp.dll"
|
||
};
|
||
|
||
DWORD lastErr = 0;
|
||
std::string tried;
|
||
for (const auto& path : candidates) {
|
||
bool exists = fs::exists(fs::u8path(path));
|
||
module_ = path.find('\\') == std::string::npos
|
||
? LoadLibraryA(path.c_str())
|
||
: LoadLibraryExA(path.c_str(), nullptr, LOAD_WITH_ALTERED_SEARCH_PATH);
|
||
if (module_) {
|
||
loadedFrom_ = path;
|
||
break;
|
||
}
|
||
lastErr = GetLastError();
|
||
if (!tried.empty()) tried += ", ";
|
||
tried += path;
|
||
tried += exists ? "(load error=" + std::to_string(lastErr) + ")" : "(not found)";
|
||
}
|
||
|
||
if (!module_) {
|
||
lastError_ = "No loadable RocketMQ DLL. Tried: " + tried
|
||
+ ". LastError=" + std::to_string(lastErr)
|
||
+ ". Error 126 means the DLL itself or one of its dependent DLLs is missing. "
|
||
"This pq_tool.exe is 32-bit, so RocketMQ must be built as x86.";
|
||
return false;
|
||
lastError_ = "未找到可加载的 RocketMQ DLL，已尝试: " + tried
|
||
+ "，GetLastError=" + std::to_string(lastErr)
|
||
+ "。当前 exe 是 32 位，DLL 也需要是 x86。";
|
||
return false;
|
||
}
|
||
|
||
std::vector<std::string> missing;
|
||
#define LOAD_MQ_FN(name) \
|
||
do { \
|
||
name = reinterpret_cast<decltype(name)>(GetProcAddress(module_, #name)); \
|
||
if (!name) missing.push_back(#name); \
|
||
} while (0)
|
||
LOAD_MQ_FN(CreateProducer);
|
||
LOAD_MQ_FN(DestroyProducer);
|
||
LOAD_MQ_FN(StartProducer);
|
||
LOAD_MQ_FN(ShutdownProducer);
|
||
LOAD_MQ_FN(SetProducerNameServerAddress);
|
||
LOAD_MQ_FN(SetProducerSessionCredentials);
|
||
LOAD_MQ_FN(SetProducerLogPath);
|
||
LOAD_MQ_FN(CreateMessage);
|
||
LOAD_MQ_FN(DestroyMessage);
|
||
LOAD_MQ_FN(SetMessageTags);
|
||
LOAD_MQ_FN(SetMessageKeys);
|
||
LOAD_MQ_FN(SetMessageBody);
|
||
LOAD_MQ_FN(SendMessageSync);
|
||
LOAD_MQ_FN(CreatePushConsumer);
|
||
LOAD_MQ_FN(DestroyPushConsumer);
|
||
LOAD_MQ_FN(StartPushConsumer);
|
||
LOAD_MQ_FN(ShutdownPushConsumer);
|
||
LOAD_MQ_FN(SetPushConsumerNameServerAddress);
|
||
LOAD_MQ_FN(SetPushConsumerSessionCredentials);
|
||
LOAD_MQ_FN(SetPushConsumerLogPath);
|
||
LOAD_MQ_FN(SetPushConsumerThreadCount);
|
||
LOAD_MQ_FN(Subscribe);
|
||
LOAD_MQ_FN(RegisterMessageCallback);
|
||
LOAD_MQ_FN(GetMessageTopic);
|
||
LOAD_MQ_FN(GetMessageTags);
|
||
LOAD_MQ_FN(GetMessageKeys);
|
||
LOAD_MQ_FN(GetMessageBody);
|
||
LOAD_MQ_FN(GetMessageStoreTimestamp);
|
||
#undef LOAD_MQ_FN
|
||
|
||
if (!missing.empty()) {
|
||
std::ostringstream oss;
|
||
for (size_t i = 0; i < missing.size(); ++i) {
|
||
if (i) oss << ", ";
|
||
oss << missing[i];
|
||
}
|
||
lastError_ = "RocketMQ DLL 缺少 C 接口导出: " + oss.str();
|
||
FreeLibrary(module_);
|
||
module_ = nullptr;
|
||
return false;
|
||
}
|
||
|
||
return true;
|
||
}
|
||
|
||
DetachedHandles detachLocked() {
|
||
DetachedHandles h;
|
||
h.producer = producer_;
|
||
h.consumer = consumer_;
|
||
h.module = module_;
|
||
h.ShutdownProducer = ShutdownProducer;
|
||
h.DestroyProducer = DestroyProducer;
|
||
h.ShutdownPushConsumer = ShutdownPushConsumer;
|
||
h.DestroyPushConsumer = DestroyPushConsumer;
|
||
active_ = false;
|
||
producer_ = nullptr;
|
||
consumer_ = nullptr;
|
||
module_ = nullptr;
|
||
loadedFrom_.clear();
|
||
return h;
|
||
}
|
||
|
||
void shutdownDetached(const DetachedHandles& h) {
|
||
if (h.consumer) {
|
||
if (h.ShutdownPushConsumer) h.ShutdownPushConsumer(h.consumer);
|
||
if (h.DestroyPushConsumer) h.DestroyPushConsumer(h.consumer);
|
||
}
|
||
if (h.producer) {
|
||
if (h.ShutdownProducer) h.ShutdownProducer(h.producer);
|
||
if (h.DestroyProducer) h.DestroyProducer(h.producer);
|
||
}
|
||
if (h.module) FreeLibrary(h.module);
|
||
}
|
||
|
||
void shutdownDetachedAsync(DetachedHandles h) {
|
||
std::thread([h]() {
|
||
try {
|
||
if (h.consumer) {
|
||
if (h.ShutdownPushConsumer) h.ShutdownPushConsumer(h.consumer);
|
||
if (h.DestroyPushConsumer) h.DestroyPushConsumer(h.consumer);
|
||
}
|
||
if (h.producer) {
|
||
if (h.ShutdownProducer) h.ShutdownProducer(h.producer);
|
||
if (h.DestroyProducer) h.DestroyProducer(h.producer);
|
||
}
|
||
if (h.module) FreeLibrary(h.module);
|
||
} catch (...) {
|
||
}
|
||
}).detach();
|
||
}
|
||
|
||
void shutdownLocked() {
|
||
active_ = false;
|
||
if (consumer_) {
|
||
ShutdownPushConsumer(consumer_);
|
||
DestroyPushConsumer(consumer_);
|
||
consumer_ = nullptr;
|
||
}
|
||
if (producer_) {
|
||
ShutdownProducer(producer_);
|
||
DestroyProducer(producer_);
|
||
producer_ = nullptr;
|
||
}
|
||
if (module_) {
|
||
FreeLibrary(module_);
|
||
module_ = nullptr;
|
||
}
|
||
}
|
||
|
||
std::mutex mu_;
|
||
PqToolConfig cfg_;
|
||
PqMqEventFn event_;
|
||
HMODULE module_ = nullptr;
|
||
std::string loadedFrom_;
|
||
std::string lastError_;
|
||
std::string currentTraceDir_;
|
||
std::string currentXmlPath_;
|
||
std::string currentMonitorId_;
|
||
std::string currentMonitorName_;
|
||
std::unordered_map<std::string, TraceSession> sessions_;
|
||
std::unordered_map<std::string, std::string> sessionsByGuid_;
|
||
bool active_ = false;
|
||
long long dataTimestamp_ = 0;
|
||
int originCounter_ = 0;
|
||
CProducer* producer_ = nullptr;
|
||
CPushConsumer* consumer_ = nullptr;
|
||
|
||
CProducer* (*CreateProducer)(const char*) = nullptr;
|
||
int (*DestroyProducer)(CProducer*) = nullptr;
|
||
int (*StartProducer)(CProducer*) = nullptr;
|
||
int (*ShutdownProducer)(CProducer*) = nullptr;
|
||
int (*SetProducerNameServerAddress)(CProducer*, const char*) = nullptr;
|
||
int (*SetProducerSessionCredentials)(CProducer*, const char*, const char*, const char*) = nullptr;
|
||
int (*SetProducerLogPath)(CProducer*, const char*) = nullptr;
|
||
CMessage* (*CreateMessage)(const char*) = nullptr;
|
||
int (*DestroyMessage)(CMessage*) = nullptr;
|
||
int (*SetMessageTags)(CMessage*, const char*) = nullptr;
|
||
int (*SetMessageKeys)(CMessage*, const char*) = nullptr;
|
||
int (*SetMessageBody)(CMessage*, const char*) = nullptr;
|
||
int (*SendMessageSync)(CProducer*, CMessage*, CSendResult*) = nullptr;
|
||
CPushConsumer* (*CreatePushConsumer)(const char*) = nullptr;
|
||
int (*DestroyPushConsumer)(CPushConsumer*) = nullptr;
|
||
int (*StartPushConsumer)(CPushConsumer*) = nullptr;
|
||
int (*ShutdownPushConsumer)(CPushConsumer*) = nullptr;
|
||
int (*SetPushConsumerNameServerAddress)(CPushConsumer*, const char*) = nullptr;
|
||
int (*SetPushConsumerSessionCredentials)(CPushConsumer*, const char*, const char*, const char*) = nullptr;
|
||
int (*SetPushConsumerLogPath)(CPushConsumer*, const char*) = nullptr;
|
||
int (*SetPushConsumerThreadCount)(CPushConsumer*, int) = nullptr;
|
||
int (*Subscribe)(CPushConsumer*, const char*, const char*) = nullptr;
|
||
int (*RegisterMessageCallback)(CPushConsumer*, MessageCallback) = nullptr;
|
||
const char* (*GetMessageTopic)(CMessageExt*) = nullptr;
|
||
const char* (*GetMessageTags)(CMessageExt*) = nullptr;
|
||
const char* (*GetMessageKeys)(CMessageExt*) = nullptr;
|
||
const char* (*GetMessageBody)(CMessageExt*) = nullptr;
|
||
long long (*GetMessageStoreTimestamp)(CMessageExt*) = nullptr;
|
||
};
|
||
|
||
RocketMqRuntime& rocketRuntime() {
|
||
return RocketMqRuntime::instance();
|
||
}
|
||
|
||
} // namespace
|
||
|
||
PqToolConfig pqDefaultConfig() {
|
||
PqToolConfig cfg;
|
||
// 这里只覆盖需要根据 exe 位置动态计算的路径，以及 MQ 常用默认参数。
|
||
cfg.producer = "TRACE_PRODUCER";
|
||
cfg.consumer = "TRACE_CONSUMER";
|
||
cfg.dataStatTopic = "MQ_INST_1697676490574298_Bm3DLamM%LN_Topic";
|
||
cfg.dataRealtimeTopic = "MQ_INST_1697676490574298_Bm3DLamM%Real_Time_Data_Topic";
|
||
cfg.dataStatTag = "6c2adb324940e19e208d717f4970f2ee";
|
||
cfg.dataStatKey = "Test_Keys";
|
||
cfg.dataRealtimeTag = "6c2adb324940e19e208d717f4970f2ee";
|
||
cfg.dataRealtimeKey = "Test_Keys";
|
||
cfg.dataTopic = cfg.dataStatTopic;
|
||
cfg.dataTag = cfg.dataStatTag;
|
||
cfg.dataKey = cfg.dataStatKey;
|
||
cfg.xmlSaveDir = (runtimeDir() / "downloaded_xml").u8string();
|
||
cfg.traceRootDir = (runtimeDir() / "trace_data").u8string();
|
||
cfg.jsonDataPath = runtimeJsonPath("jsondata.txt").u8string();
|
||
cfg.outputPath = (runtimeDir() / "final_sorted.xls").u8string();
|
||
return cfg;
|
||
}
|
||
|
||
std::string pqTraceDirectoryFor(const PqToolConfig& cfg,
|
||
const PqLedgerDevice& device,
|
||
const PqLedgerMonitor& monitor) {
|
||
fs::path root = fs::u8path(cfg.traceRootDir.empty() ? "trace_data" : cfg.traceRootDir);
|
||
// 路径中同时放名称和 id，既方便人读，也保证同名测点不互相覆盖。
|
||
std::string proc = safePathPart(firstNonEmpty(device.processNo, "1"), "1");
|
||
std::string dev = safePathPart(firstNonEmpty(device.name, device.id), "device");
|
||
std::string mon = safePathPart(firstNonEmpty(monitor.name, monitor.id), "monitor");
|
||
if (!device.id.empty()) dev += "_" + safePathPart(device.id, "id");
|
||
if (!monitor.id.empty()) mon += "_" + safePathPart(monitor.id, "id");
|
||
fs::path dir = root / fs::u8path("process_" + proc) /
|
||
fs::u8path("device_" + dev) /
|
||
fs::u8path("monitor_" + mon);
|
||
return dir.u8string();
|
||
}
|
||
|
||
void pqClearTraceRecords(const PqToolConfig& cfg,
|
||
const PqLedgerDevice& device,
|
||
const PqLedgerMonitor& monitor) {
|
||
fs::path dir = fs::u8path(pqTraceDirectoryFor(cfg, device, monitor));
|
||
fs::create_directories(dir);
|
||
clearTraceTextRecords(dir);
|
||
}
|
||
|
||
std::string pqFindDeviceXmlPath(const PqToolConfig& cfg, const PqLedgerDevice& device) {
|
||
if (!device.xmlPath.empty() && fs::exists(fs::u8path(device.xmlPath))) return device.xmlPath;
|
||
|
||
// 下载 ICD 后会把接口响应缓存到 runtime\json\icd_response.json，这里用 devType 反查本地 XML。
|
||
fs::path saveDir = fs::u8path(cfg.xmlSaveDir.empty() ? "downloaded_xml" : cfg.xmlSaveDir);
|
||
auto existing = [&](const std::string& name) -> std::string {
|
||
if (name.empty()) return "";
|
||
fs::path p = saveDir / fs::u8path(name);
|
||
if (fs::exists(p)) return p.u8string();
|
||
return "";
|
||
};
|
||
|
||
fs::path icdCache = runtimeJsonPath("icd_response.json");
|
||
if (fs::exists(icdCache)) {
|
||
try {
|
||
std::string raw;
|
||
{
|
||
std::ifstream ifs(icdCache, std::ios::binary);
|
||
std::ostringstream ss;
|
||
ss << ifs.rdbuf();
|
||
raw = ss.str();
|
||
}
|
||
json arr = parseResponseDataArray(raw, "ICD");
|
||
for (const auto& model : arr) {
|
||
if (jsonString(model, "devType") != device.devType) continue;
|
||
std::string byId = existing(jsonString(model, "id") + ".xml");
|
||
if (!byId.empty()) return byId;
|
||
std::string fileName = jsonString(model, "fileName");
|
||
std::string byFile = existing(fileName + ".xml");
|
||
if (!byFile.empty()) return byFile;
|
||
}
|
||
} catch (...) {
|
||
}
|
||
}
|
||
|
||
std::vector<std::string> guesses = {
|
||
device.devType + ".xml",
|
||
device.devKey + ".xml",
|
||
device.id + ".xml"
|
||
};
|
||
for (const auto& g : guesses) {
|
||
std::string p = existing(g);
|
||
if (!p.empty()) return p;
|
||
}
|
||
if (!cfg.xmlPath.empty() && fs::exists(fs::u8path(cfg.xmlPath))) return cfg.xmlPath;
|
||
return "";
|
||
}
|
||
|
||
bool pqFetchLedgerAndIcd(const PqToolConfig& cfg,
|
||
std::vector<PqLedgerDevice>& devices,
|
||
std::string& firstXmlPath,
|
||
PqLogFn log) {
|
||
devices.clear();
|
||
firstXmlPath.clear();
|
||
|
||
// 1. 请求台账：frontIp + terminalStatus/runFlag。
|
||
json ledgerReq;
|
||
ledgerReq["ip"] = cfg.frontIp;
|
||
try {
|
||
ledgerReq["runFlag"] = json::parse(cfg.terminalStatus);
|
||
} catch (...) {
|
||
ledgerReq["runFlag"] = cfg.terminalStatus;
|
||
}
|
||
|
||
if (log) log("获取台账: " + ledgerReq.dump());
|
||
std::string ledgerRaw = httpRequest(cfg.webDevice, "", ledgerReq.dump(), log);
|
||
writeTextFile(runtimeJsonPath("ledger_response.json"), ledgerRaw);
|
||
|
||
std::unordered_set<std::string> terminalAllowed = parseStatusSet(cfg.terminalStatus);
|
||
std::set<std::string> devTypes;
|
||
|
||
// 2. 解析装置和 monitorData 测点，并按 terminalStatus 过滤装置。
|
||
json data = parseResponseDataArray(ledgerRaw, "台账");
|
||
for (const auto& item : data) {
|
||
PqLedgerDevice dev;
|
||
dev.id = jsonString(item, "id");
|
||
dev.name = jsonString(item, "name");
|
||
dev.ip = jsonString(item, "ip");
|
||
dev.status = jsonString(item, "status");
|
||
dev.devType = jsonString(item, "devType");
|
||
dev.devKey = jsonString(item, "devKey");
|
||
dev.orgName = jsonString(item, "org_name");
|
||
dev.stationName = jsonString(item, "stationName");
|
||
dev.subName = jsonString(item, "subName");
|
||
dev.manufacturer = jsonString(item, "manufacturer");
|
||
dev.port = jsonString(item, "port");
|
||
dev.series = jsonString(item, "series");
|
||
dev.updateTime = jsonString(item, "updateTime");
|
||
dev.processNo = jsonString(item, "processNo");
|
||
dev.maxProcessNum = jsonString(item, "maxProcessNum");
|
||
dev.fields = objectFields(item, {"monitorData"});
|
||
if (!setContains(terminalAllowed, dev.status)) continue;
|
||
if (!dev.devType.empty()) devTypes.insert(dev.devType);
|
||
|
||
auto monitors = item.find("monitorData");
|
||
if (monitors != item.end() && monitors->is_array()) {
|
||
for (const auto& m : *monitors) {
|
||
PqLedgerMonitor mon;
|
||
mon.id = jsonString(m, "id");
|
||
mon.name = jsonString(m, "name");
|
||
mon.lineNo = jsonString(m, "lineNo");
|
||
mon.voltageLevel = jsonString(m, "voltageLevel");
|
||
mon.connectType = jsonString(m, "ptType");
|
||
mon.status = jsonString(m, "status");
|
||
mon.fields = objectFields(m);
|
||
dev.monitors.push_back(mon);
|
||
}
|
||
}
|
||
devices.push_back(dev);
|
||
}
|
||
|
||
if (log) log("台账解析完成: 装置 " + std::to_string(devices.size()) + " 个");
|
||
|
||
// 3. 请求 ICD 列表。icdFlag=1 时只请求本次台账中出现的 devType。
|
||
json icdReq = json::array();
|
||
if (cfg.icdFlag == "1") {
|
||
for (const auto& t : devTypes) icdReq.push_back(t);
|
||
}
|
||
if (log) log("获取 ICD 列表: " + icdReq.dump());
|
||
std::string icdRaw = httpRequest(cfg.webIcd, "", icdReq.dump(), log);
|
||
writeTextFile(runtimeJsonPath("icd_response.json"), icdRaw);
|
||
|
||
fs::path saveDir = fs::u8path(cfg.xmlSaveDir.empty() ? "downloaded_xml" : cfg.xmlSaveDir);
|
||
fs::create_directories(saveDir);
|
||
json icdData = parseResponseDataArray(icdRaw, "ICD");
|
||
int downloaded = 0;
|
||
std::unordered_map<std::string, std::string> devTypeToXml;
|
||
// 4. 按 filePath 下载 XML，保存到 xmlSaveDir，并建立 devType -> XML 路径映射。
|
||
for (const auto& model : icdData) {
|
||
std::string modelId = jsonString(model, "id");
|
||
std::string devType = jsonString(model, "devType");
|
||
std::string fileName = jsonString(model, "fileName");
|
||
std::string filePath = jsonString(model, "filePath");
|
||
if (filePath.empty()) continue;
|
||
|
||
std::string localName = !modelId.empty() ? (sanitizeFilePart(modelId) + ".xml")
|
||
: !fileName.empty() ? sanitizeFilePart(fileName)
|
||
: (sanitizeFilePart(devType) + ".xml");
|
||
fs::path localPath = saveDir / fs::u8path(localName);
|
||
std::string xml = httpRequest(cfg.webFiledownload, "filePath=" + urlEncode(filePath), "", log);
|
||
if (!looksLikeXml(xml)) {
|
||
if (log) log("跳过非 XML 下载结果: " + filePath);
|
||
continue;
|
||
}
|
||
writeTextFile(localPath, xml);
|
||
if (!devType.empty() && !devTypeToXml.count(devType)) devTypeToXml[devType] = localPath.u8string();
|
||
if (firstXmlPath.empty()) firstXmlPath = localPath.u8string();
|
||
++downloaded;
|
||
if (log) log("已下载 XML: " + localPath.u8string());
|
||
}
|
||
|
||
if (log) log("ICD/XML 下载完成: " + std::to_string(downloaded) + " 个文件");
|
||
for (auto& dev : devices) {
|
||
auto it = devTypeToXml.find(dev.devType);
|
||
if (it != devTypeToXml.end()) dev.xmlPath = it->second;
|
||
}
|
||
return true;
|
||
}
|
||
|
||
bool pqInitializeMq(const PqToolConfig& cfg, PqLogFn log, PqMqEventFn event) {
|
||
// 初始化时先把界面配置落盘，便于用户核对，也便于离线排查 MQ 参数。
|
||
json saved = {
|
||
{"producer", cfg.producer},
|
||
{"ipport", cfg.ipport},
|
||
{"producerAccessKey", cfg.producerAccessKey},
|
||
{"producerSecretKey", cfg.producerSecretKey},
|
||
{"topicLOG", cfg.topicLOG},
|
||
{"tagLOG", cfg.tagLOG},
|
||
{"keyLOG", cfg.keyLOG},
|
||
{"topicAsk", cfg.topicAsk},
|
||
{"tagAsk", cfg.tagAsk},
|
||
{"keyAsk", cfg.keyAsk},
|
||
{"consumer", cfg.consumer},
|
||
{"consumerIpport", cfg.consumerIpport},
|
||
{"consumerAccessKey", cfg.consumerAccessKey},
|
||
{"consumerSecretKey", cfg.consumerSecretKey},
|
||
{"dataStatTopic", cfg.dataStatTopic},
|
||
{"dataStatTag", cfg.dataStatTag},
|
||
{"dataStatKey", cfg.dataStatKey},
|
||
{"dataRealtimeTopic", cfg.dataRealtimeTopic},
|
||
{"dataRealtimeTag", cfg.dataRealtimeTag},
|
||
{"dataRealtimeKey", cfg.dataRealtimeKey},
|
||
{"activeDataTopic", cfg.dataTopic},
|
||
{"dataTag", cfg.dataTag},
|
||
{"dataKey", cfg.dataKey},
|
||
{"traceTopic", cfg.traceTopic},
|
||
{"traceTag", cfg.traceTag},
|
||
{"traceKey", cfg.traceKey}
|
||
};
|
||
writeTextFile(runtimeJsonPath("mq_runtime_config.json"), saved.dump(2));
|
||
if (log) log("MQ config saved to " + runtimeJsonPath("mq_runtime_config.json").u8string());
|
||
if (!rocketRuntime().initialize(cfg, log, event) && log) {
|
||
// 这里仍返回 true，让 GUI 可以继续构造追踪请求；发送失败时会进入 outbox 兜底。
|
||
log("RocketMQ runtime is not active; trace messages will be written to mq_outbox_*.jsonl.");
|
||
log("RocketMQ source is present, but a 32-bit rocketmq-client-cpp.dll or rocketmq.dll is still required.");
|
||
}
|
||
return true;
|
||
if (log) {
|
||
log("MQ 配置已初始化并写入 mq_runtime_config.json");
|
||
log("当前 Windows 编译未链接 RocketMQ SDK，发送消息会先落地到 mq_outbox_*.jsonl。");
|
||
log("接入 RocketMQ DLL 后只需要替换 mq.cpp 中 pqInitializeMq/pqSendTraceRequest 的适配实现。");
|
||
}
|
||
return true;
|
||
}
|
||
|
||
std::string pqBuildTracePayload(const PqToolConfig& cfg,
|
||
const PqLedgerDevice& device,
|
||
const PqLedgerMonitor& monitor,
|
||
const std::string& frontType) {
|
||
std::string guid = makeGuid();
|
||
int processNo = 0;
|
||
if (frontType == "cfg_3s_data") {
|
||
// 实时数据进程约定为 0；统计数据使用台账中的 processNo。
|
||
processNo = 0;
|
||
} else {
|
||
try {
|
||
processNo = device.processNo.empty() ? 0 : std::stoi(device.processNo);
|
||
} catch (...) {
|
||
processNo = 0;
|
||
}
|
||
}
|
||
|
||
// body 是前置识别的 set_log 指令；outer 是 RocketMQ 业务包装。
|
||
json body = {
|
||
{"code", "set_log"},
|
||
{"nodeId", cfg.frontInst},
|
||
{"guid", guid},
|
||
{"processNo", processNo},
|
||
{"id", monitor.id},
|
||
{"level", "measurepoint"},
|
||
{"grade", "TRACE"},
|
||
{"logtype", 1},
|
||
{"frontType", frontType}
|
||
};
|
||
|
||
json outer = {
|
||
{"key", guid},
|
||
{"source", "WEB"},
|
||
{"sendTime", nowIsoMillis()},
|
||
{"retryTimes", 0},
|
||
{"messageBody", body.dump()}
|
||
};
|
||
return outer.dump();
|
||
}
|
||
|
||
std::string buildRealtimeAskPayload(const PqLedgerDevice& device,
|
||
const PqLedgerMonitor& monitor) {
|
||
const std::string guid = makeGuid();
|
||
json body = {
|
||
{"devSeries", device.id},
|
||
{"limit", 20},
|
||
{"line", monitor.id},
|
||
{"realData", true},
|
||
{"soeData", true}
|
||
};
|
||
json outer = {
|
||
{"key", guid},
|
||
{"source", "WEB"},
|
||
{"sendTime", nowIsoMillis()},
|
||
{"retryTimes", 0},
|
||
{"messageBody", body.dump()}
|
||
};
|
||
return outer.dump();
|
||
}
|
||
|
||
bool pqSendTraceRequest(const PqToolConfig& cfg,
|
||
const PqLedgerDevice& device,
|
||
const PqLedgerMonitor& monitor,
|
||
const std::string& frontType,
|
||
PqLogFn log,
|
||
PqMqEventFn event) {
|
||
std::string payload = pqBuildTracePayload(cfg, device, monitor, frontType);
|
||
std::string guid = guidFromJsonText(payload);
|
||
// 先设置追踪目标，确保追踪请求发出后立即到达的 MQ 回调也能正确路由。
|
||
rocketRuntime().setTraceTarget(cfg, device, monitor, frontType, guid);
|
||
fs::path traceDir = fs::u8path(pqTraceDirectoryFor(cfg, device, monitor));
|
||
fs::create_directories(traceDir);
|
||
fs::path sentPath = traceDir / "sent_trace_request.json";
|
||
writeTextFile(sentPath, prettyJsonOrRaw(payload));
|
||
writeTextFile(runtimeJsonPath("last_trace_request.json"), json::parse(payload).dump(2));
|
||
if (event) event({"sent", eventTitle("Send", sentPath.u8string()), payload, sentPath.u8string(), monitor.id, traceDir.u8string()});
|
||
auto sendOrSave = [&](const std::string& topic,
|
||
const std::string& tag,
|
||
const std::string& messageKey,
|
||
const std::string& body,
|
||
const std::string& label) {
|
||
std::string mqError;
|
||
if (rocketRuntime().sendPayload(topic, tag, messageKey, body, label, mqError, log)) return;
|
||
|
||
// RocketMQ 未初始化或发送失败时不丢请求，按目标 Topic 追加到对应 outbox。
|
||
std::string topicPart = sanitizeFilePart(topic.empty() ? label : topic);
|
||
fs::path outbox = runtimeJsonPath("mq_outbox_" + topicPart + ".jsonl");
|
||
std::ofstream ofs(outbox, std::ios::binary | std::ios::app);
|
||
if (!ofs) throw std::runtime_error("无法写入 MQ outbox: " + outbox.string());
|
||
ofs << body << "\n";
|
||
if (log) log(label + "消息发送失败(" + mqError + ")，已写入 " + outbox.string());
|
||
};
|
||
|
||
// 第一条始终先发到 TopicLOG，请求前置建立本次追踪日志。
|
||
sendOrSave(cfg.topicLOG, cfg.tagLOG, cfg.keyLOG, payload, "set_log");
|
||
|
||
// cfg_3s_data 紧接着向 TopicAsk 请求该装置、测点的实时/SOE 数据。
|
||
if (frontType == "cfg_3s_data") {
|
||
std::string askPayload = buildRealtimeAskPayload(device, monitor);
|
||
fs::path askPath = traceDir / "sent_realtime_request.json";
|
||
writeTextFile(askPath, prettyJsonOrRaw(askPayload));
|
||
writeTextFile(runtimeJsonPath("last_realtime_request.json"), json::parse(askPayload).dump(2));
|
||
if (event) event({"sent", eventTitle("Send Ask", askPath.u8string()), askPayload, askPath.u8string(), monitor.id, traceDir.u8string()});
|
||
sendOrSave(cfg.topicAsk, cfg.tagAsk, cfg.keyAsk, askPayload, "realtime ask");
|
||
}
|
||
|
||
if (log) log("Trace target monitor=" + monitor.name + "(" + monitor.id + "), frontType=" + frontType);
|
||
return true;
|
||
}
|