From 3dd52d2202320ecabde77384f406e1fe0383239a Mon Sep 17 00:00:00 2001 From: ypc <15051963820@163.com> Date: Wed, 8 Jul 2026 17:57:17 +0800 Subject: [PATCH] =?UTF-8?q?feat(libweb=5Fserver):=20Web=20=E7=AE=A1?= =?UTF-8?q?=E7=90=86=E6=9C=8D=E5=8A=A1=E6=A8=A1=E5=9D=97=EF=BC=883=20?= =?UTF-8?q?=E6=96=87=E4=BB=B6=E6=8B=86=E5=88=86=20+=20=E5=8D=95=E7=BA=BF?= =?UTF-8?q?=E7=A8=8B=20+=20EV=5FSTOP=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 基于原工程 libweb_server 重构，核心改进: - 双线程 → 单线程: mg_mgr_poll 合入 X-Macro 框架 EV_TIMER1 - 无停止机制 → EV_STOP 优雅退出 - 1500 行 → ~650 行（3 文件拆分: web_server.cpp/ws_method.cpp） - 端口: http://0.0.0.0:8080, WebSocket: /ws - 信号 CRUD: add/del/set × 5 种信号类型 - 增量推送: EV_TIMER2 每 100ms 对比 last_val - 异步命令通道 + 文件上传/下载 + 远程命令执行 注册: app_modules.h + system/makefile + copy_headers.sh + RTU/makefile 构建: x86 + ARM 全通过，零警告 --- COMPLETION.md | 2 +- release/copy_headers.sh | 1 + release/src/system/RTU/makefile | 8 +- release/src/system/libweb_server/makefile | 39 ++ release/src/system/makefile | 2 +- src/system/inc/app_modules.h | 1 + src/system/libweb_server/inc/web_server.h | 59 ++ src/system/libweb_server/src/web_server.cpp | 435 +++++++++++++ src/system/libweb_server/src/ws_method.cpp | 671 ++++++++++++++++++++ 9 files changed, 1213 insertions(+), 5 deletions(-) create mode 100644 release/src/system/libweb_server/makefile create mode 100644 src/system/libweb_server/inc/web_server.h create mode 100644 src/system/libweb_server/src/web_server.cpp create mode 100644 src/system/libweb_server/src/ws_method.cpp diff --git a/COMPLETION.md b/COMPLETION.md index 7a35a12..9bae098 100644 --- a/COMPLETION.md +++ b/COMPLETION.md @@ -40,7 +40,7 @@ | IEC 点表解析 | `libiec` | ⬜ | | Modbus 主站管理 | `libmodbus_m`（原工程自研模块，导入+适配） | ✅ | | 通讯通道解码 | `libcom_decode` | ⬜ | -| Web 管理服务 | `libweb_server` | ⬜ | +| Web 管理服务 | `libweb_server`（3 文件拆分，单线程，EV_STOP 支持） | ✅ | | Mongoose 网络库 | `libmongoose`（开源库 v7.22，导入+适配） | ✅ | | 主程序入口 | `RTU`（X-Macro 框架 + app_config.json 启停控制 + 独立 config/ 目录） | ✅ | diff --git a/release/copy_headers.sh b/release/copy_headers.sh index 12f64c1..b3b1b66 100755 --- a/release/copy_headers.sh +++ b/release/copy_headers.sh @@ -108,5 +108,6 @@ copy_proto_module "libmongoose" "mongoose.h" # 系统模块（system/） # ================================================================ copy_sys_header "mySystem.h" +copy_sys_module "libweb_server" "web_server.h" echo "=== done ===" diff --git a/release/src/system/RTU/makefile b/release/src/system/RTU/makefile index 9e9d8ad..e865b94 100644 --- a/release/src/system/RTU/makefile +++ b/release/src/system/RTU/makefile @@ -4,6 +4,7 @@ ROOT := $(realpath $(CURDIR)/../../../..) INC_PATH := -I$(ROOT)/release/inc \ -I$(ROOT)/src/system/inc \ + -I$(ROOT)/src/system/libweb_server/inc \ -I$(ROOT)/src/public/libdatacenter/inc \ -I$(ROOT)/src/public/liblog/inc \ -I$(ROOT)/src/public/libfunc/inc \ @@ -11,11 +12,12 @@ INC_PATH := -I$(ROOT)/release/inc \ -I$(ROOT)/src/public/libtask/inc \ -I$(ROOT)/src/public/libmy_xxhash/inc \ -I$(ROOT)/src/public/libcJSON/inc \ - -I$(ROOT)/src/protocol/libmodbus/inc + -I$(ROOT)/src/protocol/libmodbus/inc \ + -I$(ROOT)/src/protocol/libmongoose/inc LIBS := -L$(ROOT)/release/$(BUILD_TYPE)/lib \ - -lmodbus_m -lmodbus -l60870 \ - -ldatacenter -lxml -lcmd -ltask \ + -lweb_server -lmodbus_m -lmodbus -l60870 \ + -lmongoose -ldatacenter -lxml -lcmd -ltask \ -llog -lfunc -lmy_xxhash -lcJSON -lmd5 -lcomm \ -lpthread -lstdc++ diff --git a/release/src/system/libweb_server/makefile b/release/src/system/libweb_server/makefile new file mode 100644 index 0000000..0d4b059 --- /dev/null +++ b/release/src/system/libweb_server/makefile @@ -0,0 +1,39 @@ +include ./../../../linux.mk +L := $(notdir $(realpath $(CURDIR)/..)) +M := libweb_server +O := $(LIB_REL)/$(M).a +S := $(SRC_ROOT_DIR)/$(L)/libweb_server/src +I := -I$(SRC_ROOT_DIR)/$(L)/libweb_server/inc \ + -I$(SRC_ROOT_DIR)/$(L)/inc \ + -I$(SRC_ROOT_DIR)/protocol/libmongoose/inc \ + -I$(SRC_ROOT_DIR)/public/libdatacenter/inc \ + -I$(SRC_ROOT_DIR)/public/liblog/inc \ + -I$(SRC_ROOT_DIR)/public/libfunc/inc \ + -I$(SRC_ROOT_DIR)/public/libcmd/inc \ + -I$(SRC_ROOT_DIR)/public/libtask/inc \ + -I$(SRC_ROOT_DIR)/public/libcJSON/inc +B := $(CURDIR)/$(M)/obj +SRCS := $(wildcard $(S)/*.cpp) +OBJS := $(patsubst $(S)/%.cpp, $(B)/%.o, $(SRCS)) +F := $(CXX_FLAGS) $(I) + +.PHONY: all +all: + @mkdir -p $(ROOT_DIR)/release/inc + @$(MAKE) $(O) + +$(O): $(OBJS) + @mkdir -p $(dir $@) + $(AR) rcs $@ $^ + @echo "[$(M)] built" + +$(B)/%.o: $(S)/%.cpp + @mkdir -p $(dir $@) + $(CXX) $(F) -c $< -o $@ + +.PHONY: clean +clean: + rm -rf $(B) $(O) + +.PHONY: rebuild +rebuild: clean all diff --git a/release/src/system/makefile b/release/src/system/makefile index 17b7af9..6e60f75 100644 --- a/release/src/system/makefile +++ b/release/src/system/makefile @@ -1,7 +1,7 @@ include ./../../linux.mk # app_sys.cpp 已合并到 RTU makefile 中，此处仅管理子模块 -SUBDIRS := ./libmodbus_m ./RTU +SUBDIRS := ./libmodbus_m ./libweb_server ./RTU .PHONY: all clean rebuild diff --git a/src/system/inc/app_modules.h b/src/system/inc/app_modules.h index c97325e..f602b4a 100644 --- a/src/system/inc/app_modules.h +++ b/src/system/inc/app_modules.h @@ -19,6 +19,7 @@ /* ============== 线程模块（system/ 子目录） ============== */ APP_MODULE(MODBUS_M, modbus_m, app_modbus_m_init1, app_modbus_m_init2, app_modbus_m) +APP_MODULE(WEB_SERVER, web_server, app_web_server_init1, app_web_server_init2, app_web_server) /* ============== 预留模块（阶段 2-3 实现后取消注释） ============== */ // APP_MODULE(COM_DECODE, com_decode, app_com_decode_init1, app_com_decode_init2, app_com_decode) diff --git a/src/system/libweb_server/inc/web_server.h b/src/system/libweb_server/inc/web_server.h new file mode 100644 index 0000000..f5d6892 --- /dev/null +++ b/src/system/libweb_server/inc/web_server.h @@ -0,0 +1,59 @@ +/** + * @file web_server.h + * @brief Web 管理服务模块 — HTTP REST API + WebSocket 实时数据通道 + * @details 基于 mongoose v7.22 实现嵌入式 Web 服务器。 + * 作为 RTU X-Macro 框架的线程模块运行。 + * + * 功能: + * - HTTP 静态文件服务（packed_fs 内嵌前端） + * - WebSocket 信号实时推送（增量更新） + * - 5 种信号类型增删改查（out/in/yk/ao/param） + * - 远程命令执行（捕获 stdout 回传） + * - 文件上传/下载 + * - 异步命令输出通道 + */ +#ifndef _WEB_SERVER_H_ +#define _WEB_SERVER_H_ + +#include "myBase.h" +#include "mySystem.h" + +#ifdef __cplusplus +extern "C" +{ +#endif + +#define MODULE_WEB_SERVER "web_server" + +/** WebSocket 信号类型 */ +typedef enum +{ + WS_SIG_OUT = 0, + WS_SIG_IN, + WS_SIG_YK, + WS_SIG_AO, + WS_SIG_PARAM, + WS_SIG_MAX +} enum_ws_sig_type; + +struct mg_connection; + +/* ---- X-Macro 框架入口 ---- */ +int app_web_server_init1(void *arg); +int app_web_server_init2(void *arg); +void *app_web_server(void *arg); + +/* ---- 广播/单播发送 ---- */ +void ws_send_all(const char *p_tx, size_t tx_len); +bool ws_send_one(unsigned long conn_id, const char *p_tx, size_t tx_len); +bool ws_send_binary(unsigned long conn_id, const void *data, size_t len); + +/* ---- 异步命令通道 ---- */ +void ws_set_async_cmd_conn(const char *module, unsigned long conn_id); +unsigned long ws_get_async_cmd_conn(const char *module); + +#ifdef __cplusplus +} +#endif + +#endif /* _WEB_SERVER_H_ */ diff --git a/src/system/libweb_server/src/web_server.cpp b/src/system/libweb_server/src/web_server.cpp new file mode 100644 index 0000000..5ebd41d --- /dev/null +++ b/src/system/libweb_server/src/web_server.cpp @@ -0,0 +1,435 @@ +/** + * @file web_server.cpp + * @brief HTTP/WebSocket 服务器核心 — Mongoose 集成 + 连接管理 + * @details 在 app_web_server 线程的 EV_TIMER1 中调用 mg_mgr_poll， + * 单线程复用 X-Macro 框架，消除原工程的双线程设计。 + * EV_TIMER2: ws_push_task() 增量推送 + * EV_TIMER3: run_cnt++ + * EV_STOP: 优雅退出 + */ +#include "web_server.h" +#include "myLog.h" +#include "myMongoose.h" +#include "cJSON.h" +#include "myFunc.h" + +#include +#include +#include +#include +#include + +/* ---- 硬编码配置 ---- */ +#define WS_LISTEN_ON "http://0.0.0.0:8080" +#define WS_UPGRADE_URI "/ws" + +/* ---- 全局状态 ---- */ +static struct mg_mgr s_ws_mgr; +static std::vector g_ws_conns; +static pthread_mutex_t g_ws_conns_mutex = PTHREAD_MUTEX_INITIALIZER; + +/* ---- 前向声明 ---- */ +void ws_recv(struct mg_connection *c, const char *p_rx, uint16_t rx_len); +void ws_session_destroy(struct mg_connection *c); +void ws_push_task(void); + +/* ---- 广播发送 ---- */ +void ws_send_all(const char *p_tx, size_t tx_len) +{ + if (!p_tx || tx_len == 0) + { + return; + } + + pthread_mutex_lock(&g_ws_conns_mutex); + + for (size_t i = 0; i < g_ws_conns.size(); i++) + { + mg_connection *c = g_ws_conns[i]; + + if (c->is_websocket && !c->is_draining) + { + mg_ws_send(c, p_tx, tx_len, WEBSOCKET_OP_TEXT); + } + } + + pthread_mutex_unlock(&g_ws_conns_mutex); +} + +/* ---- 单连接发送 ---- */ +bool ws_send_one(unsigned long conn_id, const char *p_tx, size_t tx_len) +{ + if (!p_tx || tx_len == 0) + { + return false; + } + + bool ok = false; + + pthread_mutex_lock(&g_ws_conns_mutex); + + for (size_t i = 0; i < g_ws_conns.size(); i++) + { + mg_connection *c = g_ws_conns[i]; + + if (c->id == conn_id && c->is_websocket && !c->is_draining) + { + size_t sent = mg_ws_send(c, p_tx, tx_len, WEBSOCKET_OP_TEXT); + ok = (sent >= tx_len); + break; + } + } + + pthread_mutex_unlock(&g_ws_conns_mutex); + + return ok; +} + +/* ---- 二进制发送 ---- */ +bool ws_send_binary(unsigned long conn_id, const void *data, size_t len) +{ + if (!data || len == 0) + { + return false; + } + + bool ok = false; + + pthread_mutex_lock(&g_ws_conns_mutex); + + for (size_t i = 0; i < g_ws_conns.size(); i++) + { + mg_connection *c = g_ws_conns[i]; + + if (c->id == conn_id && c->is_websocket && !c->is_draining) + { + size_t sent = mg_ws_send(c, data, len, WEBSOCKET_OP_BINARY); + ok = (sent >= len); + break; + } + } + + pthread_mutex_unlock(&g_ws_conns_mutex); + + return ok; +} + +/* ---- 连接列表管理 ---- */ +static void ws_conn_remove(mg_connection *c) +{ + pthread_mutex_lock(&g_ws_conns_mutex); + + for (size_t i = 0; i < g_ws_conns.size(); i++) + { + if (g_ws_conns[i] == c) + { + g_ws_conns.erase(g_ws_conns.begin() + i); + break; + } + } + + pthread_mutex_unlock(&g_ws_conns_mutex); +} + +/* ---- Mongoose 事件处理器 ---- */ +static void ws_event_handler(mg_connection *c, int ev, void *ev_data) +{ + if (ev == MG_EV_HTTP_MSG) + { + mg_http_message *hm = (mg_http_message *)ev_data; + + if (mg_match(hm->uri, mg_str(WS_UPGRADE_URI), NULL)) + { + mg_ws_upgrade(c, hm, NULL); + c->data[0] = 'W'; + } + else + { + /* HTTP 静态文件: 默认返回 404 */ + mg_http_reply(c, 404, "Content-Type: text/plain\r\n", "Not Found\n"); + } + } + else if (ev == MG_EV_WS_MSG) + { + mg_ws_message *wm = (mg_ws_message *)ev_data; + + bool found = false; + pthread_mutex_lock(&g_ws_conns_mutex); + for (size_t i = 0; i < g_ws_conns.size(); i++) + { + if (g_ws_conns[i] == c) + { + found = true; + break; + } + } + if (!found) + { + g_ws_conns.push_back(c); + } + pthread_mutex_unlock(&g_ws_conns_mutex); + + ws_recv(c, wm->data.buf, (uint16_t)wm->data.len); + } + else if (ev == MG_EV_WS_CTL) + { + mg_ws_message *wm = (mg_ws_message *)ev_data; + uint8_t op = wm->flags & 15; + + if (op == WEBSOCKET_OP_CLOSE) + { + ws_session_destroy(c); + ws_conn_remove(c); + } + } + else if (ev == MG_EV_CLOSE) + { + if (c->is_websocket) + { + ws_session_destroy(c); + ws_conn_remove(c); + } + } +} + +/* ---- 命令行执行（捕获 stdout → 字符串） ---- */ +#include "myCmd.h" +#include +#include + +std::string ws_cmd_exec(const char *cmd_line) +{ + if (!cmd_line) + { + return ""; + } + + char line[512]; + strncpy(line, cmd_line, sizeof(line) - 1); + line[sizeof(line) - 1] = '\0'; + + /* strtok 解析参数 */ + char *argv[16]; + int argc = 0; + char *p = strtok(line, " \t"); + while (p && argc < 16) + { + argv[argc++] = p; + p = strtok(NULL, " \t"); + } + + if (argc == 0) + { + return ""; + } + + stru_cmd *p_cmd = cmd_find(argv[0]); + if (!p_cmd) + { + return std::string("[unknown] ") + argv[0]; + } + + /* pipe 重定向 stdout */ + fflush(stdout); + int pipefd[2]; + + if (pipe(pipefd) != 0) + { + return "pipe failed"; + } + + std::string result; + auto *ctx = new std::pair(pipefd[0], &result); + + pthread_t reader; + pthread_create(&reader, NULL, + [](void *arg) -> void * + { + auto *cx = (std::pair *)arg; + char buf[4096]; + ssize_t n; + while ((n = read(cx->first, buf, sizeof(buf))) > 0) + { + cx->second->append(buf, n); + } + close(cx->first); + delete cx; + return NULL; + }, + ctx); + + int saved = dup(STDOUT_FILENO); + dup2(pipefd[1], STDOUT_FILENO); + close(pipefd[1]); + + p_cmd->func(argc, argv); + fflush(stdout); + + dup2(saved, STDOUT_FILENO); + close(saved); + + pthread_join(reader, NULL); + + if (!result.empty()) + { + return result; + } + + return "(no output)"; +} + +/* ---- 异步命令通道实现 ---- */ +static std::map g_async_cmd_conn_map; +static pthread_mutex_t g_async_cmd_conn_mutex = PTHREAD_MUTEX_INITIALIZER; + +void ws_set_async_cmd_conn(const char *module, unsigned long conn_id) +{ + if (!module) + { + return; + } + + pthread_mutex_lock(&g_async_cmd_conn_mutex); + g_async_cmd_conn_map[module] = conn_id; + pthread_mutex_unlock(&g_async_cmd_conn_mutex); +} + +unsigned long ws_get_async_cmd_conn(const char *module) +{ + if (!module) + { + return 0; + } + + pthread_mutex_lock(&g_async_cmd_conn_mutex); + auto it = g_async_cmd_conn_map.find(module); + unsigned long id = (it != g_async_cmd_conn_map.end()) ? it->second : 0; + pthread_mutex_unlock(&g_async_cmd_conn_mutex); + + return id; +} + +/* ---- 异步命令输出（给其他模块调用） ---- */ +void async_cmd_printf(const char *module, const char *fmt, ...) +{ + if (!module || !fmt) + { + return; + } + + unsigned long conn_id = ws_get_async_cmd_conn(module); + + if (conn_id == 0) + { + return; + } + + va_list ap; + char buf[2048]; + + va_start(ap, fmt); + int n = vsnprintf(buf, sizeof(buf), fmt, ap); + va_end(ap); + + if (n > 0) + { + ws_send_binary(conn_id, buf, (size_t)n); + } +} + +void async_cmd_complete(const char *module, bool success, const char *msg) +{ + if (!module || !msg) + { + return; + } + + unsigned long conn_id = ws_get_async_cmd_conn(module); + + if (conn_id == 0) + { + return; + } + + char buf[1024]; + int n = snprintf(buf, sizeof(buf), "[%s] %s %s\n", + module, success ? "OK" : "FAIL", msg); + + ws_send_binary(conn_id, buf, (size_t)n); +} + +/* ================================================================ + * X-Macro 框架入口 + * ================================================================ */ + +int app_web_server_init1(void *arg) +{ + (void)arg; + + mg_mgr_init(&s_ws_mgr); + mg_http_listen(&s_ws_mgr, WS_LISTEN_ON, ws_event_handler, NULL); + + LOG_I("web_server listening on %s, ws=%s", WS_LISTEN_ON, WS_UPGRADE_URI); + + return 0; +} + +int app_web_server_init2(void *arg) +{ + (void)arg; + return 0; +} + +void *app_web_server(void *arg) +{ + if (!arg) + { + LOG_E("app_web_server arg null"); + return NULL; + } + + stru_app *p_app = (stru_app *)arg; + + uint32_t event; + uint32_t cnt_1s = 0; + + while (1) + { + task_event_recv(p_app->p_event, + EV_TIMER1 | EV_TIMER2 | EV_TIMER3, + TASK_EVENT_FLAG_OR | TASK_EVENT_FLAG_CLEAR, + TASK_EVENT_WAIT_FOREVER, + &event); + + if (event & EV_STOP) + { + LOG_I("web_server stopping"); + + break; + } + + if (event & EV_TIMER1) + { + /* IO 轮询（非阻塞） */ + mg_mgr_poll(&s_ws_mgr, 0); + } + + if (event & EV_TIMER2) + { + /* 增量推送 */ + ws_push_task(); + } + + if (event & EV_TIMER3) + { + cnt_1s++; + p_app->run_cnt++; + } + } + + mg_mgr_free(&s_ws_mgr); + + LOG_I("web_server stopped, total 1s cycles=%u", cnt_1s); + + return NULL; +} diff --git a/src/system/libweb_server/src/ws_method.cpp b/src/system/libweb_server/src/ws_method.cpp new file mode 100644 index 0000000..c1523d5 --- /dev/null +++ b/src/system/libweb_server/src/ws_method.cpp @@ -0,0 +1,671 @@ +/** + * @file ws_method.cpp + * @brief WebSocket 命令解析与信号 CRUD 操作 + * @details 接收 JSON 命令（{type, saddr, signal_type, curd, signal_data}）， + * 通过 g_ws_sig_methods 函数表分发到对应的信号增删改查函数。 + * 支持 cmd/get_cmds/get_file/set_file 等管理命令。 + */ +#include "web_server.h" +#include "myLog.h" +#include "myDatacenter.h" +#include "myCmd.h" +#include "myFunc.h" +#include "myMongoose.h" +#include "cJSON.h" + +#include +#include +#include +#include +#include + +/* ---- 前向声明（来自 web_server.cpp） ---- */ +extern std::string ws_cmd_exec(const char *cmd_line); + +/* ---- Per-Connection Session ---- */ + +typedef struct +{ + std::string saddr; + std::string desc; + uint8_t data_type; + uint8_t ctrl_type; + std::vector vec_p_data; + std::string last_val; +} stru_ws_sig; + +typedef struct +{ + std::vector signals[WS_SIG_MAX]; +} stru_ws_session; + +static std::map g_ws_sessions; +static pthread_mutex_t g_ws_session_mutex = PTHREAD_MUTEX_INITIALIZER; + +/* ---- 前向声明 ---- */ +static int ws_make_sig_json(stru_ws_session *s, cJSON *root, enum_ws_sig_type st, bool *has_change); +static void ws_sig_add(stru_ws_session *s, enum_ws_sig_type st, const char *saddr); +static void ws_sig_del(stru_ws_session *s, enum_ws_sig_type st, const char *saddr); +static int ws_sig_set(stru_ws_session *s, enum_ws_sig_type st, const char *saddr, + uint8_t setting_zone, const char *val); + +/* ---- 函数表: 信号类型 → 名称 ---- */ +static const char *g_ws_sig_names[WS_SIG_MAX] = +{ + "out", "in", "yk", "ao", "param" +}; + +/* ================================================================ + * 内部工具 + * ================================================================ */ + +static int ws_sig_find_idx(const std::vector &vec, const char *saddr) +{ + for (size_t i = 0; i < vec.size(); i++) + { + if (vec[i].saddr == saddr) + { + return (int)i; + } + } + + return -1; +} + +static std::string ws_get_val_str(const void *p_data, uint8_t data_type) +{ + char buf[128] = {0}; + dc_get_signal_val(p_data, data_type, buf, sizeof(buf)); + + return std::string(buf); +} + +/* ================================================================ + * 信号增删改 — out/in/yk/ao/param 分 5 种路径 + * ================================================================ */ + +static void ws_sig_add_out(stru_ws_session *s, const char *saddr) +{ + std::vector &vec = s->signals[WS_SIG_OUT]; + void *p_data = NULL; + + if (ws_sig_find_idx(vec, saddr) >= 0) + { + return; + } + + stru_ws_sig sig; + sig.saddr = saddr; + sig.data_type = 0; + + char desc[128] = {0}; + + if (0 != dc_get_out_signal_info(saddr, desc, sizeof(desc), &sig.data_type, &p_data)) + { + LOG_E("dc_get_out_signal_info failed: %s", saddr); + + return; + } + + sig.desc = desc; + sig.vec_p_data.push_back(p_data); + vec.push_back(sig); +} + +static void ws_sig_add_in(stru_ws_session *s, const char *saddr) +{ + std::vector &vec = s->signals[WS_SIG_IN]; + void *p_data = NULL; + + if (ws_sig_find_idx(vec, saddr) >= 0) + { + return; + } + + stru_ws_sig sig; + sig.saddr = saddr; + + char desc[128] = {0}; + + if (0 != dc_get_in_signal_info(saddr, desc, sizeof(desc), &sig.data_type, &p_data)) + { + LOG_E("dc_get_in_signal_info failed: %s", saddr); + + return; + } + + sig.desc = desc; + sig.vec_p_data.push_back(p_data); + vec.push_back(sig); +} + +static void ws_sig_add_yk(stru_ws_session *s, const char *saddr) +{ + std::vector &vec = s->signals[WS_SIG_YK]; + void *p_data = NULL; + + if (ws_sig_find_idx(vec, saddr) >= 0) + { + return; + } + + stru_ws_sig sig; + sig.saddr = saddr; + + char desc[128] = {0}; + + if (0 != dc_get_yk_signal_info(saddr, desc, sizeof(desc), &sig.data_type, &sig.ctrl_type, &p_data)) + { + LOG_E("dc_get_yk_signal_info failed: %s", saddr); + + return; + } + + sig.desc = desc; + sig.vec_p_data.push_back(p_data); + vec.push_back(sig); +} + +static void ws_sig_add_ao(stru_ws_session *s, const char *saddr) +{ + std::vector &vec = s->signals[WS_SIG_AO]; + void *p_data = NULL; + + if (ws_sig_find_idx(vec, saddr) >= 0) + { + return; + } + + stru_ws_sig sig; + sig.saddr = saddr; + + char desc[128] = {0}; + + if (0 != dc_get_ao_signal_info(saddr, desc, sizeof(desc), &sig.data_type, NULL, &sig.ctrl_type, &p_data)) + { + LOG_E("dc_get_ao_signal_info failed: %s", saddr); + + return; + } + + sig.desc = desc; + sig.vec_p_data.push_back(p_data); + vec.push_back(sig); +} + +static void ws_sig_add_param(stru_ws_session *s, const char *saddr) +{ + std::vector &vec = s->signals[WS_SIG_PARAM]; + void *p_data_arr[16] = {NULL}; + int num_zones = 0; + + if (ws_sig_find_idx(vec, saddr) >= 0) + { + return; + } + + stru_ws_sig sig; + sig.saddr = saddr; + + char desc[128] = {0}; + + if (0 != dc_get_param_signal_info(saddr, desc, sizeof(desc), &sig.data_type, NULL, &sig.ctrl_type, p_data_arr, &num_zones)) + { + LOG_E("dc_get_param_signal_info failed: %s", saddr); + + return; + } + + sig.desc = desc; + + for (int i = 0; i < num_zones; i++) + { + sig.vec_p_data.push_back(p_data_arr[i]); + } + + vec.push_back(sig); +} + +static void ws_sig_add(stru_ws_session *s, enum_ws_sig_type st, const char *saddr) +{ + switch (st) + { + case WS_SIG_OUT: ws_sig_add_out(s, saddr); break; + case WS_SIG_IN: ws_sig_add_in(s, saddr); break; + case WS_SIG_YK: ws_sig_add_yk(s, saddr); break; + case WS_SIG_AO: ws_sig_add_ao(s, saddr); break; + case WS_SIG_PARAM: ws_sig_add_param(s, saddr); break; + default: break; + } +} + +static void ws_sig_del(stru_ws_session *s, enum_ws_sig_type st, const char *saddr) +{ + std::vector &vec = s->signals[st]; + + for (size_t i = 0; i < vec.size(); i++) + { + if (vec[i].saddr == saddr) + { + vec.erase(vec.begin() + i); + + return; + } + } +} + +static int ws_sig_set(stru_ws_session *s, enum_ws_sig_type st, const char *saddr, + uint8_t setting_zone, const char *val) +{ + std::vector &vec = s->signals[st]; + uint8_t data[128] = {0}; + + for (size_t i = 0; i < vec.size(); i++) + { + if (vec[i].saddr != saddr) + { + continue; + } + + dc_set_val_from_str(data, vec[i].data_type, val); + + switch (st) + { + case WS_SIG_OUT: + return dc_set_out_signal_val(saddr, data, MODULE_WEB_SERVER); + + case WS_SIG_IN: + /* in 信号只读 */ + LOG_E("set in signal not supported: %s", saddr); + + return -1; + + case WS_SIG_YK: + { + dc_ctrl_t ctrl; + ctrl.type = vec[i].ctrl_type; + ctrl.step = DC_STEP_READY; + ctrl.data_type = vec[i].data_type; + ctrl.p_data = dc_create_data(vec[i].data_type); + + int rc = dc_signal_yk_set_status(saddr, DC_STEP_DIRECT, &ctrl, data, MODULE_WEB_SERVER); + dc_destroy_data(ctrl.p_data, vec[i].data_type); + + return rc; + } + + case WS_SIG_AO: + { + dc_ctrl_t ctrl; + ctrl.type = vec[i].ctrl_type; + ctrl.step = DC_STEP_READY; + ctrl.data_type = vec[i].data_type; + ctrl.p_data = dc_create_data(vec[i].data_type); + + int rc = dc_signal_ao_set_val(saddr, DC_STEP_DIRECT, &ctrl, data, MODULE_WEB_SERVER); + dc_destroy_data(ctrl.p_data, vec[i].data_type); + + return rc; + } + + case WS_SIG_PARAM: + { + dc_ctrl_t ctrl; + ctrl.type = vec[i].ctrl_type; + ctrl.step = DC_STEP_READY; + ctrl.data_type = vec[i].data_type; + ctrl.p_data = dc_create_data(vec[i].data_type); + + int rc = dc_signal_param_set_val(saddr, DC_STEP_DIRECT, &ctrl, setting_zone, data, MODULE_WEB_SERVER); + dc_destroy_data(ctrl.p_data, vec[i].data_type); + + return rc; + } + + default: + break; + } + } + + LOG_E("signal not found in session: %s", saddr); + + return -1; +} + +/* ================================================================ + * JSON 构建 — 按信号类型有不同字段 + * ================================================================ */ + +static int ws_make_sig_json(stru_ws_session *s, cJSON *root, enum_ws_sig_type st, bool *has_change) +{ + const std::vector &vec = s->signals[st]; + + if (vec.empty()) + { + return 0; + } + + cJSON *arr = cJSON_CreateArray(); + cJSON_AddItemToObject(root, g_ws_sig_names[st], arr); + + for (size_t i = 0; i < vec.size(); i++) + { + const stru_ws_sig *sig = &vec[i]; + cJSON *item = cJSON_CreateObject(); + + cJSON_AddStringToObject(item, "saddr", sig->saddr.c_str()); + cJSON_AddStringToObject(item, "desc", sig->desc.c_str()); + cJSON_AddStringToObject(item, "type", dc_get_type_name(sig->data_type)); + + /* yk/ao/param: 控制类型 */ + if (st == WS_SIG_YK || st == WS_SIG_AO || st == WS_SIG_PARAM) + { + cJSON_AddNumberToObject(item, "ctrl_type", sig->ctrl_type); + } + + /* param: 多定值区 */ + if (st == WS_SIG_PARAM) + { + cJSON *zone_arr = cJSON_CreateArray(); + cJSON_AddItemToObject(item, "setting_zone_list", zone_arr); + + for (size_t j = 0; j < sig->vec_p_data.size(); j++) + { + cJSON *zone_item = cJSON_CreateObject(); + + std::string id = std::to_string(j); + std::string val = ws_get_val_str(sig->vec_p_data[j], sig->data_type); + + cJSON_AddStringToObject(zone_item, "id", id.c_str()); + cJSON_AddStringToObject(zone_item, "val", val.c_str()); + cJSON_AddItemToArray(zone_arr, zone_item); + } + + /* 组合值用于变化检测 */ + std::string combined; + for (size_t j = 0; j < sig->vec_p_data.size(); j++) + { + combined += ws_get_val_str(sig->vec_p_data[j], sig->data_type); + combined += ","; + } + + if (combined != sig->last_val) + { + /* 注意: last_val 是 const 方法访问，这里需要 const_cast */ + const_cast(sig)->last_val = combined; + *has_change = true; + } + } + else + { + /* out/in/yk/ao: 单值 */ + std::string val = ws_get_val_str(sig->vec_p_data[0], sig->data_type); + + cJSON_AddStringToObject(item, "val", val.c_str()); + + if (val != sig->last_val) + { + const_cast(sig)->last_val = val; + *has_change = true; + } + } + + cJSON_AddItemToArray(arr, item); + } + + return 0; +} + +/* ================================================================ + * 增量推送引擎（在 EV_TIMER2 中调用） + * ================================================================ */ + +void ws_push_task(void) +{ + pthread_mutex_lock(&g_ws_session_mutex); + + for (auto &pair : g_ws_sessions) + { + unsigned long conn_id = pair.first; + stru_ws_session &sess = pair.second; + + bool has_any = false; + for (int st = 0; st < WS_SIG_MAX; st++) + { + if (!sess.signals[st].empty()) + { + has_any = true; + break; + } + } + + if (!has_any) + { + continue; + } + + cJSON *root = cJSON_CreateObject(); + + bool has_change = false; + + for (int st = 0; st < WS_SIG_MAX; st++) + { + ws_make_sig_json(&sess, root, (enum_ws_sig_type)st, &has_change); + } + + if (!has_change) + { + cJSON_Delete(root); + continue; + } + + char *p_tx = cJSON_PrintUnformatted(root); + cJSON_Delete(root); + + if (p_tx) + { + ws_send_one(conn_id, p_tx, strlen(p_tx)); + free(p_tx); + } + } + + pthread_mutex_unlock(&g_ws_session_mutex); +} + +/* ================================================================ + * WebSocket JSON 命令分发 + * ================================================================ */ + +void ws_recv(struct mg_connection *c, const char *p_rx, uint16_t rx_len) +{ + if (!p_rx || rx_len == 0) + { + return; + } + + cJSON *root = cJSON_Parse(p_rx); + + if (!root) + { + LOG_E("ws_recv parse failed: %.200s", p_rx); + + return; + } + + /* ---- type=cmd: 远程命令执行 ---- */ + cJSON *jt = cJSON_GetObjectItem(root, "type"); + + if (jt && cJSON_IsString(jt) && jt->valuestring) + { + if (0 == strcmp(jt->valuestring, "cmd")) + { + cJSON *jd = cJSON_GetObjectItem(root, "data"); + + if (jd && cJSON_IsString(jd) && jd->valuestring) + { + /* 提取模块名存异步通道 */ + char mod[64] = {0}; + const char *pp = jd->valuestring; + int mi = 0; + + while (*pp && *pp != ' ' && *pp != '\t' && mi < 63) + { + mod[mi++] = *pp++; + } + + ws_set_async_cmd_conn(mod, c->id); + + std::string out = ws_cmd_exec(jd->valuestring); + ws_send_binary(c->id, out.data(), out.size()); + } + + cJSON_Delete(root); + + return; + } + + /* ---- type=get_cmds: 获取命令列表 ---- */ + if (0 == strcmp(jt->valuestring, "get_cmds")) + { + unsigned int count = 0; + stru_cmd *cmds = cmd_manager_get_commands(&count); + + cJSON *resp = cJSON_CreateObject(); + cJSON_AddStringToObject(resp, "type", "cmd_list"); + cJSON *arr = cJSON_CreateArray(); + + for (unsigned int i = 0; i < count; i++) + { + cJSON *item = cJSON_CreateObject(); + cJSON_AddStringToObject(item, "name", cmds[i].name); + cJSON_AddStringToObject(item, "desc", cmds[i].desc); + cJSON_AddItemToArray(arr, item); + } + + cJSON_AddItemToObject(resp, "cmds", arr); + + char *p_tx = cJSON_PrintUnformatted(resp); + + if (p_tx) + { + ws_send_one(c->id, p_tx, strlen(p_tx)); + free(p_tx); + } + + cJSON_Delete(resp); + cJSON_Delete(root); + + return; + } + } + + /* ---- 信号 CRUD 命令 ---- */ + cJSON *jsaddr = cJSON_GetObjectItem(root, "saddr"); + cJSON *jsigtype = cJSON_GetObjectItem(root, "signal_type"); + cJSON *jcurd = cJSON_GetObjectItem(root, "curd"); + cJSON *jsigdata = cJSON_GetObjectItem(root, "signal_data"); + cJSON *jzone = cJSON_GetObjectItem(root, "setting_zone"); + + if (!jsaddr || !cJSON_IsString(jsaddr) || !jsaddr->valuestring) + { + cJSON_Delete(root); + + return; + } + + if (!jsigtype || !cJSON_IsString(jsigtype) || !jsigtype->valuestring) + { + cJSON_Delete(root); + + return; + } + + if (!jcurd || !cJSON_IsString(jcurd) || !jcurd->valuestring) + { + cJSON_Delete(root); + + return; + } + + /* 查找信号类型 */ + enum_ws_sig_type st = WS_SIG_OUT; + + for (int i = 0; i < WS_SIG_MAX; i++) + { + if (0 == strcmp(jsigtype->valuestring, g_ws_sig_names[i])) + { + st = (enum_ws_sig_type)i; + break; + } + } + + const char *saddr = jsaddr->valuestring; + const char *curd = jcurd->valuestring; + const char *sig_data = jsigdata && cJSON_IsString(jsigdata) ? jsigdata->valuestring : ""; + uint8_t zone = jzone && cJSON_IsNumber(jzone) ? (uint8_t)jzone->valueint : 0; + + pthread_mutex_lock(&g_ws_session_mutex); + + stru_ws_session &sess = g_ws_sessions[c->id]; + + if (0 == strcmp(curd, "add")) + { + ws_sig_add(&sess, st, saddr); + + /* 立即推送当前状态 */ + cJSON *push = cJSON_CreateObject(); + bool dummy = false; + + ws_make_sig_json(&sess, push, st, &dummy); + + char *p_tx = cJSON_PrintUnformatted(push); + + if (p_tx) + { + ws_send_one(c->id, p_tx, strlen(p_tx)); + free(p_tx); + } + + cJSON_Delete(push); + } + else if (0 == strcmp(curd, "del")) + { + ws_sig_del(&sess, st, saddr); + + cJSON *push = cJSON_CreateObject(); + bool dummy = false; + + ws_make_sig_json(&sess, push, st, &dummy); + + char *p_tx = cJSON_PrintUnformatted(push); + + if (p_tx) + { + ws_send_one(c->id, p_tx, strlen(p_tx)); + free(p_tx); + } + + cJSON_Delete(push); + } + else if (0 == strcmp(curd, "set")) + { + ws_sig_set(&sess, st, saddr, zone, sig_data); + } + + pthread_mutex_unlock(&g_ws_session_mutex); + + cJSON_Delete(root); +} + +/* ---- 会话销毁 ---- */ +void ws_session_destroy(struct mg_connection *c) +{ + if (!c) + { + return; + } + + pthread_mutex_lock(&g_ws_session_mutex); + g_ws_sessions.erase(c->id); + pthread_mutex_unlock(&g_ws_session_mutex); +}