#include "Local_Server/Data_Feed/Data_Feed.h" #include "../server/Global.h" #include bool Data_Feed::registered() { auto all_feed = Global::instance()->mode_acs.data_feed_config.map.list(); for (auto& item : all_feed) { if (item.get() == this) { return true; } } return false; } asio::awaitable Data_Feed_UDP_Server::handle_in_loop_coro() { co_await svr.tick_coro(); co_return; } asio::awaitable Data_Feed_UDP_Server::send_coro(std::string_view data) { co_await svr.write_to_all_clients_coro(data); co_return; } asio::awaitable Data_Feed_UDP_Server::_open() { svr.set_bind_address("0.0.0.0", port); svr.create(); server_logger->c_debug({}, {}, to_string() + "开启!"); co_return; } JSON Data_feed_Config::get_feed(std::string_view key) { auto ret = map.get(std::string(key)); if (ret.has_value()) { auto r = ret.value().get(); return r->to_json(); } return {nullptr}; } void Data_feed_Config::server(Global *g) { auto& svr = g->svr; auto& api = g->api; std::string name = "data_feed"; svr.Post(api + "get_data_feed_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_json(); } res->setBody(warp(ret).to_json_string()); }); svr.Post(api + "update_data_feed_config", [this](HTTP_Param) { CHECK_JSON_PARAM Get_J(packet_byte_size) Get_J(empty_wait_milliseconds) res->setBody(warp(to_json()).to_json_string()); Global::save(); }); svr.Post(api + "get_data_feed_config", [this](HTTP_Param) { res->setBody(warp(to_json()).to_json_string()); }); svr.Post(api + "get_data_feed_server_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()) { auto dfs = dynamic_cast(t.value().get()); if (dfs) { ret = dfs->get_clients_json(); } auto dfs2 = dynamic_cast(t.value().get()); if (dfs2) { ret = dfs2->get_clients_json(); } } res->setBody(warp(ret).to_json_string()); }); svr.Post(api + svr.search + name, [this](HTTP_Param) { CHECK_JSON_PARAM HTTP_REQUIRE_VALUE(type, params.try_get_string("key")) res->setBody(warp(get_feed(type)).to_json_string()); }); svr.Post(api + svr.update + name, [g, this](HTTP_Param) { CHECK_JSON_PARAM HTTP_REQUIRE_VALUE(key, params.try_get_string("key")) HTTP_REQUIRE_VALUE(df, map.get(key)) // 直接关闭 // t->close(); df->from_json(¶ms); Global::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("index")) HTTP_REQUIRE_PTR(data, params.get("data")) HTTP_REQUIRE_VALUE(t, data->try_get_string("type")) auto df = create_data_feed_from_type(t); df->from_json(data); HTTP_REQUIRE_VALUE(enable, data->try_get_bool("enable")) HTTP_REQUIRE_TRUE(map.insert(index, df), "index") res->setBody(warp(to_json()).to_json_string()); Global::save(); g->mode_acs.source_feed_relation_config.set_need_refresh(); }); svr.Post(api + svr.remove + name, [g, this](HTTP_Param) { CHECK_JSON_PARAM HTTP_REQUIRE_VALUE(index, params.try_get_number("index")) HTTP_REQUIRE_VALUE(df, map.try_get(index)) df->async_stop(); 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("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("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 BIN_Msg_Buffer::push(std::string_view msg) { auto output_packet_size = Global::instance()->mode_acs.data_feed_config.packet_byte_size.load(); std::lock_guard g(mtx); auto& pool = Global::instance()->mode_acs.data_feed_config.pool_; if (msg_list_cache.empty() || msg_list_cache.back()->size() > output_packet_size) { auto t = pool.get(); t->append(msg); msg_list_cache.push_back(t); } else { msg_list_cache.back()->append(msg); } } const std::vector& BIN_Msg_Buffer::get_all() { std::lock_guard g(mtx); std::swap(msg_list_cache, msg_list); msg_list_cache.clear(); mode_s_msg_num = 0; mode_other_msg_num = 0; return msg_list; } BIN_Msg_Buffer::BIN_Msg_Buffer() { msg_list_cache.reserve(8 * 1024); msg_list.reserve(8 * 1024); } Psc::JSON BIN_Msg_Buffer::state_json() { auto output_packet_size = Global::instance()->mode_acs.data_feed_config.packet_byte_size.load(); size_t size, cache_size, pool_capacity, pool_free_count; { std::lock_guard g(mtx); auto& pool = Global::instance()->mode_acs.data_feed_config.pool_; size = msg_list.size(); cache_size = msg_list_cache.size(); pool_capacity = pool.capacity(); pool_free_count = pool.free_count(); } return VAR_JSON_7(mode_s_msg_num, mode_other_msg_num, output_packet_size, cache_size, size, pool_capacity, pool_free_count); }