diff --git a/module/Local_Server/DSP/DSP_Config.cpp b/module/Local_Server/DSP/DSP_Config.cpp index c53996a..44f1d41 100644 --- a/module/Local_Server/DSP/DSP_Config.cpp +++ b/module/Local_Server/DSP/DSP_Config.cpp @@ -237,7 +237,7 @@ void DSP_Config::init_env() { } }); #else - manager.test_and_start_thread("iq回调处理线程", [this, handle_iq](std::atomic &running) { + manager.test_and_start_thread("iq回调处理", [this, handle_iq](std::atomic &running) { Frequency_Limit_ST limit(frame_rate); while (running) { auto list = iq_memory_buffer.get_all(); diff --git a/module/Local_Server/Data_Feed/Data_Feed.h b/module/Local_Server/Data_Feed/Data_Feed.h index 62b6c1a..a006074 100644 --- a/module/Local_Server/Data_Feed/Data_Feed.h +++ b/module/Local_Server/Data_Feed/Data_Feed.h @@ -385,5 +385,4 @@ struct Data_feed_Config { void server(Global* g); }; class Data_Source; -std::optional convert_to_send_format(Data_Source* ds, const std::shared_ptr& feed, const std::shared_ptr& msg); diff --git a/module/Local_Server/Data_Feed/data_feed_thread.cpp b/module/Local_Server/Data_Feed/data_feed_thread.cpp index f1b5b92..acfd701 100644 --- a/module/Local_Server/Data_Feed/data_feed_thread.cpp +++ b/module/Local_Server/Data_Feed/data_feed_thread.cpp @@ -11,127 +11,6 @@ Frequency_Limit too_many_msg_limit; Frequency_Limit flush_limit; -std::optional convert_to_send_format(Data_Source* ds, const std::shared_ptr& feed, const std::shared_ptr& msg) -{ - auto& type = msg->type; - auto fs = dynamic_cast(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 send_msg = std::nullopt; - if (type == SSR::Msg::HULC_Status) { - if (!use_status) return std::nullopt; - auto data = static_cast(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(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(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(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; -} ucoro::awaitable data_feed_thread_coro(std::atomic& running) { auto g = Global::instance(); @@ -225,7 +104,7 @@ ucoro::awaitable data_feed_thread_coro(std::atomic& running) { co_return; } -void data_feed_thread(std::atomic& running) { +void io_coro(std::atomic& running) { try { ucoro::sync_await(data_feed_thread_coro(running)); diff --git a/module/Local_Server/Data_Source/Data_Source.cpp b/module/Local_Server/Data_Source/Data_Source.cpp index 2106f49..ae36a17 100644 --- a/module/Local_Server/Data_Source/Data_Source.cpp +++ b/module/Local_Server/Data_Source/Data_Source.cpp @@ -110,10 +110,7 @@ void File_Data_Source::before_handle_msg(std::shared_ptr& msg) { ucoro::awaitable 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; }; @@ -146,41 +143,6 @@ 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 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 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 readLines(const std::string& path) { @@ -344,9 +306,6 @@ std::string get_true_from_raw_line(Data_Source* ds, File_Data_Type data_type, in std::string info = log_info; std::string result = ds->last_char; ds->last_char = ""; - - - for (char c : info) { if (std::isxdigit(c)) { // 会判断字符 c 是否是十六进制数字字符 @@ -367,26 +326,6 @@ std::string get_true_from_raw_line(Data_Source* ds, File_Data_Type data_type, in 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(); @@ -526,13 +465,15 @@ std::vector File_Data_Source::readBinaryFileAsString(const std::str } -ucoro::awaitable File_Data_Source::read_coro() { +void File_Data_Source::origin_data_transform_mode_data(std::string& mode_data) +{ std::lock_guard g(mtx); + assert(mode_data.size() == 0); int ret_index = 0; if (play_mode != Play_Mode::analysis) { std::optional 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()) : ""; + mode_data = log_info.has_value() ? get_true_from_raw_line(this, data_type, ret_index, log_info.value()) : ""; } else { int num = 0; @@ -551,7 +492,7 @@ ucoro::awaitable File_Data_Source::read_coro() { log_info = get_raw_line(ret_index); } // std::cout << "size == " << t.size() << std::endl; - co_return ret; + mode_data = ret; } } @@ -590,11 +531,11 @@ void File_Data_Source::handle_mode_s(std::shared_ptr msg) { } -ucoro::awaitable Dll_Data_Source::read_coro() { - +void Dll_Data_Source::origin_data_transform_mode_data(std::string& mode_data) +{ auto start = std::chrono::steady_clock::now(); - if (read_func_ptr == nullptr) co_return ""; + if (read_func_ptr == nullptr) return; auto buf = reinterpret_cast(buffer.data()); auto len = read_func_ptr(buf, buffer_size); @@ -606,7 +547,7 @@ ucoro::awaitable Dll_Data_Source::read_coro() { auto end = std::chrono::steady_clock::now(); auto cost = std::chrono::duration_cast(end - start).count(); //std::cout << "Dll_Data_Source::read cost: " << cost << " us\n"; - co_return ""; + return; } auto ret = get_true_from_raw_line(this, data_type, -1, data); @@ -614,9 +555,9 @@ ucoro::awaitable Dll_Data_Source::read_coro() { auto end = std::chrono::steady_clock::now(); auto cost = std::chrono::duration_cast(end - start).count(); - // std::cout << "Dll_Data_Source::read cost: " << cost << " us\n"; + // std::cout << "Dll_Data_Source::read cost: " << cost << " us\n"; - co_return ret; + mode_data = ret; } @@ -629,8 +570,8 @@ void Shared_Memory_Data_Source::_close() { sm.reset(); } - -ucoro::awaitable Shared_Memory_Data_Source::read_coro() { +void Shared_Memory_Data_Source::origin_data_transform_mode_data(std::string& mode_data) +{ auto size = sm->shm.size(); size_t length = size; std::string data; @@ -638,7 +579,7 @@ ucoro::awaitable Shared_Memory_Data_Source::read_coro() { bool ok = sm->read((uint8_t *)data.data(), length); if (!ok) { - co_return ""; + mode_data = ""; } if (length != size) { @@ -648,9 +589,11 @@ ucoro::awaitable Shared_Memory_Data_Source::read_coro() { std::cout << "read:" << data << std::endl; } auto ret = get_true_from_raw_line(this, data_type, -1, data); - co_return ret; + mode_data = ret; } + + void Data_Source_Config::server(Global* g) { auto& svr = g->svr; auto& api = g->api; @@ -755,124 +698,5 @@ void Data_Source_Config::server(Global* g) { 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 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(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(relation.get()); - std::shared_ptr 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 tmp_info; - - // 创建 tmp_info 向量，将 tmp 的内容转化为 Cached_Source_Info 对象 - for (const auto& key : tmp) { - tmp_info.push_back({key}); - } - - { - // 进入临界区，锁住 mutex，确保线程安全 - std::lock_guard 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& 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 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& 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 binary = convert_to_send_format(this, feed, msg); - if (binary.has_value()) { - feed->msg_buffer.push(binary.value()); - } - } -} - diff --git a/module/Local_Server/Data_Source/Data_Source.h b/module/Local_Server/Data_Source/Data_Source.h index 91e41ec..a825a13 100644 --- a/module/Local_Server/Data_Source/Data_Source.h +++ b/module/Local_Server/Data_Source/Data_Source.h @@ -70,23 +70,16 @@ class Data_Source : public std::enable_shared_from_this, public Dat public: ~Data_Source() override = default; std::shared_ptr that(); - virtual ucoro::awaitable read_coro() = 0; + virtual ucoro::awaitable read_coro() {co_return "";}; virtual Psc::JSON get_custom_state_json() = 0; Psc::JSON get_state(); std::string last_char; // 用于处理奇数字节的情况 // std::shared_ptr parse_format; - Data_Source(); std::shared_ptr last_prase_msg = nullptr; - - - virtual ucoro::awaitable handle_in_loop_coro(){co_return;} - virtual ucoro::awaitable source_step_coro(); - void refresh_data_feed_key_list(); - void push_to_feed(const std::shared_ptr &msg); std::atomic enable{}; std::string type; std::atomic base_station_show{}; @@ -172,19 +165,13 @@ public: } } - struct Cached_Source_Info { - std::string key; - Psc::Value_Statistics mode_s_cache_num; - Psc::Value_Statistics mode_other_cache_num; - }; + std::string buffer; - std::atomic need_refresh_data_feed_key_list = true; - Psc::JSON get_all_connect_feed_status(); - Psc::JSON get_all_connect_feed(); + protected: - std::mutex cdf_mtx; - std::vector cached_data_feed_key_list{}; + + bool _open_ = false; virtual bool _open() = 0; virtual void _close() = 0; @@ -323,6 +310,7 @@ struct File_Data_Source : public Data_Source { ~File_Data_Source() override = default; void before_handle_msg(std::shared_ptr& msg) override; + struct Cache_Msg { explicit Cache_Msg(std::shared_ptr msg) : msg(std::move(msg)) {} SSR::Play_Back_Time_Point time() const { @@ -356,7 +344,8 @@ struct File_Data_Source : public Data_Source { return ret; } - ucoro::awaitable read_coro() override; + void origin_data_transform_mode_data(std::string& data) override; + void handle_mode_s(std::shared_ptr msg) override; @@ -429,8 +418,7 @@ struct Dll_Data_Source : Data_Source { state = "加载成功"; return true; } - - ucoro::awaitable read_coro(); + void origin_data_transform_mode_data(std::string& data) override; }; @@ -454,7 +442,7 @@ struct Shared_Memory_Data_Source : Data_Source { bool _open() override; void _close() override; - ucoro::awaitable read_coro() override; + void origin_data_transform_mode_data(std::string& data) override; protected: std::unique_ptr sm; std::string shared_memory_name; diff --git a/module/Local_Server/Data_Source/Data_Source_Handler.cpp b/module/Local_Server/Data_Source/Data_Source_Handler.cpp index 7930c3f..ee04915 100644 --- a/module/Local_Server/Data_Source/Data_Source_Handler.cpp +++ b/module/Local_Server/Data_Source/Data_Source_Handler.cpp @@ -94,7 +94,15 @@ std::shared_ptr Data_Source_Handler::create_msg( return nullptr; } -void Data_Source_Handler::process_mode_acs_data(const std::string& mode_data) { +void Data_Source_Handler::process_mode_acs_data(const std::string& origin_data) { + // 先读取原始数据 注意多复制了一遍 + auto mode_data = origin_data; + origin_data_transform_mode_data(mode_data); + + if (mode_data.empty()) { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + auto source = ds(this); auto size = mode_data.size(); read_speed.update(size); @@ -262,3 +270,286 @@ void Data_Source_Handler::handle_HULC(const std::string& packet) { 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 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(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(relation.get()); + std::shared_ptr 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 tmp_info; + + // 创建 tmp_info 向量，将 tmp 的内容转化为 Cached_Source_Info 对象 + for (const auto& key : tmp) { + tmp_info.push_back({key}); + } + + { + // 进入临界区，锁住 mutex，确保线程安全 + std::lock_guard 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 convert_to_send_format(Data_Source_Handler* ds, const std::shared_ptr& feed, const std::shared_ptr& msg) +{ + auto& type = msg->type; + auto fs = dynamic_cast(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 send_msg = std::nullopt; + if (type == SSR::Msg::HULC_Status) { + if (!use_status) return std::nullopt; + auto data = static_cast(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(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(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(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& 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 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& 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 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 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 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; +} \ No newline at end of file diff --git a/module/Local_Server/Data_Source/Data_Source_Handler.h b/module/Local_Server/Data_Source/Data_Source_Handler.h index be1522e..434c849 100644 --- a/module/Local_Server/Data_Source/Data_Source_Handler.h +++ b/module/Local_Server/Data_Source/Data_Source_Handler.h @@ -19,7 +19,10 @@ class Data_Source_Handler : public DataBase { public: std::shared_ptr create_msg(const std::string& packet); + + virtual void origin_data_transform_mode_data(std::string& origin_data){}; virtual void before_handle_msg(std::shared_ptr& msg){}; + virtual void handle_mode_s(std::shared_ptr msg); void handle_HULC(const std::string& packet); void process_mode_acs_data(const std::string& mode_data); @@ -27,7 +30,21 @@ public: Mode_S_Statistic_Data mode_s_statistic; Psc::Speed_Statistics read_speed; Psc::Value_Statistics value_statistics; - virtual ~Data_Source_Handler(){}; + ~Data_Source_Handler() override {}; + + // 推送到 data_feed 相关代码 + void push_to_feed(const std::shared_ptr &msg); + void refresh_data_feed_key_list(); + struct Cached_Source_Info { + std::string key; + Psc::Value_Statistics mode_s_cache_num; + Psc::Value_Statistics mode_other_cache_num; + }; + std::atomic need_refresh_data_feed_key_list = true; + std::vector cached_data_feed_key_list{}; + Psc::JSON get_all_connect_feed_status(); + Psc::JSON get_all_connect_feed(); + std::mutex cdf_mtx; }; diff --git a/module/Local_Server/server/Global.h b/module/Local_Server/server/Global.h index 1d1b25c..fda26ab 100644 --- a/module/Local_Server/server/Global.h +++ b/module/Local_Server/server/Global.h @@ -291,7 +291,7 @@ public: DELETE_COPY(Global) }; -void data_feed_thread(std::atomic& running); +void io_coro(std::atomic& running); void handle_buffer_muti_start(std::string& buffer, const std::string& data, const std::set& prefix_list, diff --git a/module/Local_Server_main.cpp b/module/Local_Server_main.cpp index bc6bbd2..0dcfa0b 100644 --- a/module/Local_Server_main.cpp +++ b/module/Local_Server_main.cpp @@ -171,7 +171,7 @@ int wyc_main(int argc, char *argv[]) { } } }); - manager.test_and_start_thread("data_feed_thread", data_feed_thread); + manager.test_and_start_thread("io_coro 协程线程", io_coro); static int t = Global::instance()->mode_acs.read_milliseconds; for (std::shared_ptr &source : Global::instance()->mode_acs.data_source_config.map.list()) {