#define WS_MSG_TYPE_STOP 0x03
#define WS_MSG_TYPE_RAW_MESSAGE 0x04
+/* 发送方式控制宏 */
+#ifndef WS_USE_SEND_QUEUE
+#define WS_USE_SEND_QUEUE 1 // 1=使用发送队列, 0=不使用发送队列
+#endif
+
+#ifndef WS_SEND_DIRECT_ENABLE
+#define WS_SEND_DIRECT_ENABLE 0 // 0=通过队列发送, 1=直接发送（仅在WS_USE_SEND_QUEUE=0时有效）
+#endif
+
/* 连接状态定义 */
#define WS_STATE_DISCONNECTED 0 // 未连接
#define WS_STATE_CONNECTING 1 // 连接中
static int g_ws_client_fd = -1;
static volatile int g_ws_connected = WS_STATE_DISCONNECTED;
static uint32_t g_ws_main_thread = 0;
+static uint32_t g_ws_send_thread = 0;
static volatile int g_ws_running = 0;
static volatile unsigned int g_ws_reconnect_count = 0; /* 断线后递增，用于URL reconnect参数 */
static osMessageQueueId_t g_ws_msg_queue = NULL;
+static osMessageQueueId_t g_ws_send_queue = NULL;
static ws_call_backs_t g_ws_callbacks = {0}; /* WebSocket回调函数结构体 */
static volatile ws_close_status_t g_ws_close_status = WS_CLOSE_NOT_EXECUTED; /* WebSocket关闭状态标志位 */
volatile int g_ws_is_reconnecting = 0; /* 是否正在重连的标志位 */
}
}
+/* 发送队列创建函数 */
+static void uapi_ws_send_queue_init(void)
+{
+ HAISI_WS_DEBUG("uapi_ws_send_queue_init\n");
+ g_ws_send_queue = osMessageQueueNew(WS_MSG_QUEUE_COUNT, sizeof(ws_msg_info_t*), NULL);
+ if (g_ws_send_queue == NULL) {
+ HAISI_WS_ERROR("uapi_ws_send_queue_init failed: failed to create send queue\n");
+ } else {
+ HAISI_WS_DEBUG("uapi_ws_send_queue_init success: send queue created\n");
+ }
+}
+
/* 握手成功回调：标记连接就绪 */
static void uapi_ws_on_open(int client)
{
if (osMessageQueueGet(g_ws_msg_queue, &msg, NULL, 0) == osOK) {
HAISI_WS_DEBUG("osMessageQueueGet: g_ws_msg_queue\n");
switch (msg->msg_type) {
+ // WS_MSG_TYPE_SEND_TEXT 由发送线程处理，这里放回队列
case WS_MSG_TYPE_SEND_TEXT:
- if (g_ws_connected == WS_STATE_CONNECTED && g_ws_client_fd >= 0) {
- // 对于文本消息，我们需要特殊处理，因为现在data是ws_data_msg_t类型
- // 这里需要重新设计文本消息的存储方式
- HAISI_WS_WARN("Text message type not supported in new structure\n");
- } else {
- HAISI_WS_WARN("Cannot send text, connection state: %d\n", g_ws_connected);
- }
+#if WS_USE_SEND_QUEUE
+ // 发送消息由发送线程处理，放回队列
+ osMessageQueuePut(g_ws_send_queue, &msg, 0, 0);
+ msg = NULL;
+#else
+ // 发送队列未启用，直接释放消息
+ HAISI_WS_WARN("Send queue disabled, dropping SEND_TEXT message\n");
ws_msg_free(msg);
+ msg = NULL;
+#endif
break;
case WS_MSG_TYPE_FORCE_RECONNECT:
// g_ws_client_fd=-1
}
g_ws_connected = WS_STATE_DISCONNECTED;
+}
+
+/**
+ * @brief WebSocket发送线程
+ * 负责处理发送队列中的消息
+ */
+static void uapi_ws_send_thread(void *arg)
+{
+ (void)arg;
+ ws_msg_info_t *msg = NULL;
- // 删除消息队列
- if (g_ws_msg_queue != NULL) {
- ws_msg_info_t msg;
- while (osMessageQueueGet(g_ws_msg_queue, &msg, NULL, 0) == osOK) {
- // 原始消息数据不需要额外释放，因为存储在栈上
- HAISI_WS_DEBUG("Clearing queued message: %s\n", msg.raw_data);
- ws_msg_free(&msg);
+ HAISI_WS_INFO("WebSocket send thread started\n");
+
+ while (g_ws_running) {
+ // 从发送队列获取消息（阻塞等待，超时100ms）
+ osStatus_t status = osMessageQueueGet(g_ws_send_queue, &msg, NULL, 100);
+
+ if (status == osOK && msg != NULL) {
+ if (msg->msg_type == WS_MSG_TYPE_STOP) {
+ HAISI_WS_INFO("Send thread received stop signal\n");
+ ws_msg_free(msg);
+ break;
+ }
+ // 发送线程只处理发送消息类型
+ if (msg->msg_type == WS_MSG_TYPE_SEND_TEXT) {
+ // 检查连接状态
+ if (g_ws_connected == WS_STATE_CONNECTED && g_ws_client_fd >= 0) {
+ if (msg->raw_data && msg->data_len > 0) {
+ HAISI_WS_DEBUG("Sending text message: %s, len: %u\n", msg->raw_data, msg->data_len);
+
+ // 设置发送状态标志
+ g_ws_is_sending = 1;
+
+ // 发送数据
+ int ret = WebSocketClientSendText(g_ws_client_fd, msg->raw_data, msg->data_len);
+
+ HAISI_WS_DEBUG("Send text message ret: %d\n", ret);
+
+ // 更新最后发送时间
+ if (ret == 0) {
+ uapi_ws_update_tx_time();
+ }
+
+ // 清除发送状态标志
+ g_ws_is_sending = 0;
+ } else {
+ HAISI_WS_WARN("Invalid text message data\n");
+ }
+ } else {
+ HAISI_WS_WARN("Cannot send text, connection state: %d, fd: %d\n", g_ws_connected, g_ws_client_fd);
+ }
+
+ // 释放消息内存
+ ws_msg_free(msg);
+ msg = NULL;
+ } else {
+ // 其他消息类型放回队列或直接释放
+ HAISI_WS_WARN("Send thread: Unknown message type %d, dropping\n", msg->msg_type);
+ ws_msg_free(msg);
+ msg = NULL;
+ }
}
- osMessageQueueDelete(g_ws_msg_queue);
- g_ws_msg_queue = NULL;
}
+
+ HAISI_WS_INFO("WebSocket send thread exited\n");
}
/*
* 启动：开启监督线程与心跳机制（心跳线程首次连接成功后会自然启动）
};
(void)uapi_create_osThread_custom_with_args(&main_task);
}
+ #if WS_USE_SEND_QUEUE
+ // 创建发送线程
+ if (!g_ws_send_thread) {
+ app_task_definition_ext_t send_task = {
+ .attr = {
+ .name = "uapi_ws_send",
+ .stack_size = 0x2000,
+ .priority = osPriorityNormal
+ },
+ .func = uapi_ws_send_thread,
+ .task_handle = &g_ws_send_thread,
+ .create_queue_handle = uapi_ws_send_queue_init, // 发送队列已初始化，不需要单独创建
+ .use_mem = USE_EXTEND_MEM,
+ .inst_args = NULL
+ };
+ (void)uapi_create_osThread_custom_with_args(&send_task);
+ }
+ #endif
return 0;
}
int uapi_ws_send_text(const char *text)
{
if (!text) return -1;
+
+#if (!WS_SEND_DIRECT_ENABLE) && WS_USE_SEND_QUEUE
+ // 通过队列发送模式（当WS_SEND_DIRECT_ENABLE=0 && WS_USE_SEND_QUEUE=1时）
+ if (!g_ws_running || !g_ws_send_queue) {
+ HAISI_WS_WARN("Cannot send text via queue, ws_running: %d, send_queue: %p\n", g_ws_running, g_ws_send_queue);
+ return -1;
+ }
+
+ size_t text_len = strlen(text);
+ ws_msg_info_t *msg = ws_msg_create(text_len);
+ if (msg == NULL) {
+ HAISI_WS_ERROR("Failed to create send message\n");
+ return -1;
+ }
+
+ msg->msg_type = WS_MSG_TYPE_SEND_TEXT;
+ msg->timestamp = 0;
+ msg->isBinary = false;
+ memcpy(msg->raw_data, text, text_len);
+ msg->raw_data[text_len] = '\0';
+ msg->data_len = text_len;
+
+ osStatus_t ret = osMessageQueuePut(g_ws_send_queue, &msg, 0, 0);
+ if (ret != osOK) {
+ HAISI_WS_ERROR("Failed to put send message to queue: %d\n", ret);
+ ws_msg_free(msg);
+ return -1;
+ }
+
+ HAISI_WS_DEBUG("Text message queued for sending: %s\n", text);
+ return 0;
+#else
+ // 直接发送模式（WS_SEND_DIRECT_ENABLE=1 或 WS_USE_SEND_QUEUE=0）
int network_status = uapi_dev_runtime_get_cat1_networkstat();
if (network_status != 1 || g_ws_connected != WS_STATE_CONNECTED || g_ws_client_fd < 0) {
HAISI_WS_WARN("Cannot send text, connection state: %d , network_status: %d\n", g_ws_connected,network_status);
// 设置发送状态标志
g_ws_is_sending = 1;
- HAISI_WS_DEBUG("uapi_ws_send_text: %s\n", text);
+ HAISI_WS_DEBUG("uapi_ws_send_text (direct): %s\n", text);
// 发送数据
HAISI_WS_WARN("uapi_ws_send_text \n");
int ret = WebSocketClientSendText(g_ws_client_fd, text, strlen(text));
g_ws_is_sending = 0;
return ret;
+#endif
}
+
+
uint32_t size_uapi_ws_send_BIN=0;
int uapi_ws_send_BIN(uint8_t *data,int len)
/* 通过队列发送文本消息（异步） */
int uapi_ws_send_text_async(const char *text)
{
- if (!text || !g_ws_running) return -1;
+#if WS_USE_SEND_QUEUE
+ if (!text || !g_ws_running || !g_ws_send_queue) return -1;
- // 由于新的数据结构不支持直接文本消息，这里暂时返回错误
- // 如果需要支持文本消息，需要重新设计消息类型
- HAISI_WS_WARN("Text message not supported in new queue structure\n");
- return -1;
+ size_t text_len = strlen(text);
+ ws_msg_info_t *msg = ws_msg_create(text_len);
+ if (msg == NULL) {
+ HAISI_WS_ERROR("Failed to create send message\n");
+ return -1;
+ }
+
+ msg->msg_type = WS_MSG_TYPE_SEND_TEXT;
+ msg->timestamp = 0;
+ msg->isBinary = false;
+ memcpy(msg->raw_data, text, text_len);
+ msg->raw_data[text_len] = '\0';
+ msg->data_len = text_len;
+
+ osStatus_t ret = osMessageQueuePut(g_ws_send_queue, &msg, 0, 0);
+ if (ret != osOK) {
+ HAISI_WS_ERROR("Failed to put send message to queue: %d\n", ret);
+ ws_msg_free(msg);
+ return -1;
+ }
+
+ HAISI_WS_DEBUG("Text message queued for sending: %s\n", text);
+ return 0;
+#else
+ // 发送队列未启用，直接调用同步发送函数
+ HAISI_WS_WARN("Send queue disabled, calling uapi_ws_send_text directly\n");
+ return uapi_ws_send_text(text);
+#endif
}
/* 强制重连 */
int uapi_ws_stop(void)
{
g_ws_running = 0;
+
+#if WS_USE_SEND_QUEUE
+ // 发送停止消息到发送队列
+ if (g_ws_send_queue != NULL) {
+ ws_msg_info_t *stop_msg = ws_msg_create(0);
+ if (stop_msg != NULL) {
+ stop_msg->msg_type = WS_MSG_TYPE_STOP;
+ stop_msg->timestamp = 0;
+ stop_msg->isBinary = false;
+ osMessageQueuePut(g_ws_send_queue, &stop_msg, 0, 0);
+ }
+ }
+#endif
+
if (g_ws_main_thread) {
osThreadJoin(g_ws_main_thread);
g_ws_main_thread = 0;
}
+
+#if WS_USE_SEND_QUEUE
+ if (g_ws_send_thread) {
+ osThreadJoin(g_ws_send_thread);
+ g_ws_send_thread = 0;
+ }
+#endif
int ret = 0;
if (g_ws_client_fd >= 0) {
ret = uapi_ws_close_with_status(g_ws_client_fd);
g_ws_msg_queue = NULL;
}
+#if WS_USE_SEND_QUEUE
+ // 删除发送队列（先清理所有消息）
+ if (g_ws_send_queue != NULL) {
+ ws_msg_info_t *msg;
+ while (osMessageQueueGet(g_ws_send_queue, &msg, NULL, 0) == osOK) {
+ if (msg != NULL) {
+ HAISI_WS_DEBUG("Clearing queued send message before stop\n");
+ ws_msg_free(msg);
+ }
+ }
+ osMessageQueueDelete(g_ws_send_queue);
+ g_ws_send_queue = NULL;
+ }
+#endif
+
return ret;
}
projects / mirrors / FD230.git / commitdiff