修改ws_message_queue_t 使用osa_msg_queue

diff --git a/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h b/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h
index fc0c3ac..69cbde0 100644 (file)
--- a/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h
+++ b/code/sdk/drivers/peripheral/slip/cat1Service/bk_websocket.h
@@ -32,12 +32,15 @@ typedef struct {
 } ws_message_item_t;
 
 typedef struct {
-    ws_message_item_t messages[WS_QUEUE_SIZE];
-    uint8_t head;
-    uint8_t tail;
-    uint8_t count;
-    osMutexId_t mutex;
-    osSemaphoreId_t semaphore;
+#ifdef SUPPORT_WEBSOCKET_TEST
+    // ws_message_item_t messages[WS_QUEUE_SIZE];
+    // uint8_t head;
+    // uint8_t tail;
+    // uint8_t count;
+    // osMutexId_t mutex;
+    // osSemaphoreId_t semaphore;
+#endif
+    unsigned long queue_id;  // osal消息队列ID
     bool initialized;
 } ws_message_queue_t;
 

diff --git a/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ai_server.cpp b/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ai_server.cpp
index 3603915..5cc2543 100644 (file)
--- a/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ai_server.cpp
+++ b/code/sdk/drivers/peripheral/slip/cat1Service/cat1_ai_server.cpp
@@ -377,22 +377,10 @@ static int ws_queue_init(ws_message_queue_t* queue)
     
     memset(queue, 0, sizeof(ws_message_queue_t));
     
