ae/ECAP_Server
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

初步修改

Browse Source
This commit is contained in:
种太阳的热带美短
2026-08-09 14:13:35 +08:00
parent 1e1e77e41a
commit 6fe0bed78e
26 changed files with 2007 additions and 681 deletions
Download Patch File Download Diff File Expand all files Collapse all files
module/Local_Server/Data_Feed/Data_Feed.cpp
+13 -81
View File
@@ -19,7 +19,8 @@ asio::awaitable<void> Data_Feed_UDP_Server::send_coro(std::string_view data) {
co_return;
}
asio::awaitable<void> Data_Feed_UDP_Server::_open() {
svr.set_bind_address("0.0.0.0", port);
const auto value = specific.snapshot();
svr.set_bind_address("0.0.0.0", value.port);
svr.create();
server_logger->c_debug({}, {}, to_string() + "开启！");
co_return;
@@ -35,100 +36,31 @@ JSON Data_feed_Config::get_feed(std::string_view key) {
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);
auto value = map.get(key);
JSON ret{nullptr};
if (t.has_value()) {
ret = t.value()->get_state_json();
}
if (value.has_value())
ret = value.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);
auto value = map.get(key);
JSON ret{nullptr};
if (t.has_value()) {
auto dfs = dynamic_cast<Data_Feed_TCP_Server*>(t.value().get());
if (dfs) {
ret = dfs->get_clients_json();
}
auto dfs2 = dynamic_cast<Data_Feed_UDP_Server*>(t.value().get());
if (dfs2) {
ret = dfs2->get_clients_json();
}
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());
});
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
std::cout << "Data_Feed:" << req->path() << that_json->to_json_string();
HTTP_REQUIRE_VALUE(key, params.try_get_string("key"))
HTTP_REQUIRE_VALUE(df, map.get(key))
// 直接关闭
// t->close();
df->from_json(&params, true);
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
std::cout << "Data_Feed:" << req->path() << that_json->to_json_string();
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::shared_ptr<Data_Feed> df = create_data_feed_from_type(t);
df->from_json(data, true);
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
std::cout << "Data_Feed:" << req->path() << that_json->to_json_string();
HTTP_REQUIRE_VALUE(index, params.try_get_number<int>("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<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 BIN_Msg_Buffer::push(std::string_view msg) {
auto output_packet_size = Global::instance()->mode_acs.data_feed_config.packet_byte_size.load();
auto output_packet_size = Global::instance()->mode_acs.data_feed_config.settings.member<&Data_feed_Config_Data::packet_byte_size>().snapshot();
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) {
@@ -153,7 +85,7 @@ BIN_Msg_Buffer::BIN_Msg_Buffer() {
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();
auto output_packet_size = Global::instance()->mode_acs.data_feed_config.settings.member<&Data_feed_Config_Data::packet_byte_size>().snapshot();
size_t size, cache_size, pool_capacity, pool_free_count;
{
std::lock_guard<std::mutex> g(mtx);
module/Local_Server/Data_Feed/Data_Feed.h
+142 -45
View File
@@ -7,16 +7,30 @@ enum class Output_Data_Format { AVR, AVR_MLAT, BIN, SBS, BIN_ID };
enum class Mode_S_Output_Type { DF_11_17_18, NO_POS_Mode_S, ALL_Mode_S };
struct Output_Format {
Output_Data_Format type = Output_Data_Format::BIN;
Psc::Copyable_Atomic<bool> use_status = false;
bool use_status = false;
Mode_S_Output_Type mode_s_output_type = Mode_S_Output_Type::DF_11_17_18;
Psc::Copyable_Atomic<bool> use_mode_ac = false;
Psc::Copyable_Atomic<bool> sbs_only_pos = false;
bool use_mode_ac = false;
bool sbs_only_pos = false;
PSC_USE_JSON
};
namespace adminive {
template <>
struct Type_Descriptor<Output_Format> {
static auto get() {
using T = Output_Format;
return object<T>("output_format", "输出格式",
ADMINIVE_FIELD_LABEL(T, type, "数据格式").creatable().editable().select_input(),
ADMINIVE_FIELD_LABEL(T, use_status, "输出状态消息").creatable().editable().boolean_input(),
ADMINIVE_FIELD_LABEL(T, mode_s_output_type, "Mode S 输出类型").creatable().editable().select_input(),
ADMINIVE_FIELD_LABEL(T, use_mode_ac, "输出 Mode AC").creatable().editable().boolean_input(),
ADMINIVE_FIELD_LABEL(T, sbs_only_pos, "SBS 仅位置").creatable().editable().boolean_input().visible_on("${$self.type == 'SBS'}"));
}
};
}
class Data_Feed : public With_Loop_Coro {
public:
BIN_Msg_Buffer msg_buffer{};
Output_Format output_format;
adminive::Managed_Value<Output_Format> output_format;
Frequency_Limit_Multi sbs_flm{};
asio::awaitable<void> loop_coro() final;
virtual void handle_in_loop() {
@@ -30,12 +44,16 @@ public:
}
virtual Psc::JSON to_json() {
Psc::JSON ret = With_Loop_Coro::to_base_json();
ret.append(Psc::JSON("output_format", output_format.to_base_json()));
ret.append(Psc::JSON("output_format", output_format.snapshot().to_base_json()));
return ret;
}
virtual void from_json(const Psc::JSON* that_json, bool not_exist_use_default_value) {
With_Loop_Coro::from_base_json(that_json, not_exist_use_default_value);
output_format.from_base_json(that_json->get("output_format"), not_exist_use_default_value);
auto value = output_format.snapshot();
value.from_base_json(that_json->get("output_format"), not_exist_use_default_value);
output_format.write([&](auto& target) {
target = value;
});
}
virtual Psc::JSON get_custom_state_json() {
return Psc::JSON::object();
@@ -46,7 +64,7 @@ public:
return ret;
}
std::string to_string() {
return "Data_Feed{" + VAR_STR_3(key, enable, type) + "}";
return "Data_Feed{" + key + "," + (enabled() ? "true" : "false") + "," + type + "}";
}
bool registered();
protected:
@@ -60,16 +78,30 @@ public:
size_t connect_system_buffer_size = 409600;
PSC_USE_JSON
};
class Data_Feed_TCP_Server : public Data_Feed, public Data_Feed_TCP_Server_Data {
namespace adminive {
template <>
struct Type_Descriptor<Data_Feed_TCP_Server_Data> {
static auto get() {
using T = Data_Feed_TCP_Server_Data;
return object<T>("data_feed_tcp_server", "TCP 服务端配置",
ADMINIVE_FIELD_LABEL(T, port, "端口").creatable().editable().number_input(),
ADMINIVE_FIELD_LABEL(T, connect_user_buffer_size, "用户缓冲区").creatable().editable().number_input(),
ADMINIVE_FIELD_LABEL(T, connect_system_buffer_size, "系统缓冲区").creatable().editable().number_input());
}
};
}
class Data_Feed_TCP_Server : public Data_Feed {
public:
adminive::Managed_Value<Data_Feed_TCP_Server_Data> specific;
Psc::asio_socket::TCP_Server_Coro svr;
Psc::JSON get_custom_state_json() override {
auto& state = svr.state;
Psc::JSON ret = Psc::JSON::object();
Ret_J(connect_system_buffer_size) Ret_J(connect_user_buffer_size) ret.key = "fixed";
auto t = VAR_JSON_1(state);
t.append(ret);
return t;
auto value = specific.snapshot();
auto ret = value.to_base_json();
ret.key = "fixed";
auto result = VAR_JSON_1(state);
result.append(ret);
return result;
}
~Data_Feed_TCP_Server() override = default;
asio::awaitable<void> handle_in_loop_coro() override {
@@ -83,23 +115,26 @@ public:
}
void from_json(const Psc::JSON* that_json, bool not_exist_use_default_value) override {
Data_Feed::from_json(that_json, not_exist_use_default_value);
Data_Feed_TCP_Server_Data::from_base_json(that_json, not_exist_use_default_value);
auto value = specific.snapshot();
value.from_base_json(that_json, not_exist_use_default_value);
specific.write([&](auto& target) {
target = value;
});
}
Psc::JSON to_json() override {
Psc::JSON ret = Data_Feed::to_json();
ret += Data_Feed_TCP_Server_Data::to_base_json();
return ret;
return Data_Feed::to_json() += specific.snapshot().to_base_json();
}
asio::awaitable<void> _close() override {
co_await svr.close_coro();
co_return;
}
asio::awaitable<void> _open() override {
svr.set_connect_user_buffer_size(connect_user_buffer_size);
svr.set_connect_system_buffer_size(connect_system_buffer_size);
const auto value = specific.snapshot();
svr.set_connect_user_buffer_size(value.connect_user_buffer_size);
svr.set_connect_system_buffer_size(value.connect_system_buffer_size);
svr.set_tcp_no_delay(false);
svr.create();
co_await svr.listen_coro("0.0.0.0", port);
co_await svr.listen_coro("0.0.0.0", value.port);
co_return;
}
Psc::JSON get_clients_json() {
@@ -138,8 +173,20 @@ public:
std::uint16_t port{};
PSC_USE_JSON
};
class Data_Feed_TCP_Client : public Data_Feed, public Data_Feed_TCP_Client_Data {
namespace adminive {
template <>
struct Type_Descriptor<Data_Feed_TCP_Client_Data> {
static auto get() {
using T = Data_Feed_TCP_Client_Data;
return object<T>("data_feed_tcp_client", "TCP 客户端配置",
ADMINIVE_FIELD_LABEL(T, url, "IP 地址").creatable().editable().text_input(),
ADMINIVE_FIELD_LABEL(T, port, "端口").creatable().editable().number_input());
}
};
}
class Data_Feed_TCP_Client : public Data_Feed {
public:
adminive::Managed_Value<Data_Feed_TCP_Client_Data> specific;
asio::awaitable<void> handle_in_loop_coro() override {
co_await cli.tick_coro();
co_return;
@@ -148,12 +195,10 @@ public:
co_await cli.send_coro(data);
co_return;
}
std::string url;
Psc::JSON get_custom_state_json() override {
auto& state = cli.state;
return VAR_JSON_1(state);
}
std::uint16_t port{};
Psc::asio_socket::TCP_Client_Coro cli;
~Data_Feed_TCP_Client() override = default;
asio::awaitable<void> _close() override {
@@ -161,9 +206,10 @@ public:
co_return;
}
asio::awaitable<void> _open() override {
const auto value = specific.snapshot();
Psc::asio_socket::Sockaddr_In sockaddr_in;
sockaddr_in.ip = url;
sockaddr_in.port = port;
sockaddr_in.ip = value.url;
sockaddr_in.port = value.port;
cli.set_dest_address(sockaddr_in);
cli.create();
co_await cli.connect_coro();
@@ -171,10 +217,14 @@ public:
}
void from_json(const Psc::JSON* that_json, bool not_exist_use_default_value) override {
Data_Feed::from_json(that_json, not_exist_use_default_value);
Data_Feed_TCP_Client_Data::from_base_json(that_json, not_exist_use_default_value);
auto value = specific.snapshot();
value.from_base_json(that_json, not_exist_use_default_value);
specific.write([&](auto& target) {
target = value;
});
}
Psc::JSON to_json() override {
return Data_Feed::to_json() += Data_Feed_TCP_Client_Data::to_base_json();
return Data_Feed::to_json() += specific.snapshot().to_base_json();
}
};
class Data_Feed_UDP_Server_Data {
@@ -182,8 +232,19 @@ public:
std::uint16_t port{};
PSC_USE_JSON
};
class Data_Feed_UDP_Server : public Data_Feed, public Data_Feed_UDP_Server_Data {
namespace adminive {
template <>
struct Type_Descriptor<Data_Feed_UDP_Server_Data> {
static auto get() {
using T = Data_Feed_UDP_Server_Data;
return object<T>("data_feed_udp_server", "UDP 服务端配置",
ADMINIVE_FIELD_LABEL(T, port, "端口").creatable().editable().number_input());
}
};
}
class Data_Feed_UDP_Server : public Data_Feed {
public:
adminive::Managed_Value<Data_Feed_UDP_Server_Data> specific;
Psc::JSON get_custom_state_json() override {
auto& state = svr.state;
return VAR_JSON_1(state);
@@ -204,10 +265,14 @@ public:
}
void from_json(const Psc::JSON* that_json, bool not_exist_use_default_value) override {
Data_Feed::from_json(that_json, not_exist_use_default_value);
Data_Feed_UDP_Server_Data::from_base_json(that_json, not_exist_use_default_value);
auto value = specific.snapshot();
value.from_base_json(that_json, not_exist_use_default_value);
specific.write([&](auto& target) {
target = value;
});
}
Psc::JSON to_json() override {
return Data_Feed::to_json() += Data_Feed_UDP_Server_Data::to_base_json();
return Data_Feed::to_json() += specific.snapshot().to_base_json();
}
asio::awaitable<void> _close() override {
co_await svr.close_coro();
@@ -222,12 +287,20 @@ public:
std::string url{};
PSC_USE_JSON
};
class Data_Feed_UDP_Client : public Data_Feed, public Data_Feed_UDP_Client_Data {
namespace adminive {
template <>
struct Type_Descriptor<Data_Feed_UDP_Client_Data> {
static auto get() {
using T = Data_Feed_UDP_Client_Data;
return object<T>("data_feed_udp_client", "UDP 客户端配置",
ADMINIVE_FIELD_LABEL(T, url, "IP 地址").creatable().editable().text_input(),
ADMINIVE_FIELD_LABEL(T, port, "端口").creatable().editable().number_input());
}
};
}
class Data_Feed_UDP_Client : public Data_Feed {
public:
Data_Feed_UDP_Client() {
}
~Data_Feed_UDP_Client() override {
}
adminive::Managed_Value<Data_Feed_UDP_Client_Data> specific;
Psc::JSON get_custom_state_json() override {
auto& state = cli.state;
return VAR_JSON_1(state);
@@ -243,10 +316,14 @@ public:
Psc::asio_socket::UDP_Client_Coro cli;
void from_json(const Psc::JSON* that_json, bool not_exist_use_default_value) override {
Data_Feed::from_json(that_json, not_exist_use_default_value);
Data_Feed_UDP_Client_Data::from_base_json(that_json, not_exist_use_default_value);
auto value = specific.snapshot();
value.from_base_json(that_json, not_exist_use_default_value);
specific.write([&](auto& target) {
target = value;
});
}
Psc::JSON to_json() override {
return Data_Feed::to_json() += Data_Feed_UDP_Client_Data::to_base_json();
return Data_Feed::to_json() += specific.snapshot().to_base_json();
}
asio::awaitable<void> _close() override {
std::cout << "udp client target address:" << cli.dest_address.to_string() << " closed" << std::endl;
@@ -254,8 +331,9 @@ public:
co_return;
}
asio::awaitable<void> _open() override {
const auto value = specific.snapshot();
cli.create();
cli.set_dest_address(url, port);
cli.set_dest_address(value.url, value.port);
co_await cli.connect_coro();
co_return;
}
@@ -276,18 +354,33 @@ inline std::shared_ptr<Data_Feed> create_data_feed_from_type(std::string_view t)
std::cout << oss.str();
throw std::invalid_argument(oss.str());
}
ret->type = std::string(t);
return ret;
}
class Data_feed_Config_Data {
public:
Psc::Copyable_Atomic<std::uint16_t> empty_wait_milliseconds{};
Psc::Copyable_Atomic<std::uint16_t> packet_byte_size{};
std::uint16_t empty_wait_milliseconds{};
std::uint16_t packet_byte_size{};
size_t tcp_server_default_connect_user_buffer_size{};
size_t tcp_server_default_connect_system_buffer_size{};
PSC_USE_JSON
};
class Data_feed_Config : public Data_feed_Config_Data {
namespace adminive {
template <>
struct Type_Descriptor<Data_feed_Config_Data> {
static auto get() {
using T = Data_feed_Config_Data;
return object<T>("data_feed_settings", "数据馈送全局配置",
ADMINIVE_FIELD_LABEL(T, empty_wait_milliseconds, "忙等待时间（毫秒）").editable().number_input(),
ADMINIVE_FIELD_LABEL(T, packet_byte_size, "最小批大小（byte）").editable().number_input(),
ADMINIVE_FIELD_LABEL(T, tcp_server_default_connect_user_buffer_size, "TCP 默认用户缓冲区").editable().number_input(),
ADMINIVE_FIELD_LABEL(T, tcp_server_default_connect_system_buffer_size, "TCP 默认系统缓冲区").editable().number_input());
}
};
}
class Data_feed_Config {
public:
adminive::Managed_Value<Data_feed_Config_Data> settings;
Psc::String_Pool pool_ = Psc::String_Pool(1600, 8 * 1024); // 8192 * 1600 12.5 MB
Ordered_Map<std::string, std::shared_ptr<Data_Feed>> map;
Data_feed_Config() = default;
@@ -299,17 +392,21 @@ public:
t->from_json(&it, not_exist_use_default_value);
map.push_back(t);
}
Data_feed_Config_Data::from_base_json(that_json, not_exist_use_default_value);
auto value = settings.snapshot();
value.from_base_json(that_json, not_exist_use_default_value);
settings.write([&](auto& target) {
target = value;
});
}
Psc::JSON list() {
Psc::JSON list() const {
auto connect_json = Psc::JSON::array();
for (auto& it : map.list()) {
connect_json.children.emplace_back(it->to_json());
}
return connect_json;
}
Psc::JSON to_json() {
Psc::JSON ret = Data_feed_Config_Data::to_base_json();
Psc::JSON to_json() const {
Psc::JSON ret = settings.snapshot().to_base_json();
ret.append({"list", list()});
return ret;
}