104 lines
3.9 KiB
C++
104 lines
3.9 KiB
C++
#include "Local_Server/Data_Feed/Data_Feed.h"
|
|
#include <string_view>
|
|
#include "../server/Global.h"
|
|
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;
|
|
}
|
|
void Data_feed_Config::set_connections_need_refresh() {
|
|
for (const auto& source : Global::instance()->mode_acs.data_source_config.map.list())
|
|
source->need_refresh_data_feed_key_list = true;
|
|
}
|
|
bool Data_feed_Config::source_has_connection(std::string_view source_key) const {
|
|
for (const auto& feed : map.list()) {
|
|
if (feed->source_key == source_key)
|
|
return true;
|
|
}
|
|
return false;
|
|
}
|
|
asio::awaitable<void> Data_Feed_UDP_Server::handle_in_loop_coro() {
|
|
co_await svr.tick_coro();
|
|
co_return;
|
|
}
|
|
asio::awaitable<void> Data_Feed_UDP_Server::send_coro(std::string_view data) {
|
|
co_await svr.write_to_all_clients_coro(data);
|
|
co_return;
|
|
}
|
|
asio::awaitable<void> Data_Feed_UDP_Server::_open() {
|
|
const auto value = specific.read([](const auto& value) { return value; });
|
|
svr.set_bind_address("0.0.0.0", value.port);
|
|
svr.create();
|
|
server_logger->c_debug({}, {}, to_string() + "开启！");
|
|
co_return;
|
|
}
|
|
void Data_feed_Config::server(Global* g) {
|
|
auto& svr = g->svr;
|
|
auto& api = g->api;
|
|
svr.Post(api + "get_data_feed_state", [this](HTTP_Param) {
|
|
CHECK_JSON_PARAM
|
|
HTTP_REQUIRE_VALUE(key, params.try_get_string("key"))
|
|
auto value = map.get(key);
|
|
JSON ret{nullptr};
|
|
if (value.has_value())
|
|
ret = value.value()->get_state_json();
|
|
res->setBody(warp(ret).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 value = map.get(key);
|
|
JSON ret{nullptr};
|
|
if (value.has_value()) {
|
|
if (auto tcp = dynamic_cast<Data_Feed_TCP_Server*>(value.value().get()))
|
|
ret = tcp->get_clients_json();
|
|
if (auto udp = dynamic_cast<Data_Feed_UDP_Server*>(value.value().get()))
|
|
ret = udp->get_clients_json();
|
|
}
|
|
res->setBody(warp(ret).to_json_string());
|
|
});
|
|
}
|
|
void BIN_Msg_Buffer::push(std::string_view msg) {
|
|
auto output_packet_size = Global::instance()->mode_acs.data_feed_config.settings.member<&Data_feed_Config_Data::packet_byte_size>().read([](const auto& value) { return value; });
|
|
std::lock_guard<std::mutex> 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<std::string*>& BIN_Msg_Buffer::get_all() {
|
|
std::lock_guard<std::mutex> 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.settings.member<&Data_feed_Config_Data::packet_byte_size>().read([](const auto& value) { return value; });
|
|
size_t size, cache_size, pool_capacity, pool_free_count;
|
|
{
|
|
std::lock_guard<std::mutex> 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);
|
|
}
|