-    /* 创建互斥锁 */
-    queue->mutex = osMutexNew(NULL);
-    if (!queue->mutex) {
-        WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to create mutex for ws queue");
-        return -1;
-    }
-    
-    /* 创建信号量 */
-    queue->semaphore = osSemaphoreNew(WS_QUEUE_SIZE, 0, NULL);
-    if (!queue->semaphore) {
-        WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to create semaphore for ws queue (size=%d)", WS_QUEUE_SIZE);
-        
-    
-        
-        osMutexDelete(queue->mutex);
-        queue->mutex = NULL;
+    /* 创建osal消息队列 */
+    int ret = osal_msg_queue_create("ws_queue", WS_QUEUE_SIZE, &queue->queue_id, 0, sizeof(ws_message_item_t));
+    if (ret != OSAL_SUCCESS) {
+        WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to create osal message queue for ws queue");
         return -1;
     }
     
@@ -408,14 +396,9 @@ static void ws_queue_deinit(ws_message_queue_t* queue)
 {
     if (!queue || !queue->initialized) return;
     
-    if (queue->mutex) {
-        osMutexDelete(queue->mutex);
-        queue->mutex = NULL;
-    }
-    
-    if (queue->semaphore) {
-        osSemaphoreDelete(queue->semaphore);
-        queue->semaphore = NULL;
+    if (queue->queue_id != 0) {
+        osal_msg_queue_delete(queue->queue_id);
+        queue->queue_id = 0;
     }
     
     memset(queue, 0, sizeof(ws_message_queue_t));
@@ -431,46 +414,27 @@ static int ws_queue_put(ws_message_queue_t* queue, const uint8_t* data, uint16_t
         return -1;
     }
     
-    /* 获取互斥锁 */
-    osStatus_t status = osMutexAcquire(queue->mutex, 1000);
-    if (status != osOK) {
-        WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to acquire mutex for ws queue put");
+    /* 创建消息项 */
+    ws_message_item_t item;
+    item.data = (uint8_t*)malloc(length);
+    if (item.data == NULL) {
+        WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to malloc message data");
         return -1;
     }
     
-    /* 检查队列是否已满 */
-    if (queue->count >= WS_QUEUE_SIZE) {
-        osMutexRelease(queue->mutex);
-        WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "WebSocket queue is full");
-        return -1;
-    }
+    memcpy(item.data, data, length);
+    item.length = length;
+    item.type = type;
+    item.timestamp = osKernelGetTickCount();
+    item.in_use = true;
     
-    /* 添加消息到队列 */
-    ws_message_item_t* item = &queue->messages[queue->tail];
-    if (item->data==NULL)
-    {
-        item->data=(uint8_t*)malloc(WS_MESSAGE_MAX_SIZE);
-        if (item->data==NULL)
-        {
-            osMutexRelease(queue->mutex);
-            WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to malloc ws_message_item_t");
-            return -1;
-        }
+    /* 将消息写入osal消息队列 */
+    int ret = osal_msg_queue_write_copy(queue->queue_id, &item, sizeof(ws_message_item_t), 0);
+    if (ret != OSAL_SUCCESS) {
+        WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to write message to osal queue");
+        free(item.data);
+        return -1;
     }
-    memset(item->data,0,WS_MESSAGE_MAX_SIZE);
-    memcpy(item->data, data, length);
-    item->length = length;
-    item->type = type;
-    item->timestamp = osKernelGetTickCount();
-    item->in_use = true;
-    
-    queue->tail = (queue->tail + 1) % WS_QUEUE_SIZE;
-    queue->count++;
-    
-    osMutexRelease(queue->mutex);
-    
-    /* 通知有新消息 */
-    osSemaphoreRelease(queue->semaphore);
     
     return 0;
 }
@@ -484,46 +448,26 @@ static int ws_queue_get(ws_message_queue_t* queue, uint8_t* data, uint16_t max_l
         return -1;
     }
     
-    /* 等待信号量 */
-    osStatus_t status = osSemaphoreAcquire(queue->semaphore, timeout_ms);
-    if (status != osOK) {
+    /* 从osal消息队列读取消息 */
+    ws_message_item_t item;
+    unsigned int buffer_size = sizeof(ws_message_item_t);
+    int ret = osal_msg_queue_read_copy(queue->queue_id, &item, &buffer_size, timeout_ms);
+    if (ret != OSAL_SUCCESS) {
         return 0; /* 超时或错误 */
     }
     
-    /* 获取互斥锁 */
-    status = osMutexAcquire(queue->mutex, 1000);
-    if (status != osOK) {
-        WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to acquire mutex for ws queue get");
-        return -1;
-    }
-    
-    /* 检查队列是否为空 */
-    if (queue->count == 0) {
-        osMutexRelease(queue->mutex);
+    /* 检查消息是否有效 */
+    if (!item.in_use || item.data == NULL) {
         return 0;
     }
     
-    /* 获取消息 */
-    ws_message_item_t* item = &queue->messages[queue->head];
-    if (!item->in_use) {
-        osMutexRelease(queue->mutex);
-        return 0;
-    }
+    /* 复制数据 */
+    uint16_t copy_len = (item.length > max_length) ? max_length : item.length;
+    memcpy(data, item.data, copy_len);
+    *type = item.type;
     
-    uint16_t copy_len = (item->length > max_length) ? max_length : item->length;
-    memcpy(data, item->data, copy_len);
-    *type = item->type;
-    if (item->data)  
-    {
-        free(item->data);
-        item->data=NULL;
-    }
-    
-    /* 标记为已使用 */
-    item->in_use = false;
-    queue->head = (queue->head + 1) % WS_QUEUE_SIZE;
-    queue->count--;
-    osMutexRelease(queue->mutex);
+    /* 释放消息数据 */
+    free(item.data);
     
     return copy_len;
 }
@@ -583,46 +527,16 @@ static int ws_queue_cleanup(ws_message_queue_t* queue) {
     
     WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "开始清理WebSocket消息队列...");
     
-    // 获取互斥锁
-    osStatus_t status = osMutexAcquire(queue->mutex, 1000);
-    if (status != osOK) {
-        WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "ws_queue_cleanup: Failed to acquire mutex, status=%d", status);
-        return -1;
-    }
-    
-    // 清理队列中的所有消息
-    for (int i = 0; i < WS_QUEUE_SIZE; i++) {
-        ws_message_item_t* item = &queue->messages[i];
-        
-        if (item->in_use) {
-            WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "清理队列项 %d: type=%d, length=%d", 
-                          i, item->type, item->length);
-            
-            // 释放动态分配的数据缓冲区
-            if (item->data != NULL) {
-                free(item->data);
-                item->data = NULL;
-                WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "释放队列项 %d 的数据缓冲区", i);
-            }
-            
-            // 重置队列项状态
-            item->length = 0;
-            item->type = (WebSocketMessageType)0;
-            item->timestamp = 0;
-            item->in_use = false;
-            osSemaphoreAcquire(queue->semaphore, 0);
-
+    // 清空osal消息队列中的所有消息
+    ws_message_item_t item;
+    unsigned int buffer_size = sizeof(ws_message_item_t);
+    while (osal_msg_queue_read_copy(queue->queue_id, &item, &buffer_size, 0) == OSAL_SUCCESS) {
+        if (item.data != NULL) {
+            free(item.data);
+            WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "清理队列消息: type=%d, length=%d", item.type, item.length);
         }
     }
     
-    // 重置队列状态
-    queue->head = 0;
-    queue->tail = 0;
-    queue->count = 0;
-    
-    // 释放互斥锁
-    osMutexRelease(queue->mutex);
-    
     WEARABLE_LOGI(WEARABLE_LOG_MODULE_APP, "WebSocket消息队列清理完成");
     return 0;
 }

diff --git a/code/sdk/drivers/peripheral/slip/websocket_test.c b/code/sdk/drivers/peripheral/slip/websocket_test.c
index d547376..da78c7f 100644 (file)
--- a/code/sdk/drivers/peripheral/slip/websocket_test.c
+++ b/code/sdk/drivers/peripheral/slip/websocket_test.c
@@ -22,6 +22,7 @@ extern "C" {
 #include "custom_runtime.h"
 #include "wearable_log.h"
 #include "../tcpmanager/tcp_simple_manager.h"
+#include "osal_msgqueue.h"
 
 // 测试配置
 #define TEST_DATA_SIZE 228
@@ -87,14 +88,10 @@ static int ws_queue_init(ws_message_queue_t* queue)
     
     memset(queue, 0, sizeof(ws_message_queue_t));
     
-    queue->mutex = osMutexNew(NULL);
-    if (!queue->mutex) {
-        return -1;
-    }
-    
-    queue->semaphore = osSemaphoreNew(WS_QUEUE_SIZE, 0, NULL);
-    if (!queue->semaphore) {
-        osMutexDelete(queue->mutex);
+    /* 创建osal消息队列 */
+    int ret = osal_msg_queue_create("ws_test_queue", WS_QUEUE_SIZE, &queue->queue_id, 0, sizeof(ws_message_item_t));
+    if (ret != OSAL_SUCCESS) {
+        WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to create osal message queue for ws test queue");
         return -1;
     }
     
@@ -109,14 +106,9 @@ static void ws_queue_deinit(ws_message_queue_t* queue)
 {
     if (!queue || !queue->initialized) return;
     
-    if (queue->semaphore) {
-        osSemaphoreDelete(queue->semaphore);
-        queue->semaphore = NULL;
-    }
-    
-    if (queue->mutex) {
-        osMutexDelete(queue->mutex);
-        queue->mutex = NULL;
+    if (queue->queue_id != 0) {
+        osal_msg_queue_delete(queue->queue_id);
+        queue->queue_id = 0;
     }
     
     queue->initialized = false;
@@ -136,29 +128,28 @@ static int ws_queue_put(ws_message_queue_t* queue, const uint8_t* data, uint16_t
         return -1;
     }
     
-    if (osMutexAcquire(queue->mutex, osWaitForever) != osOK) {
+    /* 创建消息项 */
+    ws_message_item_t item;
+    item.data = (uint8_t*)malloc(length);
+    if (item.data == NULL) {
+        WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to malloc message data");
         return -1;
     }
     
-    if (queue->count >= WS_QUEUE_SIZE) {
-        osMutexRelease(queue->mutex);
-        WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Queue full");
+    memcpy(item.data, data, length);
+    item.length = length;
+    item.type = type;
+    item.timestamp = (uint32_t)osKernelGetTickCount();
+    item.in_use = true;
+    
+    /* 将消息写入osal消息队列 */
+    int ret = osal_msg_queue_write_copy(queue->queue_id, &item, sizeof(ws_message_item_t), 0);
+    if (ret != OSAL_SUCCESS) {
+        WEARABLE_LOGE(WEARABLE_LOG_MODULE_APP, "Failed to write message to osal queue");
+        free(item.data);
         return -1;
     }
     
-    ws_message_item_t* item = &queue->messages[queue->tail];
-    memcpy(item->data, data, length);
-    item->length = length;
-    item->type = type;
-    item->timestamp = (uint32_t)osKernelGetTickCount();
-    item->in_use = true;
-    
-    queue->tail = (queue->tail + 1) % WS_QUEUE_SIZE;
-    queue->count++;
-    
-    osMutexRelease(queue->mutex);
-    osSemaphoreRelease(queue->semaphore);
-    
     return 0;
 }
 
@@ -171,30 +162,26 @@ static int ws_queue_get(ws_message_queue_t* queue, uint8_t* data, uint16_t max_l
         return -1;
     }
     
-    if (osSemaphoreAcquire(queue->semaphore, timeout_ms) != osOK) {
-        return 0; // 超时
+    /* 从osal消息队列读取消息 */
+    ws_message_item_t item;
+    unsigned int buffer_size = sizeof(ws_message_item_t);
+    int ret = osal_msg_queue_read_copy(queue->queue_id, &item, &buffer_size, timeout_ms);
+    if (ret != OSAL_SUCCESS) {
+        return 0; /* 超时或错误 */
     }
     
-    if (osMutexAcquire(queue->mutex, osWaitForever) != osOK) {
-        return -1;
-    }
-    
-    if (queue->count == 0) {
-        osMutexRelease(queue->mutex);
+    /* 检查消息是否有效 */
+    if (!item.in_use || item.data == NULL) {
         return 0;
     }
     
-    ws_message_item_t* item = &queue->messages[queue->head];
-    uint16_t copy_length = (item->length > max_length) ? max_length : item->length;
-    
-    memcpy(data, item->data, copy_length);
-    *type = item->type;
-    
-    item->in_use = false;
-    queue->head = (queue->head + 1) % WS_QUEUE_SIZE;
-    queue->count--;
+    /* 复制数据 */
+    uint16_t copy_length = (item.length > max_length) ? max_length : item.length;
+    memcpy(data, item.data, copy_length);
+    *type = item.type;
     
-    osMutexRelease(queue->mutex);
+    /* 释放消息数据 */
+    free(item.data);
     
     return copy_length;
 }