879 lines
28 KiB
C++
879 lines
28 KiB
C++
#include "Data_Source.h"
|
||
|
||
#include <algorithm>
|
||
#include <asio/post.hpp>
|
||
#include <asio/thread_pool.hpp>
|
||
#include <codecvt>
|
||
#include <thread>
|
||
|
||
#include "../server/Global.h"
|
||
#include "Database.h"
|
||
|
||
namespace {
|
||
asio::thread_pool& data_source_process_pool() {
|
||
const auto hardware_threads = std::thread::hardware_concurrency();
|
||
const auto worker_count = std::max(2u, hardware_threads > 2 ? hardware_threads - 2 : hardware_threads);
|
||
static asio::thread_pool pool(worker_count);
|
||
return pool;
|
||
}
|
||
}
|
||
|
||
Psc::serial::Serial* create_serial(const std::string& serial_name, Baud_Rate_Type baud_rate) {
|
||
auto serial = new Psc::serial::Serial;
|
||
serial->set_serial_name(serial_name);
|
||
serial->set_baud_rate(baud_rate);
|
||
serial->set_parity(Psc::serial::Parity::NoParity);
|
||
serial->set_data_bits(Psc::serial::DataBits::Data8);
|
||
serial->set_stop_bits(Psc::serial::StopBits::OneStop);
|
||
serial->set_flow_control(Psc::serial::FlowControl::HardwareControl);
|
||
serial->set_buffer_byte_size(10 * 1024);
|
||
if (!serial->open())
|
||
{
|
||
std::cerr << "createSerial " + serial_name + ":" + std::to_string(baud_rate) + " 打开串口失败！\n";
|
||
Psc::fail_fast();
|
||
return nullptr;
|
||
} else
|
||
{
|
||
// std::cerr << "createSerial " + serial_name + ":" + std::to_string(baud_rate) + " 打开串口成功！\n";
|
||
}
|
||
return serial;
|
||
}
|
||
|
||
std::shared_ptr<Data_Source> create_from_json(const JSON* that_json) {
|
||
std::string type = that_json->get_string("type");
|
||
std::shared_ptr<Data_Source> ret = create_data_source_from_type(type);
|
||
ret->from_json(that_json);
|
||
// if (ret->enable)
|
||
// {
|
||
// ret->open();
|
||
// }
|
||
return ret;
|
||
}
|
||
|
||
|
||
|
||
|
||
std::shared_ptr<Data_Source> Data_Source::that() {
|
||
return shared_from_this();
|
||
}
|
||
Psc::JSON Data_Source::get_state() {
|
||
Psc::JSON ret = Psc::JSON::object();
|
||
ret.append_list(get_custom_state_json().children);
|
||
ret.append({"ds: read_speed（byte）", read_speed});
|
||
ret.append({"ds: value_statistics（byte）", value_statistics});
|
||
return ret;
|
||
}
|
||
|
||
Data_Source::Data_Source() {
|
||
// parse_format = std::make_shared<Input_Format>();
|
||
// 写入到配置文件是懒加载 其他保存时他跟着保存
|
||
base_station.handle_when_updated = [this](SSR::HULC_Status_Message msg) {
|
||
if (this->update_form_gps) {
|
||
lat = msg.get_latitude();
|
||
lon = msg.get_longitude();
|
||
alt = msg.Alt;
|
||
}
|
||
};
|
||
}
|
||
|
||
std::string Data_Source::thread_key() const {
|
||
//return "Data_Source_Handle_Thread:[" + key + "]";
|
||
return key + "_DS_HT";
|
||
}
|
||
|
||
void Data_Source::test_and_attach_thread() {
|
||
if (enable) {
|
||
open();
|
||
}
|
||
}
|
||
|
||
|
||
void File_Data_Source::before_handle_msg(std::shared_ptr<SSR::Mode_Msg>& msg) {
|
||
auto cur = std::dynamic_pointer_cast<SSR::Mode_Msg>(msg);
|
||
if (cur != nullptr) {
|
||
auto& pre = last_prase_msg;
|
||
if (!pre) {
|
||
pre = cur;
|
||
return;
|
||
}
|
||
cur->day_num = pre->day_num;
|
||
if (cur->time() < pre->time()) {
|
||
++cur->day_num;
|
||
}
|
||
}
|
||
}
|
||
|
||
|
||
|
||
|
||
|
||
ucoro::awaitable<void> Data_Source::source_step_coro() {
|
||
co_await handle_in_loop_coro();
|
||
auto mode_data = co_await read_coro();
|
||
if (mode_data.empty()) {
|
||
std::this_thread::sleep_for(std::chrono::milliseconds(10));
|
||
co_return;
|
||
}
|
||
struct Process_Result {
|
||
std::exception_ptr exception;
|
||
};
|
||
auto self = that();
|
||
auto result = co_await ucoro::callback_awaitable<Process_Result>(
|
||
[this, self = std::move(self), mode_data = std::move(mode_data)](auto done) mutable {
|
||
asio::post(
|
||
data_source_process_pool(),
|
||
[this, self = std::move(self), mode_data = std::move(mode_data), done = std::move(done)]() mutable {
|
||
try {
|
||
process_mode_acs_data(mode_data);
|
||
done(Process_Result{});
|
||
} catch (...) {
|
||
done(Process_Result{std::current_exception()});
|
||
}
|
||
});
|
||
});
|
||
|
||
if (result.exception) {
|
||
std::rethrow_exception(result.exception);
|
||
}
|
||
co_return;
|
||
}
|
||
|
||
|
||
|
||
|
||
|
||
void Data_Source::test_and_stop_thread() const {
|
||
Global::instance()->thread_manager.test_and_stop_thread(thread_key());
|
||
}
|
||
|
||
Psc::JSON Data_Source::get_all_connect_feed_status() {
|
||
Psc::JSON ret = Psc::JSON::object();
|
||
std::vector<Cached_Source_Info> list;
|
||
{
|
||
std::lock_guard g(cdf_mtx);
|
||
list = cached_data_feed_key_list;
|
||
}
|
||
for (auto& i : list) {
|
||
Psc::JSON cur = Psc::JSON::object();
|
||
auto odf = Global::instance()->mode_acs.data_feed_config.map.get(i.key);
|
||
if (odf.has_value()) {
|
||
const auto& df = odf.value();
|
||
i.mode_s_cache_num.update(df->msg_buffer.mode_s_msg_num);
|
||
i.mode_other_cache_num.update(df->msg_buffer.mode_other_msg_num);
|
||
cur.append({"mode_s_msg_num", i.mode_s_cache_num.to_json()});
|
||
cur.append({"mode_other_msg_num", i.mode_other_cache_num.to_json()});
|
||
}
|
||
ret.append({i.key, cur});
|
||
}
|
||
return ret;
|
||
}
|
||
Psc::JSON Data_Source::get_all_connect_feed() {
|
||
Psc::JSON ret = Psc::JSON::array();
|
||
std::vector<Cached_Source_Info> list;
|
||
{
|
||
std::lock_guard g(cdf_mtx);
|
||
list = cached_data_feed_key_list;
|
||
}
|
||
for (auto& i : list) {
|
||
Psc::JSON cur = Psc::JSON::object();
|
||
cur.append({"key", i.key});
|
||
ret.append(cur);
|
||
}
|
||
return ret;
|
||
}
|
||
|
||
|
||
std::vector<std::string> readLines(const std::string& path) {
|
||
std::vector<std::string> lines;
|
||
|
||
std::filesystem::path p = std::filesystem::path((const char8_t*)path.c_str());
|
||
|
||
//std::filesystem::path p = std::filesystem::u8path(path);
|
||
std::ifstream istream(p);
|
||
if (!istream) {
|
||
std::cerr << "读取模拟数据源 readLines 无法打开文件：" << path << std::endl;
|
||
Psc::fail_fast();
|
||
}
|
||
|
||
// 将整个文件读入一个字符串
|
||
std::stringstream stream;
|
||
stream << istream.rdbuf();
|
||
// 使用 stringstream 按行分割内容
|
||
std::string line;
|
||
char ch;
|
||
while (stream.get(ch)) {
|
||
if (ch == ' ') {
|
||
continue;
|
||
}
|
||
if (ch == '\r' || ch == '\n') {
|
||
if (!line.empty()) {
|
||
lines.push_back(line);
|
||
line.clear();
|
||
}
|
||
|
||
// if (ch == '\r') {
|
||
// stream.get(); // 吃掉 \n
|
||
// }
|
||
// if (ch == '\n') {
|
||
// stream.get(); // 吃掉 \n
|
||
// }
|
||
// 处理 \r\n 组合：如果当前是 \r，下一个是 \n，跳过它
|
||
// if (ch == '\r' && stream.peek() == '\n') {
|
||
// stream.get(); // 吃掉 \n
|
||
// }
|
||
|
||
} else {
|
||
line += ch;
|
||
}
|
||
if (line.size() == 1000) {
|
||
lines.push_back(line);
|
||
line.clear();
|
||
}
|
||
}
|
||
// 最后一行如果没有换行符也处理一下
|
||
if (!line.empty()) {
|
||
lines.push_back(line);
|
||
}
|
||
std::ostringstream oss;
|
||
oss << path + " 总行数：" + std::to_string(lines.size()) + " 有效行数：" + std::to_string(lines.size()) + "\n" << std::flush;
|
||
std::cout << oss.str() << std::flush;
|
||
return lines;
|
||
}
|
||
|
||
|
||
|
||
std::string extractID(const std::string& logLine) {
|
||
size_t lastSpacePos = logLine.rfind(" "); // 查找最后一个空格
|
||
if (lastSpacePos != std::string::npos) {
|
||
return logLine.substr(lastSpacePos + 1); // 从最后一个空格之后提取字符串
|
||
}
|
||
return logLine;
|
||
}
|
||
|
||
time_t convert_to_timestamp(const std::string& str) {
|
||
// 创建一个结构体 tm 来存储解析后的时间
|
||
std::tm timeStruct = {};
|
||
|
||
std::istringstream ss(str);
|
||
ss >> std::get_time(&timeStruct, "%Y-%m-%d %H:%M:%S");
|
||
if (ss.fail()) {
|
||
std::cerr << "Failed to parse time" << std::endl;
|
||
return -1; // 如果解析失败，返回 -1
|
||
}
|
||
|
||
// 将 tm 转换为 time_t（时间戳）
|
||
time_t timestamp = std::mktime(&timeStruct);
|
||
|
||
if (timestamp == -1) {
|
||
std::cerr << "Failed to convert to time_t" << std::endl;
|
||
return -1; // 如果转换失败，返回 -1
|
||
}
|
||
return timestamp;
|
||
}
|
||
|
||
std::tuple<std::string, time_t> extractID2(const std::string& logLine) {
|
||
size_t lastSpacePos = logLine.rfind(" "); // 查找最后一个空格
|
||
time_t t = convert_to_timestamp(logLine.substr(0, 19));
|
||
if (lastSpacePos != std::string::npos) {
|
||
return {logLine.substr(lastSpacePos + 1), t}; // 从最后一个空格之后提取字符串
|
||
}
|
||
return {logLine, t};
|
||
}
|
||
|
||
|
||
std::string bin_format(const std::string& hex, time_t t){
|
||
std::tm* currentTime = std::localtime(&t);
|
||
auto sec = currentTime->tm_hour * 3600 + currentTime->tm_min * 60 + currentTime->tm_sec;
|
||
SSR::MLAT_timestamp a(sec, 0);
|
||
return create_Binary_Format_memory(hex, 0, &a);
|
||
}
|
||
|
||
std::optional<std::string> File_Data_Source::get_raw_line(int& ret_index) {
|
||
if (index == 0 && part_infos.empty())
|
||
{
|
||
ret_index = -1;
|
||
return std::nullopt;
|
||
}
|
||
std::optional<std::string> log_info = std::nullopt;
|
||
if (index == part_infos.size())
|
||
{
|
||
if (play_mode == Play_Mode::loop)
|
||
{
|
||
index = 0;
|
||
} else
|
||
{
|
||
if (!have_report_play_back_all_success)
|
||
{
|
||
std::ostringstream oss;
|
||
oss << SSR::get_current_date() << " " << SSR::get_current_time() << " [进程:" << std::this_thread::get_id() << "] " << file_path + " 全部加载成功！\n" << std::flush;
|
||
std::cout << oss.str() << std::flush;
|
||
have_report_play_back_all_success = true;
|
||
}
|
||
ret_index = -1;
|
||
return std::nullopt;
|
||
}
|
||
}
|
||
ret_index = index + 1;
|
||
return part_infos[index++];
|
||
}
|
||
|
||
|
||
// 1A 33 1A 1A F1 FB 87 73 7E 7F a8001d81a87543b0a80000
|
||
|
||
std::string get_true_from_raw_line(Data_Source* ds, File_Data_Type data_type, int index, const std::string& log_info) {
|
||
std::string ret;
|
||
|
||
// 2025-01-06 07:20:17 [info] dc2541e65001401a3119b2dd1c7af67132201
|
||
if (data_type == File_Data_Type::BIN) {
|
||
return log_info;
|
||
}
|
||
else if (data_type == File_Data_Type::BIN_Text)
|
||
{
|
||
auto info = extractID(log_info);
|
||
ret = hex2mem(info);
|
||
}
|
||
else if (data_type == File_Data_Type::AVR)
|
||
{
|
||
auto [info, time] = extractID2(log_info);
|
||
ret = bin_format(info, time);
|
||
}
|
||
// 1A 33 10 01 A5 31 FB 9A 5D 8D 78 0E 47 99 08 8E 32 B0 08 BE E9 8A E6 1A 32 10 01 AA 14 61 7E 5E 02 E6 0F 38 7D 87 08
|
||
// 1A 32 10 01 AD B5 64 F7 63 5D 78 0F 9D BA 93 BA
|
||
else if (data_type == File_Data_Type::BIN_Blank_Text)
|
||
{
|
||
std::string info = log_info;
|
||
std::string result = ds->last_char;
|
||
ds->last_char = "";
|
||
|
||
|
||
|
||
for (char c : info)
|
||
{
|
||
if (std::isxdigit(c)) { // 会判断字符 c 是否是十六进制数字字符
|
||
result += c;
|
||
} else if (std::isspace(c)) {
|
||
//
|
||
} else {
|
||
std::ostringstream oss;
|
||
oss << "存在非法字符[" << c << "]: value:" << (int)c << " in：" << log_info << LOG_POS;
|
||
server_logger->c_debug("非法字符：", {}, oss.str());
|
||
return "";
|
||
}
|
||
}
|
||
auto size = result.size();
|
||
if (size % 2 != 0) {
|
||
// 保存最后一个字符
|
||
ds->last_char = result.empty() ? "" : std::string(1, result.back());
|
||
result.pop_back();
|
||
}
|
||
ret = hex2mem(result);
|
||
|
||
// if (size % 2 != 0)
|
||
// {
|
||
// std::ostringstream oss;
|
||
// size_t prefix_num = 20;
|
||
// size_t suffix_num = 20;
|
||
// auto origin_size = log_info.size();
|
||
// prefix_num = std::min(origin_size, prefix_num);
|
||
// suffix_num = std::min(origin_size, suffix_num);
|
||
// std::string prefix_data = log_info.substr(0, prefix_num);
|
||
// std::string suffix_data = log_info.substr(log_info.size() - suffix_num, suffix_num);
|
||
// oss << "index:" << index << " 长度不为偶数 不能转换成内存：" << "[内容]" << "result_size" << result.size() << " origin_size:[" << origin_size << "]" << std::endl <<
|
||
// " prefix[" << prefix_num << "]:" << prefix_data << std::endl <<
|
||
// " suffix[" << suffix_num << "]:" << suffix_data << std::endl <<
|
||
// std::endl;
|
||
// server_logger->c_debug("长度不为偶数", {}, oss.str());
|
||
// //std::cout << "result:" << log_info << std::endl << std::endl;
|
||
// return "";
|
||
// }
|
||
|
||
}
|
||
else if (data_type == File_Data_Type::BIN_Blank_One_Line_With_Escape) {
|
||
auto size = log_info.size();
|
||
if (size % 2 != 0)
|
||
{
|
||
return hex2mem(log_info.substr(0, size - 1));
|
||
}
|
||
return hex2mem(log_info);
|
||
}
|
||
else if (data_type == File_Data_Type::BIN_Blank_One_Line_No_Escape) {
|
||
auto size = log_info.size();
|
||
std::string data;
|
||
if (size % 2 != 0)
|
||
{
|
||
data = log_info.substr(0, size - 1);
|
||
} else {
|
||
data = log_info;
|
||
}
|
||
return SSR::packet_to_escape_format(hex2mem(data));
|
||
}
|
||
else if (data_type == File_Data_Type::SIMPLE_BIN_Blank)
|
||
{
|
||
std::string info = log_info;
|
||
std::string result;
|
||
for (char c : info)
|
||
{
|
||
if (c != ' ')
|
||
{
|
||
result += c;
|
||
}
|
||
}
|
||
// 1 + 1 + 6 + 1 + 2/7/14
|
||
// 1a + type time_stamp signal_level data
|
||
int size = result.size();
|
||
if (size != 2*2 && size != 7*2 && size != 14*2)
|
||
{
|
||
std::cout << "line_offset:[" << index << "] " << "大小" << size << " 不为偶数,或不为2，7，14 不能转换成二进制格式：" << result << std::endl;
|
||
return "";
|
||
}
|
||
std::string head("\x1a");
|
||
std::string type;
|
||
std::string time_stamp = "\x11\x22\x33\x44\x55\x66";
|
||
std::string signal_level("\xFF");
|
||
if (size == 2 * 2)
|
||
{
|
||
type = R"(1)";
|
||
} else if (size == 7 * 2)
|
||
{
|
||
type = R"(2)";
|
||
} else if (size == 14 * 2)
|
||
{
|
||
type = R"(3)";
|
||
}
|
||
ret = SSR::packet_to_escape_format(head + type + time_stamp + signal_level + hex2mem(result));
|
||
|
||
//std::cout << ret.size() << std::endl;
|
||
}
|
||
else
|
||
{
|
||
std::cout << "未知的回放类型 " << static_cast<int>(data_type) << std::endl;
|
||
Psc::fail_fast();
|
||
}
|
||
return ret;
|
||
}
|
||
|
||
|
||
|
||
|
||
void File_Data_Source::_close() {
|
||
std::lock_guard g(mtx);
|
||
index = 0;
|
||
part_infos.clear();
|
||
}
|
||
|
||
bool File_Data_Source::_open() {
|
||
std::lock_guard g(mtx);
|
||
auto path = get_true_file_path();
|
||
|
||
namespace fs = std::filesystem;
|
||
if (!fs::exists(path)) {
|
||
state = "文件不存在";
|
||
return false;
|
||
}
|
||
|
||
if (data_type == File_Data_Type::BIN) {
|
||
part_infos = readBinaryFileAsString(path, 1000);
|
||
}
|
||
else {
|
||
part_infos = readLines(path);
|
||
}
|
||
state = "已加载" + to_string(part_infos.size()) + "长度数据!";
|
||
return true;
|
||
}
|
||
|
||
|
||
std::vector<std::string> File_Data_Source::readBinaryFileAsString(const std::string& filepath, size_t part_size) {
|
||
// 打开文件（以二进制模式）
|
||
std::ifstream file(filepath, std::ios::binary);
|
||
|
||
if (!file) {
|
||
std::cerr << "无法打开文件: " << filepath << std::endl;
|
||
return {}; // 文件打开失败，返回空vector
|
||
}
|
||
|
||
// 获取文件大小
|
||
file.seekg(0, std::ios::end);
|
||
size_t fileSize = file.tellg();
|
||
file.seekg(0, std::ios::beg);
|
||
|
||
// 存储读取的部分
|
||
std::vector<std::string> parts;
|
||
|
||
// 读取文件的每一部分
|
||
size_t bytesRead = 0;
|
||
while (bytesRead < fileSize) {
|
||
// 计算每个部分的长度（最后一部分可能小于part_size）
|
||
size_t remaining = fileSize - bytesRead;
|
||
size_t currentPartSize = (remaining < part_size) ? remaining : part_size;
|
||
|
||
// 创建buffer来存储当前部分
|
||
std::string buffer(currentPartSize, '\0');
|
||
|
||
// 读取当前部分
|
||
file.read(&buffer[0], currentPartSize);
|
||
|
||
// 将读取的部分添加到vector
|
||
parts.push_back(std::move(buffer));
|
||
|
||
// 更新已读取字节数
|
||
bytesRead += currentPartSize;
|
||
}
|
||
|
||
// 关闭文件
|
||
file.close();
|
||
|
||
return parts;
|
||
}
|
||
|
||
|
||
ucoro::awaitable<std::string> File_Data_Source::read_coro() {
|
||
std::lock_guard g(mtx);
|
||
int ret_index = 0;
|
||
if (play_mode != Play_Mode::analysis)
|
||
{
|
||
std::optional<std::string> log_info = get_raw_line(ret_index);
|
||
co_return log_info.has_value() ? get_true_from_raw_line(this, data_type, ret_index, log_info.value()) : "";
|
||
} else
|
||
{
|
||
int num = 0;
|
||
std::string ret;
|
||
std::optional<std::string> log_info = get_raw_line(ret_index);
|
||
while (log_info.has_value())
|
||
{
|
||
num++;
|
||
auto data = log_info.value();
|
||
auto cur = get_true_from_raw_line(this, data_type, ret_index, data);
|
||
ret += cur;
|
||
// 一定要在这个位置，否则会导致遗漏报文
|
||
if (num == 1000) {
|
||
break;
|
||
}
|
||
log_info = get_raw_line(ret_index);
|
||
}
|
||
// std::cout << "size == " << t.size() << std::endl;
|
||
co_return ret;
|
||
}
|
||
}
|
||
|
||
|
||
void File_Data_Source::handle_mode_s(std::shared_ptr<SSR::Mode_S_Msg> msg) {
|
||
if (!msg) return;
|
||
|
||
cache_list.push_back(std::make_shared<Cache_Msg>(std::move(msg)));
|
||
|
||
if (cache_list.empty()) return;
|
||
if (pre_cache == nullptr) {
|
||
pre_cache = cache_list.front();
|
||
cache_list.pop_front();
|
||
|
||
SSR::Play_Back_Time_Point& start = player_clock.start_time;
|
||
start.day_num = pre_cache->day_num;
|
||
start.day_sec = pre_cache->msg->mlat_timestamp.daysec;
|
||
}
|
||
|
||
auto sys_time = player_clock.get_cur_time_point();
|
||
while (true) {
|
||
if (cache_list.empty()) break;
|
||
auto pre = pre_cache;
|
||
auto cur = cache_list.front();
|
||
cur->day_num = pre->day_num;
|
||
if (cur->time() < pre->time()) {
|
||
cur->day_num++;
|
||
}
|
||
cache_list.pop_front();
|
||
pre_cache = cur;
|
||
Data_Source::handle_mode_s(cur->msg);
|
||
if (cur->time() > sys_time) {
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
|
||
|
||
ucoro::awaitable<std::string> Dll_Data_Source::read_coro() {
|
||
|
||
auto start = std::chrono::steady_clock::now();
|
||
|
||
if (read_func_ptr == nullptr) co_return "";
|
||
|
||
auto buf = reinterpret_cast<char *>(buffer.data());
|
||
auto len = read_func_ptr(buf, buffer_size);
|
||
vs.update(len);
|
||
|
||
std::string data(buf, len);
|
||
|
||
if (data.empty()) {
|
||
auto end = std::chrono::steady_clock::now();
|
||
auto cost = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count();
|
||
//std::cout << "Dll_Data_Source::read cost: " << cost << " us\n";
|
||
co_return "";
|
||
}
|
||
|
||
auto ret = get_true_from_raw_line(this, data_type, -1, data);
|
||
|
||
auto end = std::chrono::steady_clock::now();
|
||
auto cost = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count();
|
||
|
||
// std::cout << "Dll_Data_Source::read cost: " << cost << " us\n";
|
||
|
||
co_return ret;
|
||
}
|
||
|
||
|
||
bool Shared_Memory_Data_Source::_open() {
|
||
sm = std::make_unique<Psc::SM_RingBuffer>();
|
||
sm->init(shared_memory_name, shared_memory_size);
|
||
return true;
|
||
}
|
||
void Shared_Memory_Data_Source::_close() {
|
||
sm.reset();
|
||
}
|
||
|
||
|
||
ucoro::awaitable<std::string> Shared_Memory_Data_Source::read_coro() {
|
||
auto size = sm->shm.size();
|
||
size_t length = size;
|
||
std::string data;
|
||
data.resize(size);
|
||
|
||
bool ok = sm->read((uint8_t *)data.data(), length);
|
||
if (!ok) {
|
||
co_return "";
|
||
}
|
||
|
||
if (length != size) {
|
||
data.resize(length);
|
||
}
|
||
if (Global::instance()->console_config.mode_s_console) {
|
||
std::cout << "read:" << data << std::endl;
|
||
}
|
||
auto ret = get_true_from_raw_line(this, data_type, -1, data);
|
||
co_return ret;
|
||
}
|
||
|
||
void Data_Source_Config::server(Global* g) {
|
||
auto& svr = g->svr;
|
||
auto& api = g->api;
|
||
std::string name = "data_source";
|
||
// 返回所有 JSON 数据的 API
|
||
svr.Post(api + "get_data_source_connect", [this](HTTP_Param) {
|
||
CHECK_JSON_PARAM
|
||
HTTP_REQUIRE_VALUE(key, params.try_get_string("key"))
|
||
auto t = map.get(key);
|
||
JSON ret{nullptr};
|
||
if (t.has_value()) {
|
||
ret = t.value()->get_all_connect_feed();
|
||
}
|
||
res->setBody(warp(ret).to_json_string());
|
||
});
|
||
|
||
|
||
|
||
svr.Post(api + "get_data_source_state", [this](HTTP_Param) {
|
||
CHECK_JSON_PARAM
|
||
HTTP_REQUIRE_VALUE(key, params.try_get_string("key"))
|
||
auto t = map.get(key);
|
||
JSON ret{nullptr};
|
||
if (t.has_value()) {
|
||
ret = t.value()->get_state();
|
||
}
|
||
res->setBody(warp(ret).to_json_string());
|
||
});
|
||
|
||
|
||
svr.Post(api + "get_mode_s_data_source_config", [this](HTTP_Param) {
|
||
res->setBody(warp(to_json()).to_json_string());
|
||
});
|
||
|
||
svr.Post(api + svr.update + name, [this, g](HTTP_Param) {
|
||
CHECK_JSON_PARAM
|
||
HTTP_REQUIRE_VALUE(key, params.try_get_string("key"))
|
||
HTTP_REQUIRE_VALUE(ds, map.get(key))
|
||
// 直接关闭
|
||
ds->close();
|
||
ds->from_json(¶ms);
|
||
if (ds->enable) {
|
||
ds->open();
|
||
ds->test_and_attach_thread();
|
||
} else
|
||
{
|
||
ds->test_and_stop_thread();
|
||
}
|
||
g->save();
|
||
g->mode_acs.source_feed_relation_config.set_need_refresh();
|
||
});
|
||
svr.Post(api + svr.insert + name, [g, this](HTTP_Param) {
|
||
CHECK_JSON_PARAM
|
||
|
||
HTTP_REQUIRE_VALUE(index, params.try_get_number<int>("index"))
|
||
HTTP_REQUIRE_PTR(data, params.get("data"))
|
||
HTTP_REQUIRE_VALUE(t, data->try_get_string("type"))
|
||
std::cout << params.to_json_string() << std::endl;
|
||
std::shared_ptr<Data_Source> ds = create_data_source_from_type(t);
|
||
ds->from_json(data);
|
||
HTTP_REQUIRE_VALUE(enable, data->try_get_bool("enable"))
|
||
if (enable)
|
||
{
|
||
ds->open();
|
||
ds->test_and_attach_thread();
|
||
}
|
||
|
||
HTTP_REQUIRE_TRUE(map.insert(index, ds), "index")
|
||
//std::cout << params.to_json_string() << std::endl;
|
||
g->mode_acs.source_feed_relation_config.set_need_refresh();
|
||
Config::save();
|
||
res->setBody(warp(to_json()).to_json_string());
|
||
});
|
||
|
||
svr.Post(api + svr.remove + name, [g, this](HTTP_Param) {
|
||
CHECK_JSON_PARAM
|
||
HTTP_REQUIRE_VALUE(index, params.try_get_number<int>("index"))
|
||
HTTP_REQUIRE_VALUE(ds, map.try_get(index))
|
||
ds->close();
|
||
ds->test_and_stop_thread();
|
||
|
||
HTTP_REQUIRE_TRUE(map.remove(index), "index")
|
||
g->save();
|
||
res->setBody(warp(to_json()).to_json_string());
|
||
g->mode_acs.source_feed_relation_config.set_need_refresh();
|
||
});
|
||
|
||
svr.Post(api + svr.rise + name, [g, this](HTTP_Param) {
|
||
CHECK_JSON_PARAM
|
||
HTTP_REQUIRE_VALUE(index, params.try_get_number<int>("index"))
|
||
HTTP_REQUIRE_TRUE(map.swap(index, index - 1), "index")
|
||
g->save();
|
||
res->setBody(warp(to_json()).to_json_string());
|
||
g->mode_acs.source_feed_relation_config.set_need_refresh();
|
||
});
|
||
svr.Post(api + svr.fall + name, [g, this](HTTP_Param) {
|
||
CHECK_JSON_PARAM
|
||
HTTP_REQUIRE_VALUE(index, params.try_get_number<int>("index"))
|
||
HTTP_REQUIRE_TRUE(map.swap(index, index + 1), "index")
|
||
g->save();
|
||
res->setBody(warp(to_json()).to_json_string());
|
||
g->mode_acs.source_feed_relation_config.set_need_refresh();
|
||
});
|
||
}
|
||
void Data_Source::refresh_data_feed_key_list() {
|
||
//std::cout << key << " refresh_data_feed_key_list" << std::endl;
|
||
std::vector<std::string> tmp;
|
||
for (auto& relation : Global::instance()->mode_acs.source_feed_relation_config.map.list()) {
|
||
if (!relation->enable) continue;
|
||
if (relation->type == "One_to_One_Relation") {
|
||
auto t = dynamic_cast<One_to_One_Relation*>(relation.get());
|
||
|
||
//std::cout << VAR_STR_2(t->source_key, this->key) << " refresh_data_feed_key_list" << std::endl;
|
||
if (t->source_key == this->key) {
|
||
tmp.push_back(t->feed_key);
|
||
}
|
||
} else if (relation->type == "First_Source_To_All_Feed_Relation") {
|
||
auto t = dynamic_cast<First_Source_To_All_Feed_Relation*>(relation.get());
|
||
std::shared_ptr<Data_Source> first = nullptr;
|
||
for (const auto& ds : Global::instance()->mode_acs.data_source_config.map.list()) {
|
||
if (ds->enable) {
|
||
first = ds;
|
||
break;
|
||
}
|
||
}
|
||
if (first->key == key) {
|
||
for (const auto& df : Global::instance()->mode_acs.data_feed_config.map.list()) {
|
||
if (df->enable) {
|
||
tmp.push_back(df->key);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
std::sort(tmp.begin(), tmp.end());
|
||
// 去重
|
||
auto last = std::unique(tmp.begin(), tmp.end());
|
||
tmp.erase(last, tmp.end());
|
||
|
||
|
||
|
||
{
|
||
std::vector<Cached_Source_Info> tmp_info;
|
||
|
||
// 创建 tmp_info 向量，将 tmp 的内容转化为 Cached_Source_Info 对象
|
||
for (const auto& key : tmp) {
|
||
tmp_info.push_back({key});
|
||
}
|
||
|
||
{
|
||
// 进入临界区，锁住 mutex，确保线程安全
|
||
std::lock_guard<std::mutex> lock(cdf_mtx);
|
||
|
||
// 1. 删除 tmp 中没有的 cached_data_feed_key_list 元素
|
||
auto it = cached_data_feed_key_list.begin();
|
||
while (it != cached_data_feed_key_list.end()) {
|
||
if (std::find_if(tmp_info.begin(), tmp_info.end(),
|
||
[&](const Cached_Source_Info& info) {
|
||
return info.key == it->key;
|
||
}) == tmp_info.end()) {
|
||
// 如果当前元素在 tmp 中找不到，删除它
|
||
it = cached_data_feed_key_list.erase(it);
|
||
} else {
|
||
++it;
|
||
}
|
||
}
|
||
|
||
// 2. 创建 tmp 中有但 cached_data_feed_key_list 没有的元素
|
||
for (const auto& tmp_item : tmp_info) {
|
||
auto found = std::find_if(cached_data_feed_key_list.begin(), cached_data_feed_key_list.end(),
|
||
[&](const Cached_Source_Info& cached_item) {
|
||
return cached_item.key == tmp_item.key;
|
||
});
|
||
if (found == cached_data_feed_key_list.end()) {
|
||
// 如果 tmp_item 不在 cached_data_feed_key_list 中，添加它
|
||
cached_data_feed_key_list.push_back(tmp_item);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
void Data_Source::push_to_feed(const std::shared_ptr<SSR::Msg>& msg) {
|
||
if (need_refresh_data_feed_key_list) {
|
||
refresh_data_feed_key_list();
|
||
need_refresh_data_feed_key_list = false;
|
||
}
|
||
auto& other_size = Global::instance()->mode_acs.mode_other_max_num;
|
||
auto& s_size = Global::instance()->mode_acs.mode_other_max_num;
|
||
std::vector<std::string> list;
|
||
|
||
|
||
|
||
for (const auto& item : cached_data_feed_key_list) {
|
||
auto& key = item.key;
|
||
auto opt_feed = Global::instance()->mode_acs.data_feed_config.map.get(key);
|
||
if (!opt_feed.has_value()) {
|
||
std::cout << "wrong data feed: " << key << std::endl;
|
||
break;
|
||
}
|
||
std::shared_ptr<Data_Feed>& feed = opt_feed.value();
|
||
if (!feed->enable) continue;
|
||
|
||
|
||
if (msg->type == SSR::Mode_Msg::T::S7 || msg->type == SSR::Mode_Msg::T::S14) {
|
||
if (feed->msg_buffer.mode_s_msg_num > s_size) {
|
||
continue;
|
||
} else {
|
||
++feed->msg_buffer.mode_s_msg_num;
|
||
}
|
||
} else {
|
||
if (feed->msg_buffer.mode_other_msg_num > other_size) {
|
||
continue;
|
||
} else {
|
||
++feed->msg_buffer.mode_other_msg_num;
|
||
}
|
||
}
|
||
std::optional<std::string> binary = convert_to_send_format(this, feed, msg);
|
||
if (binary.has_value()) {
|
||
feed->msg_buffer.push(binary.value());
|
||
}
|
||
}
|
||
}
|
||
|