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-10 14:51:03 +08:00
parent c268ae23c4
commit 18e92d3631
5 changed files with 153 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
module/Local_Server/server/Adminive_Config.cpp
+143 -48
View File
@@ -333,6 +333,7 @@ Data_Topology_View_Config read_topology_view(Json input, Global* global) {
};
Data_Topology_View_Config result;
result.initialized = true;
result.layout_revision = input.value("layout_revision", 0U);
result.zoom = finite_number("zoom");
result.pan_x = finite_number("pan_x");
result.pan_y = finite_number("pan_y");
@@ -637,6 +638,25 @@ Json data_topology_view(Global* global) {
result["protocol_version"] = 1;
return result;
}
Json data_topology_connection_status(Global* global, std::uint64_t id) {
const auto feed = item_by_id(global->mode_acs.data_feed_config.map.list(), id);
const auto source = global->mode_acs.data_source_config.map.get(feed->source_key);
Json result{{"连接", feed->source_key + " → " + feed->key}, {"数据馈送状态", Psc::to_string(feed->state)}, {"数据馈送启用", feed->enabled()}};
if(!source) {
result["连接状态"] = "数据源不存在";
result["已在数据源注册"] = false;
return result;
}
result["数据源状态"] = Psc::to_string(source.value()->state);
result["数据源启用"] = source.value()->enabled();
const auto all_status = adminive::parse_json<Json>(source.value()->get_all_connect_feed_status().to_json_string());
const auto status = all_status.find(feed->key);
result["已在数据源注册"] = status != all_status.end();
result["连接状态"] = status != all_status.end() && source.value()->enabled() && feed->enabled() ? "运行中" : "未运行";
if(status != all_status.end())
result["缓存统计"] = *status;
return result;
}
Json data_topology(Global* global) {
const auto sources = global->mode_acs.data_source_config.map.list();
const auto feeds = global->mode_acs.data_feed_config.map.list();
@@ -647,21 +667,65 @@ Json data_topology(Global* global) {
const auto& source = sources[index];
const auto id = "data_source:" + source->key;
source_by_key.emplace(source->key, source);
auto config = adminive::to_frontend_json<Json>(source_row(source));
config["id"] = index + 1;
Json details = Json::array();
details.push_back(Json{{"label", "运行状态"}, {"value", Psc::to_string(source->state)}});
if(const auto value = dynamic_cast<TCP_Client_Data_Source*>(source.get())) {
value->specific.read([&](const auto& config) { details.push_back(Json{{"label", "目标"}, {"value", config.ip + ":" + std::to_string(config.port)}}); });
} else if(const auto value = dynamic_cast<Serial_Data_Source*>(source.get())) {
value->specific.read([&](const auto& config) { details.push_back(Json{{"label", "串口"}, {"value", config.port_name}}); });
} else if(const auto value = dynamic_cast<File_Data_Source*>(source.get())) {
value->specific.read([&](const auto& config) { details.push_back(Json{{"label", "文件"}, {"value", config.file_path}}); });
} else if(const auto value = dynamic_cast<Dll_Data_Source*>(source.get())) {
value->specific.read([&](const auto& config) { details.push_back(Json{{"label", "动态库"}, {"value", config.library_path}}); });
} else if(const auto value = dynamic_cast<Shared_Memory_Data_Source*>(source.get())) {
value->specific.read([&](const auto& config) { details.push_back(Json{{"label", "共享内存"}, {"value", config.shared_memory_name}}); });
}
nodes.push_back(Json{{"id", id}, {"kind", "data_source"}, {"entity_id", index + 1},
{"key", source->key}, {"label", source->key}, {"type", source->type},
{"enabled", source->enabled()}, {"status_api", data_topology_path + "/status/data_source/" + std::to_string(index + 1)}});
{"key", source->key}, {"label", source->key}, {"type", source->type},
{"enabled", source->enabled()}, {"config", std::move(config)}, {"details", std::move(details)}, {"status_api", data_topology_path + "/status/data_source/" + std::to_string(index + 1)}});
}
for(std::size_t index = 0; index < feeds.size(); ++index) {
const auto& feed = feeds[index];
const auto id = "data_feed:" + feed->key;
auto config = adminive::to_frontend_json<Json>(feed_row(feed));
config["id"] = index + 1;
Json details = Json::array({Json{{"label", "运行状态"}, {"value", Psc::to_string(feed->state)}}});
feed->output_format.read([&](const auto& value) { details.push_back(Json{{"label", "数据格式"}, {"value", std::string(magic_enum::enum_name(value.type))}}); });
if(const auto value = dynamic_cast<Data_Feed_TCP_Server*>(feed.get())) {
value->specific.read([&](const auto& config) { details.push_back(Json{{"label", "监听端口"}, {"value", config.port}}); });
} else if(const auto value = dynamic_cast<Data_Feed_UDP_Server*>(feed.get())) {
value->specific.read([&](const auto& config) { details.push_back(Json{{"label", "监听端口"}, {"value", config.port}}); });
} else if(const auto value = dynamic_cast<Data_Feed_TCP_Client*>(feed.get())) {
value->specific.read([&](const auto& config) { details.push_back(Json{{"label", "目标"}, {"value", config.url + ":" + std::to_string(config.port)}}); });
} else if(const auto value = dynamic_cast<Data_Feed_UDP_Client*>(feed.get())) {
value->specific.read([&](const auto& config) { details.push_back(Json{{"label", "目标"}, {"value", config.url + ":" + std::to_string(config.port)}}); });
}
nodes.push_back(Json{{"id", id}, {"kind", "data_feed"}, {"entity_id", index + 1},
{"key", feed->key}, {"label", feed->key}, {"type", feed->type},
{"enabled", feed->enabled()}, {"status_api", data_topology_path + "/status/data_feed/" + std::to_string(index + 1)}});
{"key", feed->key}, {"label", feed->key}, {"type", feed->type},
{"enabled", feed->enabled()}, {"config", std::move(config)}, {"details", std::move(details)}, {"status_api", data_topology_path + "/status/data_feed/" + std::to_string(index + 1)}});
const auto source = source_by_key.find(feed->source_key);
if(source != source_by_key.end()) {
edges.push_back(Json{{"id", "source-feed:" + feed->key},
edges.push_back(Json{{"id", "source-feed:" + feed->key}, {"kind", "configuration"},
{"source", "data_source:" + feed->source_key}, {"target", id},
{"active", source->second->enabled() && feed->enabled()}});
{"active", source->second->enabled() && feed->enabled()},
{"status_api", data_topology_path + "/status/connection/" + std::to_string(index + 1)}});
}
if(const auto value = dynamic_cast<Data_Feed_TCP_Server*>(feed.get())) {
for(const auto& client : value->svr.get_all_clients()) {
const auto client_id = "client:tcp:" + feed->key + ":" + std::to_string(client->info.fd);
const auto endpoint = client->info.sockaddr.to_string();
nodes.push_back(Json{{"id", client_id}, {"kind", "client"}, {"entity_id", 0}, {"key", endpoint}, {"label", endpoint}, {"type", "TCP 客户端"}, {"enabled", true}, {"details", Json::array({Json{{"label", "文件描述符"}, {"value", client->info.fd}}, Json{{"label", "发送缓冲区"}, {"value", client->send_buffer.state_str()}}})}});
edges.push_back(Json{{"id", "feed-client:" + feed->key + ":" + std::to_string(client->info.fd)}, {"kind", "runtime"}, {"source", id}, {"target", client_id}, {"active", true}, {"details", Json::array({Json{{"label", "连接状态"}, {"value", "已连接"}}, Json{{"label", "远端"}, {"value", endpoint}}, Json{{"label", "发送缓冲区"}, {"value", client->send_buffer.state_str()}}})}});
}
} else if(const auto value = dynamic_cast<Data_Feed_UDP_Server*>(feed.get())) {
for(const auto& client : value->svr.clients) {
const auto endpoint = client.to_string();
const auto client_id = "client:udp:" + feed->key + ":" + endpoint;
nodes.push_back(Json{{"id", client_id}, {"kind", "client"}, {"entity_id", 0}, {"key", endpoint}, {"label", endpoint}, {"type", "UDP 客户端"}, {"enabled", true}, {"details", Json::array()}});
edges.push_back(Json{{"id", "feed-client:" + feed->key + ":" + endpoint}, {"kind", "runtime"}, {"source", id}, {"target", client_id}, {"active", true}, {"details", Json::array({Json{{"label", "连接状态"}, {"value", "已连接"}}, Json{{"label", "远端"}, {"value", endpoint}}})}});
}
}
}
return Json{{"protocol", "ecap.data-topology"}, {"protocol_version", 1},
@@ -724,27 +788,51 @@ Json feed_amis_schema(Global* global) {
set_select_options(schema, "source_key", options);
return schema;
}
Json topology_editor(std::string kind, std::string title, std::string base_api, Json create_data, Json collection_schema) {
const auto& crud = collection_schema["body"][0];
Json create_schema;
for(const auto& item : crud["headerToolbar"]) {
if(item.is_object() && item.value("actionType", std::string{}) == "dialog") {
create_schema = item["dialog"]["body"];
break;
}
}
const auto& columns = crud["columns"];
Json edit_schema = columns.back()["buttons"][0]["dialog"]["body"];
create_schema.erase("reload");
edit_schema.erase("reload");
edit_schema["initApi"] = Json{{"method", "get"}, {"url", base_api + "/${id}"}};
return Json{{"kind", std::move(kind)}, {"title", std::move(title)}, {"create_data", std::move(create_data)}, {"create_schema", std::move(create_schema)}, {"edit_schema", std::move(edit_schema)}};
}
Json data_topology_schema(Global* global) {
auto source_create_data = adminive::to_frontend_json<Json>(Data_Source_Row{});
source_create_data.erase("state");
auto feed_create_data = adminive::to_frontend_json<Json>(Data_Feed_Row{});
feed_create_data.erase("format");
feed_create_data.erase("state");
return Json{
{"title", "数据源与数据馈送拓扑"},
{"data_api", data_topology_path},
{"view_api", data_topology_path + "/view"},
{"graph", Json{
{"height", 620},
{"layout", Json{{"type", "dagre"}, {"rankdir", "LR"}, {"align", "UL"}, {"nodesep", 34}, {"ranksep", 100}}},
{"layout", Json{{"type", "dagre"}, {"rankdir", "TB"}, {"align", "UL"}, {"nodesep", 42}, {"ranksep", 92}}},
{"layout_revision", 2},
{"poll_interval_ms", 2000},
{"behaviors", Json::array({"drag-canvas", "zoom-canvas", "drag-element"})},
{"active_edge_color", "#52c41a"},
{"inactive_edge_color", "#bfbfbf"}
}},
{"node_kinds", Json::array({
Json{{"kind", "data_source"}, {"title", "数据源"}, {"fill", "#e6f4ff"}, {"stroke", "#1677ff"}},
Json{{"kind", "data_feed"}, {"title", "数据馈送"}, {"fill", "#f6ffed"}, {"stroke", "#52c41a"}}
Json{{"kind", "data_feed"}, {"title", "数据馈送"}, {"fill", "#f6ffed"}, {"stroke", "#52c41a"}},
Json{{"kind", "client"}, {"title", "已连接客户端"}, {"fill", "#fff7e6"}, {"stroke", "#fa8c16"}}
})},
{"status", Json{{"title", "运行状态"}, {"poll_interval_ms", 1000},
{"collapsed_hint", "展开后才向后台轮询，收起立即停止"}}},
{"editors", Json::array({
Json{{"kind", "data_source"}, {"title", "数据源管理"}, {"button_label", "管理数据源"}, {"schema", source_amis_schema()}},
Json{{"kind", "data_feed"}, {"title", "数据馈送管理"}, {"button_label", "管理数据馈送"}, {"schema", feed_amis_schema(global)}}
topology_editor("data_source", "数据源", "/api/adminive/config/data_sources", std::move(source_create_data), source_amis_schema()),
topology_editor("data_feed", "数据馈送", "/api/adminive/config/data_feeds", std::move(feed_create_data), feed_amis_schema(global))
})}
};
}
@@ -920,18 +1008,21 @@ private:
}, put);
app.registerHandler(data_topology_path + "/status/data_source/{1:id}", [this](const drogon::HttpRequestPtr&, std::function<void(const drogon::HttpResponsePtr&)>&& callback, std::uint64_t id) {
complete_business_request(callback, [this, id] {
std::lock_guard lock(topology_mutex_);
const auto source = item_by_id(global_->mode_acs.data_source_config.map.list(), id);
return adminive::make_http_success<Json>(adminive::parse_json<Json>(source->get_state().to_json_string()));
});
}, get);
app.registerHandler(data_topology_path + "/status/data_feed/{1:id}", [this](const drogon::HttpRequestPtr&, std::function<void(const drogon::HttpResponsePtr&)>&& callback, std::uint64_t id) {
complete_business_request(callback, [this, id] {
std::lock_guard lock(topology_mutex_);
const auto feed = item_by_id(global_->mode_acs.data_feed_config.map.list(), id);
return adminive::make_http_success<Json>(adminive::parse_json<Json>(feed->get_state_json().to_json_string()));
});
}, get);
app.registerHandler(data_topology_path + "/status/connection/{1:id}", [this](const drogon::HttpRequestPtr&, std::function<void(const drogon::HttpResponsePtr&)>&& callback, std::uint64_t id) {
complete_business_request(callback, [this, id] {
return adminive::make_http_success<Json>(data_topology_connection_status(global_, id));
});
}, get);
}
void bind_backend_config(drogon::HttpAppFramework& app) {
const auto get = adminive::make_drogon_constraints(drogon::Get, {});
@@ -972,24 +1063,26 @@ private:
return adminive::make_http_success<Json>(adminive::to_frontend_json<Json>(source_row(source)), "created");
});
}, post);
const auto update = [this](const drogon::HttpRequestPtr& request, std::function<void(const drogon::HttpResponsePtr&)>&& callback, std::uint64_t id) {
complete_business_request(callback, [this, id, body = std::string(request->getBody())] {
std::lock_guard lock(topology_mutex_);
const auto list = global_->mode_acs.data_source_config.map.list();
auto source = item_by_id(list, id);
auto input = read_request_body(body);
auto row = source_row(source);
const auto result = adminive::apply_frontend_patch<Json>(row, input);
if(!result.success)
return adminive::make_http_error<Json>(422, result);
apply_source_row(source, row);
Data_feed_Config::set_connections_need_refresh();
save_data_sources(global_);
return adminive::make_http_success<Json>(adminive::to_frontend_json<Json>(source_row(source)), result.message);
});
const auto make_update = [this] {
return [this](const drogon::HttpRequestPtr& request, std::function<void(const drogon::HttpResponsePtr&)>&& callback, std::uint64_t id) {
complete_business_request(callback, [this, id, body = std::string(request->getBody())] {
std::lock_guard lock(topology_mutex_);
const auto list = global_->mode_acs.data_source_config.map.list();
auto source = item_by_id(list, id);
auto input = read_request_body(body);
auto row = source_row(source);
const auto result = adminive::apply_frontend_patch<Json>(row, input);
if(!result.success)
return adminive::make_http_error<Json>(422, result);
apply_source_row(source, row);
Data_feed_Config::set_connections_need_refresh();
save_data_sources(global_);
return adminive::make_http_success<Json>(adminive::to_frontend_json<Json>(source_row(source)), result.message);
});
};
};
app.registerHandler(data_sources_path + "/{1:id}", update, put);
app.registerHandler(data_sources_path + "/{1:id}", update, patch);
app.registerHandler(data_sources_path + "/{1:id}", make_update(), put);
app.registerHandler(data_sources_path + "/{1:id}", make_update(), patch);
app.registerHandler(data_sources_path + "/{1:id}", [this](const drogon::HttpRequestPtr&, std::function<void(const drogon::HttpResponsePtr&)>&& callback, std::uint64_t id) {
complete_business_request(callback, [this, id] {
std::lock_guard lock(topology_mutex_);
@@ -1047,26 +1140,28 @@ private:
return adminive::make_http_success<Json>(adminive::to_frontend_json<Json>(feed_row(feed)), "created");
});
}, post);
const auto update = [this](const drogon::HttpRequestPtr& request, std::function<void(const drogon::HttpResponsePtr&)>&& callback, std::uint64_t id) {
complete_business_request(callback, [this, id, body = std::string(request->getBody())] {
std::lock_guard lock(topology_mutex_);
const auto list = global_->mode_acs.data_feed_config.map.list();
auto feed = item_by_id(list, id);
auto input = read_request_body(body);
auto row = feed_row(feed);
const auto result = adminive::apply_frontend_patch<Json>(row, input);
if(!result.success)
return adminive::make_http_error<Json>(422, result);
if(!row.source_key.empty() && !global_->mode_acs.data_source_config.map.get(row.source_key))
throw std::invalid_argument("连接的数据源不存在: " + row.source_key);
apply_feed_row(feed, row);
Data_feed_Config::set_connections_need_refresh();
save_data_feeds(global_);
return adminive::make_http_success<Json>(adminive::to_frontend_json<Json>(feed_row(feed)), result.message);
});
const auto make_update = [this] {
return [this](const drogon::HttpRequestPtr& request, std::function<void(const drogon::HttpResponsePtr&)>&& callback, std::uint64_t id) {
complete_business_request(callback, [this, id, body = std::string(request->getBody())] {
std::lock_guard lock(topology_mutex_);
const auto list = global_->mode_acs.data_feed_config.map.list();
auto feed = item_by_id(list, id);
auto input = read_request_body(body);
auto row = feed_row(feed);
const auto result = adminive::apply_frontend_patch<Json>(row, input);
if(!result.success)
return adminive::make_http_error<Json>(422, result);
if(!row.source_key.empty() && !global_->mode_acs.data_source_config.map.get(row.source_key))
throw std::invalid_argument("连接的数据源不存在: " + row.source_key);
apply_feed_row(feed, row);
Data_feed_Config::set_connections_need_refresh();
save_data_feeds(global_);
return adminive::make_http_success<Json>(adminive::to_frontend_json<Json>(feed_row(feed)), result.message);
});
};
};
app.registerHandler(data_feeds_path + "/{1:id}", update, put);
app.registerHandler(data_feeds_path + "/{1:id}", update, patch);
app.registerHandler(data_feeds_path + "/{1:id}", make_update(), put);
app.registerHandler(data_feeds_path + "/{1:id}", make_update(), patch);
app.registerHandler(data_feeds_path + "/{1:id}", [this](const drogon::HttpRequestPtr&, std::function<void(const drogon::HttpResponsePtr&)>&& callback, std::uint64_t id) {
complete_business_request(callback, [this, id] {
std::lock_guard lock(topology_mutex_);