From 72bdcae67fcbce93de93ebc1b0398b058da61977 Mon Sep 17 00:00:00 2001 From: xue <2144439972@qq.com> Date: Mon, 15 Sep 2025 19:40:49 +0800 Subject: [PATCH] =?utf8?q?websocket=E6=A1=86=E6=9E=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=utf8 Content-Transfer-Encoding: 8bit --- .../sdk/zhihui/leiniao/imsl_ln/CMakeLists.txt | 3 + .../zhihui/leiniao/imsl_ln/leiniao_api_at.c | 57 +++++ code/sdk/zhihui/leiniao/imsl_ln/ln_api.c | 87 +++++++ code/sdk/zhihui/leiniao/imsl_ln/ln_api.h | 20 +- .../zhihui/leiniao/imsl_ln/ln_api_websocket.c | 237 ++++++++++++++++++ 5 files changed, 403 insertions(+), 1 deletion(-) create mode 100644 code/sdk/zhihui/leiniao/imsl_ln/ln_api_websocket.c diff --git a/code/sdk/zhihui/leiniao/imsl_ln/CMakeLists.txt b/code/sdk/zhihui/leiniao/imsl_ln/CMakeLists.txt index cb1e08d00..99a82e328 100644 --- a/code/sdk/zhihui/leiniao/imsl_ln/CMakeLists.txt +++ b/code/sdk/zhihui/leiniao/imsl_ln/CMakeLists.txt @@ -17,6 +17,7 @@ set(SOURCES ${CMAKE_CURRENT_SOURCE_DIR}/ln_api_story_finish.c ${CMAKE_CURRENT_SOURCE_DIR}/ln_api_device_info.c ${CMAKE_CURRENT_SOURCE_DIR}/ln_api_ota.c + ${CMAKE_CURRENT_SOURCE_DIR}/ln_api_websocket.c ) # 公共头文件路径（供其他模块引用） @@ -26,6 +27,8 @@ set(PUBLIC_HEADER ${ROOT_DIR}/open_source/openharmony/third_party/lwip/lwip_v2.1.3/src/include ${ROOT_DIR}/open_source/openharmony/third_party/lwip/lwip_adapter/liteos/src/include ${ROOT_DIR}/open_source/openharmony/third_party/cJSON/ + ${ROOT_DIR}/open_source/openharmony/foundation/ace/ace_engine_lite/modules/netstack/websocket/include + ) diff --git a/code/sdk/zhihui/leiniao/imsl_ln/leiniao_api_at.c b/code/sdk/zhihui/leiniao/imsl_ln/leiniao_api_at.c index 7b48c9759..d83cadbc5 100644 --- a/code/sdk/zhihui/leiniao/imsl_ln/leiniao_api_at.c +++ b/code/sdk/zhihui/leiniao/imsl_ln/leiniao_api_at.c @@ -635,7 +635,57 @@ static uint32_t ln_ota_download_at(uint8_t *para, uint32_t paraLen, char *strCom do_ln_http_server_send_data(LN_HTTP_SERVICE_CMD_OTA_DOWNLOAD, NULL, 0); return ERRCODE_SUCC; } +/* ----------------------------- WebSocket 控制 ----------------------------- */ +/* AT^LN_WS_START */ +static uint32_t ln_ws_start_at(uint8_t *para, uint32_t paraLen, char *strCommonCmd) +{ + VAD_UNUSED(para); VAD_UNUSED(paraLen); VAD_UNUSED(strCommonCmd); + printf("[LN][WS_START]\n"); + int ret = ln_ws_start(); + if (ret != 0) + { + printf("[LN][WS_START] failed: %d\n", ret); + return ERRCODE_FAIL; + } + return ERRCODE_SUCC; +} + +/* AT^LN_WS_STOP */ +static uint32_t ln_ws_stop_at(uint8_t *para, uint32_t paraLen, char *strCommonCmd) +{ + VAD_UNUSED(para); VAD_UNUSED(paraLen); VAD_UNUSED(strCommonCmd); + printf("[LN][WS_STOP]\n"); + int ret = ln_ws_stop(); + if (ret != 0) + { + printf("[LN][WS_STOP] failed: %d\n", ret); + return ERRCODE_FAIL; + } + return ERRCODE_SUCC; +} +/* AT^LN_WS_SEND= */ +static uint32_t ln_ws_send_text_at(uint8_t *para, uint32_t paraLen, char *strCommonCmd) +{ + VAD_UNUSED(strCommonCmd); + uint8_t argv[LN_MAX_ARGC][LN_MAX_ARGV_LEN] = {0}; + uint32_t argc = 0; + (void)ConvertCmdToParam(para, paraLen, argv, &argc); + if (argc < 1) + { + printf("Usage: AT^LN_WS_SEND=\n"); + return ERRCODE_FAIL; + } + const char *text = (const char *)argv[0]; + int ret = ln_ws_send_text(text); + if (ret != 0) + { + printf("[LN][WS_SEND] failed: %d\n", ret); + return ERRCODE_FAIL; + } + printf("[LN][WS_SEND] ok: %s\n", text); + return ERRCODE_SUCC; +} /* 音频播放: AT^LN_STORY_PLAY_NEXT */ static uint32_t ln_audio_play_at(uint8_t *para, uint32_t paraLen, char *strCommonCmd) { @@ -714,6 +764,12 @@ static const at_cmd_table_t g_ln_api_cmd[] = { {(char *)"AT^LN_DEV_USERID", ln_device_get_userid_at, (char *)"获取设备绑定用户ID(JSON,GET): AT^LN_DEV_USERID"}, {(char *)"AT^LN_OTA_GET_TASK", ln_ota_get_task_at, (char *)"获取OTA升级任务(JSON,GET): AT^LN_OTA_GET_TASK"}, {(char *)"AT^LN_OTA_DOWNLOAD", ln_ota_download_at, (char *)"下载OTA升级包(JSON,GET->下载): AT^LN_OTA_DOWNLOAD"}, + + /* -------------------------------- WebSocket -------------------------------- */ + {(char *)"AT^LN_WS_START", ln_ws_start_at, (char *)"启动WebSocket连接: AT^LN_WS_START"}, + {(char *)"AT^LN_WS_STOP", ln_ws_stop_at, (char *)"停止WebSocket连接: AT^LN_WS_STOP"}, + {(char *)"AT^LN_WS_SEND", ln_ws_send_text_at, (char *)"发送WS文本: AT^LN_WS_SEND=hello"}, + /* 音频播放 */ {(char *)"AT^LN_STORY_PLAY_AT", ln_audio_play_at, (char *)"播放音频: AT^LN_STORY_PLAY_AT"}, {(char *)"AT^LN_STORY_STOP_AT", ln_audio_stop_at, (char *)"停止播放音频: AT^LN_STORY_STOP_AT"}, @@ -722,6 +778,7 @@ static const at_cmd_table_t g_ln_api_cmd[] = { {(char *)"AT^LN_AUDIO_RESUME", ln_audio_resume_at, (char *)"恢复播放: AT^LN_AUDIO_RESUME"}, }; + at_cmd_table_t *get_leiniao_api_cmd_table(uint32_t *cmd_count) { *cmd_count = ARRAY_COUNT(g_ln_api_cmd); diff --git a/code/sdk/zhihui/leiniao/imsl_ln/ln_api.c b/code/sdk/zhihui/leiniao/imsl_ln/ln_api.c index bcb0a6f55..16c88dac3 100644 --- a/code/sdk/zhihui/leiniao/imsl_ln/ln_api.c +++ b/code/sdk/zhihui/leiniao/imsl_ln/ln_api.c @@ -366,6 +366,93 @@ static void handle_ota_download_msg(const OtaDownloadMsg *msg) /* ============================ OTA 包下载实现 End ============================ */ + +/* ============================ WebSocket 业务消息 上层处理 ============================ */ +void ln_ws_process_message(int cmd, const char *json_str) +{ + if (!json_str) return; + cJSON *root = cJSON_Parse(json_str); + if (!root) { + printf("[WS] invalid json\n"); + return; + } + const cJSON *msgId = cJSON_GetObjectItem(root, "messageId"); + const cJSON *command = cJSON_GetObjectItem(root, "command"); + const char *message_id = (msgId && cJSON_IsString(msgId)) ? msgId->valuestring : NULL; + const char *cmd_str = (command && cJSON_IsString(command)) ? command->valuestring : ""; + + switch (cmd) { + case LN_WS_RECV_CHARACTER_CHANGE: { + const cJSON *data = cJSON_GetObjectItem(root, "data"); + const cJSON *name = data ? cJSON_GetObjectItem(data, "characterName") : NULL; + const cJSON *cid = data ? cJSON_GetObjectItem(data, "characterId") : NULL; + printf("[WS] CharacterChange: name=%s id=%s\n", + (name && cJSON_IsString(name)) ? name->valuestring : "", + (cid && cJSON_IsString(cid)) ? cid->valuestring : ""); + break; + } + case LN_WS_RECV_CHARACTER_CONFIG_CHANGE: { + printf("[WS] CharacterConfigChange\n"); + break; + } + case LN_WS_RECV_SLEEP_MODE_START: { + const cJSON *data = cJSON_GetObjectItem(root, "data"); + const cJSON *theme = data ? cJSON_GetObjectItem(data, "storyTheme") : NULL; + const char *device_id = uapi_ln_get_device_id(); + const char *token = uapi_ln_get_device_token(); + if (device_id && token) { + ln_story_random_http(device_id, token, + (theme && cJSON_IsString(theme)) ? theme->valuestring : NULL); + } + printf("[WS] SleepModeStart: storyTheme=%s\n", + (theme && cJSON_IsString(theme)) ? theme->valuestring : ""); + break; + } + case LN_WS_RECV_SLEEP_MODE_END: { + const char *device_id = uapi_ln_get_device_id(); + const char *token = uapi_ln_get_device_token(); + if (device_id && token) { + ln_story_finish_http(device_id, token); + } + printf("[WS] SleepModeEnd\n"); + break; + } + case LN_WS_RECV_DEVICE_BIND: { + const cJSON *data = cJSON_GetObjectItem(root, "data"); + const cJSON *userId = data ? cJSON_GetObjectItem(data, "userId") : NULL; + printf("[WS] DeviceBind: userId=%s\n", + (userId && cJSON_IsString(userId)) ? userId->valuestring : ""); + break; + } + case LN_WS_RECV_DEVICE_UNBIND: { + const cJSON *data = cJSON_GetObjectItem(root, "data"); + const cJSON *userId = data ? cJSON_GetObjectItem(data, "userId") : NULL; + printf("[WS] DeviceUnbind: userId=%s\n", + (userId && cJSON_IsString(userId)) ? userId->valuestring : ""); + break; + } + case LN_WS_RECV_SCHEDULE_REMINDER: { + const cJSON *data = cJSON_GetObjectItem(root, "data"); + const cJSON *name = data ? cJSON_GetObjectItem(data, "scheduleName") : NULL; + printf("[WS] ScheduleReminder: %s\n", + (name && cJSON_IsString(name)) ? name->valuestring : ""); + break; + } + case LN_WS_RECV_SCHEDULE_REMINDER_CFG: { + const cJSON *data = cJSON_GetObjectItem(root, "data"); + const cJSON *name = data ? cJSON_GetObjectItem(data, "scheduleName") : NULL; + printf("[WS] ScheduleReminderConfig: %s\n", + (name && cJSON_IsString(name)) ? name->valuestring : ""); + break; + } + default: { + printf("[WS] Unknown command=%s\n", cmd_str); + break; + } + } + cJSON_Delete(root); +} + // 发送队列 static void ln_http_server_send_task(void const *arg) { diff --git a/code/sdk/zhihui/leiniao/imsl_ln/ln_api.h b/code/sdk/zhihui/leiniao/imsl_ln/ln_api.h index 6bde4a0bd..29fe6072a 100644 --- a/code/sdk/zhihui/leiniao/imsl_ln/ln_api.h +++ b/code/sdk/zhihui/leiniao/imsl_ln/ln_api.h @@ -77,6 +77,18 @@ typedef struct #define LN_HTTP_SERVICE_CMD_OTA_DOWNLOAD 0x51 /* 触发 OTA 包下载（根据 getTask 的返回 packageUrl）*/ #define LN_HTTP_RECV_OTA_DOWNLOAD_MSG 0x61 /* 可选：OTA 下载控制消息（JSON）接收）*/ + +/* WebSocket 业务消息接收命令（供 WS 回调上报分类使用）*/ +#define LN_WS_RECV_UNKNOWN 0x70 +#define LN_WS_RECV_CHARACTER_CHANGE 0x71 +#define LN_WS_RECV_CHARACTER_CONFIG_CHANGE 0x72 +#define LN_WS_RECV_SLEEP_MODE_START 0x73 +#define LN_WS_RECV_SLEEP_MODE_END 0x74 +#define LN_WS_RECV_DEVICE_BIND 0x75 +#define LN_WS_RECV_DEVICE_UNBIND 0x76 +#define LN_WS_RECV_SCHEDULE_REMINDER 0x77 +#define LN_WS_RECV_SCHEDULE_REMINDER_CFG 0x78 + /* 音频播放相关命令 */ #define LN_HTTP_SERVICE_CMD_AUDIO_PLAY 0x1001 /* 播放音频 */ #define LN_HTTP_SERVICE_CMD_AUDIO_STOP 0x1002 /* 停止播放音频 */ @@ -201,7 +213,13 @@ uint32_t ln_ota_get_task_http(const char *device_id, uint32_t ln_ota_download_flow_http(const char *device_id, const char *device_token, const char *source_version); -/* 文件解析和数据处理函数 */ + +/* ============================== WebSocket 客户端接口 ============================== */ +int ln_ws_start(void); +int ln_ws_send_text(const char *text); +int ln_ws_stop(void); + + /* 文件解析和数据处理函数 */ uint32_t ln_parse_file_and_process_data(const char *file_path, int interface_type); uint32_t ln_parse_and_process_json_data(const char *json_data, int interface_type, int segment_index); #ifdef __cplusplus diff --git a/code/sdk/zhihui/leiniao/imsl_ln/ln_api_websocket.c b/code/sdk/zhihui/leiniao/imsl_ln/ln_api_websocket.c new file mode 100644 index 000000000..ddf37ce38 --- /dev/null +++ b/code/sdk/zhihui/leiniao/imsl_ln/ln_api_websocket.c @@ -0,0 +1,237 @@ +#include "ln_api.h" +#include "net_websocket.h" +#include +#include +#include +#include +#include "cJSON.h" // Added for JSON parsing + +/* + * - g_ws_client_fd 当前 WebSocket 客户端fd + * - g_ws_connected 握手成功后置1，关闭/异常置0 + * - g_ws_running 心跳线程运行标志 + */ +static int g_ws_client_fd = -1; +static volatile int g_ws_connected = 0; +static pthread_t g_ws_heartbeat_thread = 0; +static pthread_t g_ws_supervisor_thread = 0; +static volatile int g_ws_running = 0; +static volatile unsigned int g_ws_reconnect_count = 0; /* 断线后递增，用于URL reconnect参数 */ + +/* 握手成功回调：标记连接就绪 */ +static void ln_ws_on_open(int client) +{ + (void)client; + g_ws_connected = 1; +} + +/* + * 收到服务器下发的文本消息 + * - 上层可根据业务定义的 JSON 字段进行再分发 + */ +static void ln_ws_on_message(int client, const char *data, size_t length, bool isFinished) +{ + (void)client; + (void)isFinished; + if (data == NULL || length == 0) return; + char *buf = (char *)malloc(length + 1); + if (!buf) return; + memcpy(buf, data, length); + buf[length] = '\0'; + + int dispatch_cmd = LN_WS_RECV_UNKNOWN; + + /* 简单判断 */ + if (length >= 2 && buf[0] == '{' && buf[length - 1] == '}') { + /* 使用 cJSON 解析 command 字段 */ + cJSON *root = cJSON_Parse(buf); + if (root) { + cJSON *cmd = cJSON_GetObjectItem(root, "command"); + if (cmd && cJSON_IsString(cmd) && cmd->valuestring) { + const char *c = cmd->valuestring; + if (strcmp(c, "CharacterChange") == 0) { + dispatch_cmd = LN_WS_RECV_CHARACTER_CHANGE; + } else if (strcmp(c, "CharacterConfigChange") == 0) { + dispatch_cmd = LN_WS_RECV_CHARACTER_CONFIG_CHANGE; + } else if (strcmp(c, "SleepModeStart") == 0) { + dispatch_cmd = LN_WS_RECV_SLEEP_MODE_START; + } else if (strcmp(c, "SleepModeEnd") == 0) { + dispatch_cmd = LN_WS_RECV_SLEEP_MODE_END; + } else if (strcmp(c, "DeviceBind") == 0) { + dispatch_cmd = LN_WS_RECV_DEVICE_BIND; + } else if (strcmp(c, "DeviceUnbind") == 0) { + dispatch_cmd = LN_WS_RECV_DEVICE_UNBIND; + } else if (strcmp(c, "ScheduleReminder") == 0) { + dispatch_cmd = LN_WS_RECV_SCHEDULE_REMINDER; + } else if (strcmp(c, "ScheduleReminderConfig") == 0) { + dispatch_cmd = LN_WS_RECV_SCHEDULE_REMINDER_CFG; + } else { + dispatch_cmd = LN_WS_RECV_UNKNOWN; + } + } + cJSON_Delete(root); + } + } + printf("[LN][WS] on_message: %s\n", buf); + + /* 透传给接收队列：上层可基于 dispatch_cmd 做进一步处理 */ + //do_ln_http_server_recv_data(dispatch_cmd, buf, (uint32_t)length); + /* 立即在同线程内调上层处理（可选：如担心阻塞可删除此行，仅由接收任务处理） */ + ln_ws_process_message(dispatch_cmd, buf); + free(buf); +} + +/* 出错回调：仅打印，实际重连策略可在上层统一管理 */ +static void ln_ws_on_error(int client, struct WebSocketErrorResult error) +{ + (void)client; + printf("[LN][WS] error: code=%d msg=%s\n", error.errorCode, error.errorMessage); +} + +/* 断开回调：标记为未连接。是否自动重连交由上层决定 */ +static void ln_ws_on_close(int client) +{ + (void)client; + g_ws_connected = 0; + /* 断开后增加重连计数，由监督线程在下一次连接时携带 */ + if (g_ws_running) { + if (g_ws_reconnect_count < 0xFFFFFFFFu) { + g_ws_reconnect_count++; + } + } +} + +/* + * 心跳线程：每30秒发送一次 "ping" + * - 使用文本帧维持链路活性 + * - 收到的 "pong" 可在 onMessage 中由上层识别（当前透传） + */ +static void *ln_ws_heartbeat(void *arg) +{ + (void)arg; + while (g_ws_running) { + if (g_ws_connected && g_ws_client_fd >= 0) { + const char *ping = "ping"; + (void)WebSocketClientSendText(g_ws_client_fd, ping, strlen(ping)); + } + /* 30s 间隔 */ + for (int i = 0; i < 30 && g_ws_running; i++) { + usleep(1000 * 1000); + } + } + return NULL; +} + +/* + * 拼接连接URL：环境域名 + 业务必须的查询参数 + * - 认证方式采用“查询参数方式”：至少包含 deviceToken（规范允许） + * - 兼容性增强：同时附带 deviceId 与 reconnect 计数，方便服务端校验与审计 + * - 如需切换环境，仅修改 base 即可 + */ +static void ln_ws_build_url(char *out, size_t out_size) +{ + const char *base = "wss://ambybox-api.test.leiniao.com/ambybox-device-api/v1/websocket"; + const char *device_id = uapi_ln_get_device_id(); + const char *token = uapi_ln_get_device_token(); + if (!device_id) device_id = ""; + if (!token) token = ""; + /* 仅使用 deviceToken 也可通过；额外带上 deviceId 与 reconnect 便于后端管理 */ + snprintf(out, out_size, "%s?deviceToken=%s&deviceId=%s&reconnect=%u", + base, token, device_id, g_ws_reconnect_count); +} + +/* + * 监督线程：负责初次连接与断线后的自动重连 + * - 重连次数随断线递增并写入URL的 reconnect 参数 + * - 使用简单指数退避，最大60s + */ +static void *ln_ws_supervisor(void *arg) +{ + (void)arg; + while (g_ws_running) { + if (!g_ws_connected) { + if (g_ws_client_fd < 0) { + g_ws_client_fd = WebSocketClientConstructor(ln_ws_on_open, ln_ws_on_message, ln_ws_on_error, ln_ws_on_close); + if (g_ws_client_fd < 0) { + printf("[LN][WS] constructor failed: %d\n", g_ws_client_fd); + /* 失败则短暂休眠后重试创建 */ + usleep(2000 * 1000); + continue; + } + } + + char url[512] = {0}; + ln_ws_build_url(url, sizeof(url)); + int ret = WebSocketClientConnect(g_ws_client_fd, url); + if (ret != 0) { + printf("[LN][WS] connect failed: %d\n", ret); + (void)WebSocketClientClose(g_ws_client_fd); + g_ws_client_fd = -1; + /* 指数退避*/ + unsigned int delay = 1u << (g_ws_reconnect_count > 5 ? 5 : g_ws_reconnect_count); + if (delay > 60) delay = 60; + for (unsigned int i = 0; i < delay && g_ws_running; i++) { + usleep(1000 * 1000); + } + continue; + } + + /* 首次连接成功，启动心跳线程*/ + if (!g_ws_heartbeat_thread) { + (void)pthread_create(&g_ws_heartbeat_thread, NULL, ln_ws_heartbeat, NULL); + printf("[LN][WS] heartbeat thread created\n"); + } + } + /* 已连接/或刚触发连接，短暂休眠 */ + usleep(500 * 1000); + } + return NULL; +} + +/* + * 启动：开启监督线程与心跳机制（心跳线程首次连接成功后会自然启动） + * - 若已在运行则直接返回成功 + */ +int ln_ws_start(void) +{ + if (g_ws_running) { + return 0; + } + g_ws_running = 1; + g_ws_reconnect_count = 0; /* 启动时从0开始 */ + if (!g_ws_supervisor_thread) { + (void)pthread_create(&g_ws_supervisor_thread, NULL, ln_ws_supervisor, NULL); + } + return 0; +} + +/* 发送文本消息：仅在连接就绪时有效 */ +int ln_ws_send_text(const char *text) +{ + if (!text) return -1; + if (!g_ws_connected || g_ws_client_fd < 0) return -1; + return WebSocketClientSendText(g_ws_client_fd, text, strlen(text)); +} + +/* 停止：关闭心跳并关闭连接 */ +int ln_ws_stop(void) +{ + g_ws_running = 0; + if (g_ws_supervisor_thread) { + pthread_join(g_ws_supervisor_thread, NULL); + g_ws_supervisor_thread = 0; + } + if (g_ws_heartbeat_thread) { + pthread_join(g_ws_heartbeat_thread, NULL); + g_ws_heartbeat_thread = 0; + } + int ret = 0; + if (g_ws_client_fd >= 0) { + ret = WebSocketClientClose(g_ws_client_fd); + g_ws_client_fd = -1; + } + g_ws_connected = 0; + return ret; +} + + -- 2.17.1