585 lines
25 KiB
C++
585 lines
25 KiB
C++
#include "Data_Source_Handler.h"
|
||
#include "Data_Source.h"
|
||
#include "Local_Server/server/Global.h"
|
||
#include <string_view>
|
||
#include "Local_Server/server/io_coro.h"
|
||
using namespace Psc;
|
||
std::shared_ptr<Data_Source> ds(Data_Source_Handler* dsh) {
|
||
return dynamic_cast<Data_Source*>(dsh)->that();
|
||
}
|
||
std::shared_ptr<SSR::Msg> Data_Source_Handler::create_msg(std::string_view packet) {
|
||
auto source = ds(this);
|
||
if (packet[0] != 0x1a) {
|
||
std::cout << "first:" << mem2hex(std::string(packet)) << std::endl;
|
||
return nullptr;
|
||
}
|
||
std::string error_len = source->key + " " + LOG_POS_SIMPLE +
|
||
std::string(" ") + "mode_s_error_length";
|
||
if (packet.size() < 2) return nullptr;
|
||
auto mt = packet[1];
|
||
if (mt == SSR::Msg::AC) {
|
||
if (packet.size() != SSR::Msg::AC_len) {
|
||
auto packet_hex = mem2hex(std::string(packet));
|
||
SSR::mode_s_logger->c_debug(
|
||
error_len, {},
|
||
std::to_string(mt) + " size:" + std::to_string(packet.size()) +
|
||
" hex:" + packet_hex +
|
||
" should:" + std::to_string(SSR::Msg::AC_len));
|
||
mode_ac_statistic.add_length_error();
|
||
return nullptr;
|
||
}
|
||
mode_ac_statistic.add();
|
||
return std::make_shared<SSR::Mode_AC_Msg>(source, packet);
|
||
}
|
||
if (mt == SSR::Msg::S7) {
|
||
if (packet.size() != SSR::Mode_S_Msg::S7_len) {
|
||
auto packet_hex = mem2hex(std::string(packet));
|
||
SSR::mode_s_logger->c_debug(
|
||
error_len, {},
|
||
std::to_string(mt) + " size:" + std::to_string(packet.size()) +
|
||
" hex:" + packet_hex +
|
||
" should:" + std::to_string(SSR::Msg::S7_len));
|
||
mode_s_statistic.add_length_error(
|
||
"[length error] Msg::S7 handle_mode_s_source");
|
||
return nullptr;
|
||
}
|
||
return std::make_shared<SSR::Mode_S_Msg>(source, packet);
|
||
}
|
||
if (mt == SSR::Msg::S14) {
|
||
if (packet.size() != SSR::Msg::S14_len) {
|
||
auto packet_hex = mem2hex(std::string(packet));
|
||
SSR::mode_s_logger->c_debug(
|
||
error_len, {},
|
||
std::to_string(mt) + " size:" + std::to_string(packet.size()) +
|
||
" hex:" + packet_hex +
|
||
" should:" + std::to_string(SSR::Msg::S14_len));
|
||
mode_s_statistic.add_length_error(
|
||
"[length error] Msg::S14 handle_mode_s_source");
|
||
return nullptr;
|
||
}
|
||
return std::make_shared<SSR::Mode_S_Msg>(source, packet);
|
||
}
|
||
if (mt == SSR::Msg::Radarcape_status) {
|
||
return nullptr; // 不知道如何解析跳过
|
||
if (packet.size() != SSR::Msg::Radarcape_status_len) {
|
||
auto packet_hex = mem2hex(std::string(packet));
|
||
SSR::mode_s_logger->c_debug(
|
||
error_len, {},
|
||
std::to_string(mt) + " size:" + std::to_string(packet.size()) +
|
||
" hex:" + packet_hex +
|
||
" should:" + std::to_string(SSR::Msg::Radarcape_status_len));
|
||
mode_s_statistic.add_length_error(
|
||
"[length error] Msg::Radarcape_status_len handle_mode_s_source");
|
||
return nullptr;
|
||
}
|
||
return std::make_shared<SSR::Msg>(source, packet);
|
||
}
|
||
if (mt == SSR::Msg::HULC_Status) {
|
||
if (packet.size() != SSR::Msg::HULC_len) {
|
||
auto packet_hex = mem2hex(std::string(packet));
|
||
// 找不到协议 size:23 hex:1A34195F0000001500B502FF06F423CE50000090000000
|
||
// should:5
|
||
SSR::mode_s_logger->c_debug(
|
||
error_len, {},
|
||
Psc::to_string(source->type) + " HULC size:" +
|
||
std::to_string(packet.size()) + " hex:" + packet_hex);
|
||
return nullptr;
|
||
}
|
||
return std::make_shared<SSR::Msg>(source, packet);
|
||
}
|
||
// std::cout << "unknown_type:" << memory2hex(packet);
|
||
rotating_log("unknown_type", mem2hex(std::string(packet)));
|
||
return nullptr;
|
||
}
|
||
extern bool data_source_debug;
|
||
size_t Data_Source_Handler::process_mode_acs_data(std::string_view origin_data) {
|
||
// 先读取原始数据 注意多复制了一遍
|
||
std::string mode_data(origin_data);
|
||
std::string name = type + "_" + key;
|
||
if (data_source_debug) {
|
||
std::cout << "before origin_data_transform_mode_data " << VAR_STR_2(name, mode_data.size()) << std::endl;
|
||
}
|
||
origin_data_transform_mode_data(mode_data);
|
||
if (data_source_debug) {
|
||
std::cout << " after origin_data_transform_mode_data " << VAR_STR_2(name, mode_data.size()) <<
|
||
key << std::endl;
|
||
}
|
||
static Value_Growth_Multi_T mt;
|
||
auto t = key;
|
||
if (data_source_debug) {
|
||
std::cout << name << " 读取到:" << VAR_STR_2(key, mode_data.size()) << " " << std::endl;
|
||
}
|
||
if (mt.test(key, mode_data.size())) {
|
||
std::cout << VAR_STR_2(key, mode_data.size()) << " 数据增长过快，可能内存积压" << std::endl;
|
||
}
|
||
size_t ret = 0;
|
||
if (mode_data.empty()) {
|
||
return ret;
|
||
}
|
||
auto source = ds(this);
|
||
auto size = mode_data.size();
|
||
read_speed.update(size);
|
||
value_statistics.update(size);
|
||
// 预处理数据
|
||
static bool mode_s_console = Global::instance()->console_config.mode_s_console;
|
||
static bool record_playback = Global::instance()->console_config.record_playback;
|
||
if (record_playback) {
|
||
pure_log(get_exe_dir() + "/playback/" + source->key + "_playback.dat",
|
||
mem2hex(mode_data, true, " ") + "\n");
|
||
}
|
||
if (mode_s_console) {
|
||
std::cout << source->key + " read:[mode_s_serial]:" << mem2hex(mode_data)
|
||
<< std::endl;
|
||
}
|
||
auto handle_packet = [this, source, &ret](std::string& packet) {
|
||
ret++;
|
||
auto msg = create_msg(packet);
|
||
if (!msg) return;
|
||
source->push_to_feed(msg);
|
||
auto mt = msg->type;
|
||
bool mode_s = mt == SSR::Msg::S7 || mt == SSR::Msg::S14;
|
||
if (mt == SSR::Msg::HULC_Status) {
|
||
source->handle_HULC(packet);
|
||
}
|
||
else if (mt == SSR::Msg::Radarcape_status) {
|
||
auto radarcape_msg = SSR::create_Radarcape_STATUS_Message(packet);
|
||
std::cout << radarcape_msg.toJson().to_json_string() << std::endl;
|
||
SSR::mode_s_logger->debug("Radarcape_status/radarcape", {},
|
||
radarcape_msg.toJson().to_json_string());
|
||
}
|
||
else if (mode_s) {
|
||
// 拓展点
|
||
handle_mode_s(std::dynamic_pointer_cast<SSR::Mode_S_Msg>(msg));
|
||
}
|
||
};
|
||
auto f = [this, source, handle_packet](std::string& packet) {
|
||
auto& wait = Global::instance()->mode_acs.wait_process_msg;
|
||
if (wait) {
|
||
handle_packet(packet);
|
||
}
|
||
else {
|
||
auto co = Coro::instance();
|
||
auto executor = co->process_data->get_executor();
|
||
asio::post(executor, [handle_packet, packet = std::move(packet)]() mutable {
|
||
handle_packet(packet);
|
||
});
|
||
}
|
||
};
|
||
SSR::Binary_Format_handle_buffer(source->buffer, mode_data, f);
|
||
return ret;
|
||
}
|
||
void Data_Source_Handler::handle_mode_s(std::shared_ptr<SSR::Mode_S_Msg> mode_s_msg) {
|
||
auto source = ds(this);
|
||
std::string t = mode_s_msg->mlat_timestamp.to_memory() +
|
||
mode_s_msg->signal_level + Psc::hex2mem(mode_s_msg->msg_hex);
|
||
auto p = mode_s_msg->packet.substr(2);
|
||
if (t != p) {
|
||
std::cout << mem2hex(t, true, " ") << std::endl;
|
||
std::cout << mem2hex(mode_s_msg->packet, true, " ") << std::endl;
|
||
std::cout << VAR_STR_2(t, mode_s_msg->packet) << std::endl;
|
||
}
|
||
source->push_to_feed(mode_s_msg);
|
||
auto& cfg = Global::instance()->mode_acs;
|
||
bool time_space_filter = cfg.time_space_filter.load();
|
||
bool speed_filter = cfg.speed_filter.load();
|
||
bool use_system_time = source->ignore_msg_time.load();
|
||
|
||
SSR::ADS_B_T::Constraint air_constraint{cfg.max_speed_m_s, cfg.air_pos_timeout, SSR::cpr_cb, time_space_filter, speed_filter, use_system_time};
|
||
SSR::ADS_B_T::Constraint surface_constraint{cfg.max_speed_m_s, cfg.surface_pos_timeout, SSR::cpr_cb, time_space_filter, speed_filter, use_system_time};
|
||
SSR::parse_mode_s_bin(source.get(), mode_s_msg,
|
||
source->base_station.get_pos(), air_constraint,
|
||
surface_constraint);
|
||
auto base = source->get_aircraft(mode_s_msg->icao);
|
||
if (base) {
|
||
auto derived = std::dynamic_pointer_cast<Aircraft>(base);
|
||
derived->refresh_external_database_info();
|
||
}
|
||
if (Global::instance()->mlat.merge) {
|
||
auto& mh = Global::instance()->mlat_handler;
|
||
mh.push(mode_s_msg);
|
||
auto tt = mh.get_all();
|
||
if (!tt.empty()) {
|
||
pure_log("@/logs/mlat.log",
|
||
"receive[" + std::to_string(tt.size()) + "]:\n");
|
||
for (Mlat_MSG& t : tt) {
|
||
std::ostringstream oss;
|
||
oss << std::fixed << std::setprecision(10);
|
||
oss << "===================================================\n";
|
||
int n = static_cast<int>(t.size());
|
||
auto icao = t.icao().value_or("null");
|
||
oss << "\tmlat_msg:" << n;
|
||
oss << "\ticao:" << icao;
|
||
oss << "\thex:" << t.msg_hex() << std::endl;
|
||
bool have_pos = false;
|
||
if (icao != "null") {
|
||
auto air = source->get_aircraft(icao);
|
||
if (air != nullptr) {
|
||
auto callsign = air->bds20_call_sign();
|
||
oss << "\tcall_sign：" << callsign.value_or("null") << std::endl;
|
||
auto o_pos = air->pos();
|
||
auto o_alt = air->altitude_meter();
|
||
if (o_pos.has_value() && o_alt.has_value()) {
|
||
have_pos = true;
|
||
auto pos = o_pos.value();
|
||
source->alt = o_alt.value();
|
||
double x, y, z;
|
||
SSR::CPR::WGS84_LBH_to_XYZ(pos.lon, pos.lat, source->alt, x, y,
|
||
z);
|
||
oss << "\tpos：[" << pos.lon << "," << pos.lat << ","
|
||
<< source->alt << "]" << std::endl;
|
||
oss << "\tXYZ：[" << x << "," << y << "," << z << "]"
|
||
<< std::endl;
|
||
}
|
||
}
|
||
}
|
||
for (int i = 0; i < t.size(); i++) {
|
||
auto& msg = t.list[i];
|
||
oss << "\t\tds:" << msg->source->get_key();
|
||
oss << "\ttime:" << msg->mlat_timestamp.to_string() << std::endl;
|
||
}
|
||
// if (have_pos) {
|
||
// std::cout << oss.str() << std::endl;
|
||
pure_log("@/logs/mlat.log", oss.str());
|
||
// }
|
||
// if (t.size() > 3 && have_pos && icao == "78139F") {
|
||
// size_t i = 1;
|
||
// auto range = hex_mlat_map.equal_range(t.msg_hex());
|
||
// // 遍历与key_to_find对应的所有元素
|
||
// for (auto it = range.first; it != range.second; ++it) {
|
||
// oss << "\t" << i++ << " " << it->second.msg << std::endl;
|
||
// }
|
||
// oss << "===================================================\n";
|
||
// std::cout << oss.str() << std::endl;
|
||
// pure_log("@/logs/icao/" + icao + "_mlat.log", oss.str(), true);
|
||
// }
|
||
}
|
||
}
|
||
}
|
||
}
|
||
void Data_Source_Handler::handle_HULC(std::string_view packet) {
|
||
std::uint8_t len = SSR::get_len(packet);
|
||
std::uint8_t id = SSR::get_id(packet);
|
||
auto msg = mem2hex(std::string(packet));
|
||
if (len != packet.size() - 4) {
|
||
SSR::mode_s_logger->debug("HULC/error_length", {},
|
||
"需要: " + std::to_string(len) +
|
||
" 当前: " + std::to_string(packet.size()) +
|
||
" hex: " + mem2hex(std::string(packet)));
|
||
return;
|
||
}
|
||
if (id == 1) {
|
||
// 状态消息
|
||
auto status_msg = SSR::create_HULC_Status_Message(packet);
|
||
bool gps_ok = status_msg.GPS_device_detected() && status_msg.GPS_valid() &&
|
||
status_msg.GPS_has_valid_fix();
|
||
if (gps_ok) {}
|
||
auto g = Global::instance();
|
||
base_station.set_msg(status_msg);
|
||
Log_Type type({}, {{"msg", std::string(msg)}});
|
||
SSR::mode_s_logger->debug("HULC/status", type,
|
||
status_msg.toJson().to_json_string());
|
||
if (gps_ok) {}
|
||
}
|
||
else if (id == 24) {
|
||
SSR::mode_s_logger->debug("HULC/reply", {}, msg);
|
||
}
|
||
else {
|
||
SSR::mode_s_logger->debug("HULC/unknown_id", {}, msg);
|
||
}
|
||
}
|
||
void Data_Source_Handler::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);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
std::optional<std::string> convert_to_send_format(Data_Source_Handler* ds,
|
||
const std::shared_ptr<Data_Feed>& feed,
|
||
const std::shared_ptr<SSR::Msg>& msg) {
|
||
auto& type = msg->type;
|
||
auto fs = dynamic_cast<File_Data_Source*>(msg->source.get());
|
||
if (fs) {
|
||
if (fs->play_mode == Play_Mode::analysis) {
|
||
// std::cout << fs->key << " analysis i ==" << i << std::endl;
|
||
}
|
||
}
|
||
// 出口输出的条件
|
||
auto& output_format = feed->output_format.type;
|
||
auto& use_mode_ac = feed->output_format.use_mode_ac;
|
||
auto& use_status = feed->output_format.use_status;
|
||
auto& mode_s_output_type = feed->output_format.mode_s_output_type;
|
||
// 计算出是否输出消息
|
||
std::optional<std::string> send_msg = std::nullopt;
|
||
if (type == SSR::Msg::HULC_Status) {
|
||
if (!use_status) return std::nullopt;
|
||
auto data = static_cast<SSR::Msg*>(msg.get());
|
||
if (output_format == Output_Data_Format::BIN ||
|
||
output_format == Output_Data_Format::BIN_ID) {
|
||
send_msg = SSR::packet_to_escape_format(data->packet);
|
||
}
|
||
}
|
||
else if (type == SSR::Msg::Radarcape_status) {
|
||
auto data = static_cast<SSR::Msg*>(msg.get());
|
||
if (output_format == Output_Data_Format::BIN ||
|
||
output_format == Output_Data_Format::BIN_ID) {
|
||
send_msg = SSR::packet_to_escape_format(data->packet);
|
||
}
|
||
}
|
||
else if (type == SSR::Msg::AC) {
|
||
if (!use_mode_ac) return std::nullopt;
|
||
auto data = static_cast<SSR::Mode_AC_Msg*>(msg.get());
|
||
std::string& msg_hex = data->msg_hex;
|
||
char signal_level = data->signal_level;
|
||
SSR::MLAT_timestamp& mlat_timestamp = data->mlat_timestamp;
|
||
if (output_format == Output_Data_Format::BIN) {
|
||
// send_msg = create_Binary_Format_memory(msg_hex, signal_level,
|
||
// &mlat_timestamp);
|
||
send_msg = SSR::packet_to_escape_format(msg->packet);
|
||
}
|
||
else if (output_format == Output_Data_Format::BIN_ID) {
|
||
// std::uint32_t id;
|
||
// {
|
||
// id = Global::instance()->mlat_server.id;
|
||
// }
|
||
// auto bin_id_packet = mode_s_msg_packet_to_server_packet(id,
|
||
// data->packet); send_msg = packet_to_escape_format(bin_id_packet);
|
||
}
|
||
else if (output_format == Output_Data_Format::AVR) {
|
||
send_msg = SSR::create_AVR_format(msg_hex);
|
||
}
|
||
else if (output_format == Output_Data_Format::AVR_MLAT) {
|
||
send_msg = create_MLAT_AVR_format(msg_hex, &mlat_timestamp);
|
||
}
|
||
}
|
||
else if (type == SSR::Msg::S7 || type == SSR::Msg::S14) {
|
||
auto s_msg = static_cast<SSR::Mode_S_Msg*>(msg.get());
|
||
std::string& icao = s_msg->icao;
|
||
SSR::Downlink_Format& df = s_msg->df;
|
||
std::string& msg_hex = s_msg->msg_hex;
|
||
auto& signal_level = s_msg->signal_level;
|
||
SSR::MLAT_timestamp& mlat_timestamp = s_msg->mlat_timestamp;
|
||
auto aircraft = ds->get_aircraft(s_msg->icao);
|
||
if (output_format == Output_Data_Format::SBS) {
|
||
bool sbs_out_put = false;
|
||
if (feed->output_format.sbs_only_pos) {
|
||
auto s = ds->get_aircraft(s_msg->icao);
|
||
if (s) {
|
||
auto pos = s->pos();
|
||
if (pos.has_value()) {
|
||
if (feed->sbs_flm.test(s_msg->icao)) {
|
||
sbs_out_put = true;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
else {
|
||
if (feed->sbs_flm.test(s_msg->icao)) {
|
||
sbs_out_put = true;
|
||
}
|
||
}
|
||
if (!sbs_out_put) return std::nullopt;
|
||
send_msg = "";
|
||
SSR::SBS_MSG msg_base;
|
||
if (!aircraft) return std::nullopt;
|
||
msg_base.set(aircraft.get());
|
||
for (int msg_t = 1; msg_t <= 8; ++msg_t) {
|
||
send_msg->append(msg_base.to_msg(msg_t) + "\n");
|
||
}
|
||
send_msg->append(msg_base.to_ID() + "\n");
|
||
send_msg->append(msg_base.to_AIR() + "\n");
|
||
send_msg->append(msg_base.to_CLK() + "\n");
|
||
send_msg->append(msg_base.to_SEL() + "\n");
|
||
send_msg->append(msg_base.to_STA() + "\n");
|
||
return send_msg;
|
||
}
|
||
bool DF_11_17_18 =
|
||
df == SSR::Downlink_Format::All_Call_Reply_11 ||
|
||
df == SSR::Downlink_Format::Extended_Squitter_17 ||
|
||
df == SSR::Downlink_Format::Extended_Squitter_Non_Transponder_18;
|
||
if (mode_s_output_type == Mode_S_Output_Type::DF_11_17_18 && !DF_11_17_18) return std::nullopt;
|
||
if (mode_s_output_type == Mode_S_Output_Type::NO_POS_Mode_S) {
|
||
if (aircraft && aircraft->pos().has_value()) {
|
||
return std::nullopt;
|
||
}
|
||
}
|
||
if (output_format == Output_Data_Format::BIN) {
|
||
send_msg = SSR::packet_to_escape_format(s_msg->packet);
|
||
// send_msg = create_Binary_Format_memory(msg_hex, signal_level,
|
||
// &mlat_timestamp);
|
||
}
|
||
else if (output_format == Output_Data_Format::BIN_ID) {
|
||
// std::uint32_t id;
|
||
// {
|
||
// id = Global::instance()->mlat_server.id;
|
||
// }
|
||
// auto bin_id_packet = mode_s_msg_packet_to_server_packet(id,
|
||
// data->packet); send_msg = packet_to_escape_format(bin_id_packet);
|
||
}
|
||
else if (output_format == Output_Data_Format::AVR) {
|
||
send_msg = SSR::create_AVR_format(msg_hex);
|
||
}
|
||
else if (output_format == Output_Data_Format::AVR_MLAT) {
|
||
send_msg = create_MLAT_AVR_format(msg_hex, &mlat_timestamp);
|
||
}
|
||
else {
|
||
std::cerr << "Unknown output format " << VAR_STR_1(output_format)
|
||
<< std::endl;
|
||
return std::nullopt;
|
||
}
|
||
}
|
||
return send_msg;
|
||
}
|
||
void Data_Source_Handler::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) {
|
||
// static Frequency_Limit fl;
|
||
// if (!fl.test()) {
|
||
// std::ostringstream oss;
|
||
// oss << "收到消息 ===" << VAR_STR_3(msg->type
|
||
// ,feed->msg_buffer.mode_s_msg_num, s_size) << std::endl; std::cout
|
||
// << oss.str() << std::endl;
|
||
// }
|
||
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());
|
||
}
|
||
}
|
||
}
|
||
Psc::JSON Data_Source_Handler::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_Handler::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;
|
||
}
|