init add msg manager

app/app_main.c

@@ -31,6 +31,7 @@
 #include "ringbuffer.h"
 #include "av_ringbuffer.h"
 #include "xbuf.h"
+#include "has_task_msg.h"
 
 #include "mw_tcpserver.h"
 

mw/has_task_msg_manager/test_msg/has_task_msg.h

@@ -0,0 +1,50 @@
+#ifndef HAS_TASK_MSG_H
+#define HAS_TASK_MSG_H
+
+
+/* 静态订阅表 */
+// TODO：后续可支持动态增加模块订阅，感觉目前没必要
+
+typedef enum {
+    INVALID_ID = 0,
+    GUI,
+    SENSOR,
+    ACM,
+    WIFI,
+    VOICE,
+    MODULE_MAX
+} has_module_ID_e;
+
+/* 模块 订阅数量 订阅模块 */
+/* 模块 和 订阅数量一定要写，完全不加入表中的模块，或订阅数量为0的模块无法接收消息 */
+#define SUBSCIBE_INFO \
+{ \
+    {GUI, 2, {ACM, VOICE}}, \
+    {SENSOR, 1, {ACM}}, \
+    {ACM, 1, {WIFI}}, \
+    {WIFI, 0}, \
+    {VOICE, 3, {ACM, VOICE, GUI}}, \
+}
+
+/* 消息处理回调 */
+typedef int (*has_msg_handle_cb)(unsigned char module_id, const unsigned char *buf, unsigned int len);
+
+/* 基础功能 */
+int has_msg_init(void);
+int has_msg_publish(has_module_ID_e module_id, void *buffer, unsigned int length);
+int has_msg_handle(has_module_ID_e module_id, has_msg_handle_cb cb);
+
+unsigned int has_msg_is_message_empty(has_module_ID_e module_id);
+unsigned int has_msg_get_message_number(has_module_ID_e module_id);
+int has_msg_delete_all_message(has_module_ID_e module_id);
+
+/* 扩展功能 */
+int has_msg_handle_by_module(has_module_ID_e module_id, has_msg_handle_cb cb, has_module_ID_e pub_module_id);
+int has_msg_handle_latest(has_module_ID_e module_id, has_msg_handle_cb cb);
+unsigned int has_msg_receive(has_module_ID_e module_id, unsigned char *pub_module_id, unsigned char *buf_out);
+unsigned int has_msg_receive_latest(has_module_ID_e module_id, unsigned char *pub_module_id, unsigned char *buf_out);
+int has_msg_printf_subscribe(has_module_ID_e module_id);
+// int has_msg_subscribe(has_module_ID_e module_id, has_module_ID_e sub_id);
+// int has_msg_unsubscribe(has_module_ID_e module_id, has_module_ID_e sub_id);
+
+#endif

mw/has_task_msg_manager/test_msg/has_task_msg_mamanger.c

@@ -0,0 +1,581 @@
+#include <string.h>
+#include <stdio.h>
+#include <stdint.h>
+#include "list.h"
+#include "has_task_msg.h"
+#include "has_task_msg_os_port.h"
+
+// #define MSG_OPT_DYNAMIC_SUB // 允许动态订阅
+// #define MSG_OPT_DEBUG // 调试
+#define MSG_OPT_PUB_LIST // 使用静态发布表（多条消息单次分配内存）
+
+/* 暂时不支持发布列表和动态订阅同时使用，后续需要再更新支持 */
+#if defined(MSG_OPT_DYNAMIC_SUB)  && defined(MSG_OPT_PUB_LIST)
+#error "These two functions are not currently supported to be used together."
+#endif
+
+#ifdef MSG_OPT_DEBUG
+/* 打开检查订阅表是否配置正确 */
+#define MSG_OPT_CHECK_SUBSCRIBE_LIST
+#endif
+
+
+#pragma pack(1)
+#ifdef MSG_OPT_PUB_LIST
+/* 消息数据 */
+typedef struct {
+    unsigned char module_id; // 发布的模块ID
+    unsigned int node_cnt; // 被引用节点数
+    unsigned int length;
+    unsigned char buffer[];
+} has_msg_buffer_t;
+/* 消息节点 */
+typedef struct {
+    struct list_head list; // 消息链表
+    has_msg_buffer_t *msg_buff;
+} has_msg_node_t;
+#else
+/* 消息节点和数据 */
+typedef struct {
+    unsigned char module_id; // 发布的模块ID
+    struct list_head list; // 消息链表
+    unsigned int length;
+    unsigned char buffer[];
+} has_msg_node_t;
+#endif
+#pragma pack()
+
+#ifdef MSG_OPT_PUB_LIST
+/* 静态发布列表 */
+typedef struct {
+    unsigned char pub_module[MODULE_MAX - 1]; // 发布表
+#ifdef MSG_OPT_USE_MUTEX
+    msg_mutex_t buffer_mutex; // 消息互斥量
+#endif
+    unsigned char pub_num;
+} has_static_pub_list_t;
+static has_static_pub_list_t g_pub_list[MODULE_MAX] = {0};
+#endif
+
+/* 静态订阅列表 */
+typedef struct {
+#if defined(MSG_OPT_CHECK_SUBSCRIBE_LIST) || defined(MSG_OPT_DYNAMIC_SUB)
+    unsigned char module_id; // 当前模块ID
+    unsigned char sub_module_cnt; // 订阅数量
+    unsigned char sub_module[MODULE_MAX - 1]; // 订阅表
+#else
+    const unsigned char module_id; // 当前模块ID
+    const unsigned char sub_module_cnt; // 订阅数量
+    const unsigned char sub_module[MODULE_MAX - 1]; // 订阅表
+#endif
+#ifdef MSG_OPT_USE_MUTEX
+    msg_mutex_t msg_mutex; // 消息互斥量
+#endif
+    struct list_head msg_list; // 消息链表
+} has_static_sub_list_t;
+static has_static_sub_list_t g_sub_list[] = SUBSCIBE_INFO;
+static unsigned char g_sub_list_index[MODULE_MAX] = {0}; // 订阅表id索引
+
+#ifdef MSG_OPT_CHECK_SUBSCRIBE_LIST
+#define MSG_CHECK_SUBSCIBE_INFO(id, ret) \
+if ((id) == INVALID_ID) { \
+    msg_printf("%s invalid id:0, please check SUBSCIBE_INFO\n", __func__); \
+    return (ret);}
+#else
+#define MSG_CHECK_SUBSCIBE_INFO(id, ret)
+#endif
+
+/* 检查ID是否合法，id:检查的ID，ret:不合法时return的结果 */
+#define MSG_CHECK_MODULE_ID(id, ret) \
+if (((id) >= MODULE_MAX) || ((id) == INVALID_ID)) { \
+    msg_printf("%s ID wrong:%d\n", __func__, (id)); return (ret); }
+
+int has_msg_init() {
+    int j;
+    memset(g_sub_list_index, 0, sizeof(g_sub_list_index));
+#ifdef MSG_OPT_PUB_LIST
+    memset(g_pub_list, 0, sizeof(g_pub_list));
+#endif
+    for (int i = 0; i < (sizeof(g_sub_list) / sizeof(g_sub_list[0])); i++) {
+#ifdef MSG_OPT_CHECK_SUBSCRIBE_LIST
+        /* 检查各模块ID是否配置正确 */
+        if ((g_sub_list[i].module_id >= MODULE_MAX) || (g_sub_list[i].module_id == INVALID_ID)) {
+            msg_printf("module_id:%d is invalid, please check the macro:SUBSCIBE_INFO row:%d\n", g_sub_list[i].module_id, i);
+            memset(g_sub_list, 0, sizeof(g_sub_list));
+#ifdef MSG_OPT_PUB_LIST
+            memset(g_pub_list, 0, sizeof(g_pub_list));
+#endif
+            return -1;
+        }
+
+        /* 检查订阅表ID是否配置正确 */
+        for (j = 0; j < (MODULE_MAX - 1); j++) {
+            if (g_sub_list[i].sub_module[j] == INVALID_ID) {
+                break;
+            } else if (g_sub_list[i].sub_module[j] >= MODULE_MAX) {
+                msg_printf("sub_module id:%d is invalid, module:id %d\n please check the macro:\
+                    SUBSCIBE_INFO row:%d, sub list:%d\n", g_sub_list[i].sub_module[j], g_sub_list[i].module_id, i, j);
+                memset(g_sub_list, 0, sizeof(g_sub_list));
+#ifdef MSG_OPT_PUB_LIST
+                memset(g_pub_list, 0, sizeof(g_pub_list));
+#endif
+                return -1;
+            }
+        }
+        /* 检查订阅数量是否配置正确 */
+        if (g_sub_list[i].sub_module_cnt != j) {
+            msg_printf("sub_module_cnt is config as %d, which actually is %d, please check the\
+                macro: SUBSCIBE_INFO row:%d\n", g_sub_list[i].sub_module_cnt, j, i);
+            memset(g_sub_list, 0, sizeof(g_sub_list));
+#ifdef MSG_OPT_PUB_LIST
+            memset(g_pub_list, 0, sizeof(g_pub_list));
+#endif
+            return -1;
+        }
+#endif
+        g_sub_list_index[g_sub_list[i].module_id] = i; // 索引表赋值
+#ifdef MSG_OPT_PUB_LIST
+        /* 构建发布表 */
+        for (int k = 0; k < g_sub_list[i].sub_module_cnt; k++) {
+            if (g_pub_list[g_sub_list[i].sub_module[k]].pub_num == 0) {
+                msg_mutex_init(&g_pub_list[g_sub_list[i].sub_module[k]].buffer_mutex);
+            }
+
+            g_pub_list[g_sub_list[i].sub_module[k]].pub_module[g_pub_list[g_sub_list[i].sub_module[k]].pub_num] = g_sub_list[i].module_id;
+            g_pub_list[g_sub_list[i].sub_module[k]].pub_num ++;
+        }
+#endif
+        msg_mutex_init(&g_sub_list[i].msg_mutex); // 初始化该模块互斥量
+        INIT_LIST_HEAD(&g_sub_list[i].msg_list); // 初始化订阅链表
+    }
+    return 0;
+}
+
+
+int has_msg_publish(has_module_ID_e module_id, void *buffer, unsigned int length) {
+    has_msg_node_t *node;
+    if (buffer == NULL) {
+        return -1;
+    }
+    // if (length == 0) {
+    //     return -1;
+    // }
+#ifdef MSG_OPT_PUB_LIST
+    has_msg_buffer_t *msg_buffer;
+    unsigned char sub_index;
+
+    MSG_CHECK_MODULE_ID(module_id, -1);
+    if (g_pub_list[module_id].pub_num == 0) {
+        // msg_printf("no one subscribe you:%d\n", module_id);
+        return 0;
+    }
+
+    /* 根据订阅表分配内存 */
+    node = (has_msg_node_t *)msg_malloc((sizeof(has_msg_node_t) * g_pub_list[module_id].pub_num) + sizeof(has_msg_buffer_t) + length);
+    if (node == NULL) {
+        msg_printf("malloc fail,pub id:%d\n", module_id);
+        return -1;
+    }
+    msg_buffer = (has_msg_buffer_t *)(&node[g_pub_list[module_id].pub_num]);
+    msg_buffer->node_cnt = g_pub_list[module_id].pub_num;
+    msg_buffer->length = length;
+    msg_buffer->module_id = module_id;
+    memcpy(msg_buffer->buffer, buffer, length);
+    /* 查订阅表添加消息 */
+    for (int i = 0; i < g_pub_list[module_id].pub_num; i++) {
+        sub_index = g_sub_list_index[g_pub_list[module_id].pub_module[i]];
+        node[i].msg_buff = msg_buffer;
+        msg_mutex_lock(&g_sub_list[sub_index].msg_mutex);
+        list_add_tail(&node[i].list, &g_sub_list[sub_index].msg_list); // 加入消息链表
+        msg_mutex_unlock(&g_sub_list[sub_index].msg_mutex);
+    }
+#else
+    int i;
+
+    for (i = 0; i < (sizeof(g_sub_list) / sizeof(g_sub_list[0])); i++) { // 遍历模块
+#ifdef MSG_OPT_DYNAMIC_SUB
+        msg_mutex_lock(&g_sub_list[i].msg_mutex);
+#endif
+        for (int j = 0; j < g_sub_list[i].sub_module_cnt; j++) { // 查找订阅表
+            if (module_id == g_sub_list[i].sub_module[j]) { // 找到订阅信息
+                node = msg_malloc(sizeof(has_msg_node_t) + length);
+                if (node == NULL) {
+                    msg_printf("malloc fail,pub id:%d\n", module_id);
+#ifdef MSG_OPT_DYNAMIC_SUB
+                    msg_mutex_unlock(&g_sub_list[i].msg_mutex);
+#endif
+                    return -1;
+                }
+                node->length = length;
+                node->module_id = module_id;
+                memcpy(node->buffer, buffer, length);
+#ifndef MSG_OPT_DYNAMIC_SUB
+                msg_mutex_lock(&g_sub_list[i].msg_mutex);
+#endif
+                list_add_tail(&node->list, &g_sub_list[i].msg_list); // 加入消息链表
+#ifndef MSG_OPT_DYNAMIC_SUB
+                msg_mutex_unlock(&g_sub_list[i].msg_mutex);
+#endif
+
+                break; // 不可重复订阅
+            }
+        }
+#ifdef MSG_OPT_DYNAMIC_SUB
+        msg_mutex_unlock(&g_sub_list[i].msg_mutex);
+#endif
+    }
+#endif
+    return 0;
+}
+
+/* TODO:更合理的方式是每个消息一个互斥量，但是可能会导致每个消息malloc和free的时候频繁init和deinit */
+#ifdef MSG_OPT_PUB_LIST
+/* 检查计数判断该消息是否被所有订阅模块处理完，处理完就释放内存
+    msg_buff_stu：消息结构体指针 np:节点结构体指针 */
+#define MSG_CHECK_AND_FREE_MESSAGE(msg_buff_stu, np) \
+{ \
+    msg_mutex_lock(&g_pub_list[msg_buff_stu->module_id].buffer_mutex); \
+    msg_buff_stu->node_cnt --; \
+    if (msg_buff_stu->node_cnt == 0) { \
+    np = (has_msg_node_t *)((unsigned char *)msg_buff_stu \
+    - g_pub_list[msg_buff_stu->module_id].pub_num * sizeof(has_msg_node_t)); \
+    msg_free(np);} \
+    msg_mutex_unlock(&g_pub_list[msg_buff_stu->module_id].buffer_mutex); \
+}
+#endif
+
+int has_msg_handle(has_module_ID_e module_id, has_msg_handle_cb cb) {
+    has_msg_node_t *node;
+    has_static_sub_list_t *sub_list;
+    MSG_CHECK_MODULE_ID(module_id, -1);
+
+    sub_list = &g_sub_list[g_sub_list_index[module_id]];
+    MSG_CHECK_SUBSCIBE_INFO(sub_list->module_id, -1);
+
+    /* 处理一个消息，先入先出 */
+    msg_mutex_lock(&sub_list->msg_mutex);
+    if (!list_empty(&sub_list->msg_list)) {
+        node = list_first_entry(&sub_list->msg_list, has_msg_node_t, list);
+        list_del(&node->list);
+        msg_mutex_unlock(&sub_list->msg_mutex);
+    } else { // 没有消息需要处理
+        msg_mutex_unlock(&sub_list->msg_mutex);
+        return -1;
+    }
+
+#ifdef MSG_OPT_PUB_LIST
+    has_msg_buffer_t *msg_buffer;
+    msg_buffer = node->msg_buff;
+    if (cb != NULL) {
+        cb(msg_buffer->module_id, msg_buffer->buffer, msg_buffer->length);
+    }
+    MSG_CHECK_AND_FREE_MESSAGE(msg_buffer, node);
+#else
+    if (cb != NULL) {
+        cb(node->module_id, node->buffer, node->length);
+    }
+#endif
+
+    msg_free(node);
+
+    return 0;
+}
+
+int has_msg_handle_by_module(has_module_ID_e module_id, has_msg_handle_cb cb, has_module_ID_e pub_module_id) {
+    has_msg_node_t *node, *temp;
+    has_static_sub_list_t *sub_list;
+    MSG_CHECK_MODULE_ID(module_id, -1);
+    MSG_CHECK_MODULE_ID(pub_module_id, -1);
+
+    sub_list = &g_sub_list[g_sub_list_index[module_id]];
+    MSG_CHECK_SUBSCIBE_INFO(sub_list->module_id, -1);
+
+    /* 处理一个指定模块的消息，先入先出 */
+    msg_mutex_lock(&sub_list->msg_mutex);
+    if (!list_empty(&sub_list->msg_list)) {
+        list_for_each_entry_safe(node, temp, &sub_list->msg_list, list) {
+#ifdef MSG_OPT_PUB_LIST
+            if (node->msg_buff->module_id == pub_module_id) {
+#else
+            if (node->module_id == pub_module_id) {
+#endif
+                list_del(&node->list);
+                break;
+            }
+        }
+        msg_mutex_unlock(&sub_list->msg_mutex);
+    } else { // 没有消息需要处理
+        msg_mutex_unlock(&sub_list->msg_mutex);
+        return -1;
+    }
+
+#ifdef MSG_OPT_PUB_LIST
+    has_msg_buffer_t *msg_buffer;
+    msg_buffer = node->msg_buff;
+    if (cb != NULL) {
+        cb(msg_buffer->module_id, msg_buffer->buffer, msg_buffer->length);
+    }
+    MSG_CHECK_AND_FREE_MESSAGE(msg_buffer, node);
+#else
+    if (cb != NULL) {
+        cb(node->module_id, node->buffer, node->length);
+    }
+#endif
+
+    msg_free(node);
+
+    return 0;
+}
+
+int has_msg_handle_latest(has_module_ID_e module_id, has_msg_handle_cb cb) {
+    has_msg_node_t *node;
+    has_static_sub_list_t *sub_list;
+    MSG_CHECK_MODULE_ID(module_id, -1);
+
+    sub_list = &g_sub_list[g_sub_list_index[module_id]];
+    MSG_CHECK_SUBSCIBE_INFO(sub_list->module_id, -1);
+
+    /* 处理最新的消息 */
+    msg_mutex_lock(&sub_list->msg_mutex);
+    if (!list_empty(&sub_list->msg_list)) {
+        node = list_last_entry(&sub_list->msg_list, has_msg_node_t, list);
+        list_del(&node->list);
+        msg_mutex_unlock(&sub_list->msg_mutex);
+    } else { // 没有消息需要处理
+        msg_mutex_unlock(&sub_list->msg_mutex);
+        return -1;
+    }
+
+#ifdef MSG_OPT_PUB_LIST
+    has_msg_buffer_t *msg_buffer;
+    msg_buffer = node->msg_buff;
+    if (cb != NULL) {
+        cb(msg_buffer->module_id, msg_buffer->buffer, msg_buffer->length);
+    }
+    MSG_CHECK_AND_FREE_MESSAGE(msg_buffer, node);
+#else
+    if (cb != NULL) {
+        cb(node->module_id, node->buffer, node->length);
+    }
+#endif
+
+    msg_free(node);
+
+    return 0;
+}
+
+/* 将消息复制出接口处理，不建议使用 */
+#if 1
+unsigned int has_msg_receive(has_module_ID_e module_id, unsigned char *pub_module_id, unsigned char *buf_out) {
+    has_msg_node_t *node;
+    has_static_sub_list_t *sub_list;
+    unsigned int len = 0;
+    if ((module_id >= MODULE_MAX) || (module_id == INVALID_ID) || (buf_out == NULL)) {
+        msg_printf("%s id of buff wrong:%d\n", __func__, module_id);
+        return 0;
+    }
+
+    sub_list = &g_sub_list[g_sub_list_index[module_id]];
+    MSG_CHECK_SUBSCIBE_INFO(sub_list->module_id, 0);
+
+    /* 处理一个消息，先入先出 */
+    msg_mutex_lock(&sub_list->msg_mutex);
+    if (!list_empty(&sub_list->msg_list)) {
+        node = list_first_entry(&sub_list->msg_list, has_msg_node_t, list);
+        list_del(&node->list);
+        msg_mutex_unlock(&sub_list->msg_mutex);
+    } else { // 没有消息需要处理
+        msg_mutex_unlock(&sub_list->msg_mutex);
+        return 0;
+    }
+
+#ifdef MSG_OPT_PUB_LIST
+    has_msg_buffer_t *msg_buffer;
+    msg_buffer = node->msg_buff;
+    memcpy(buf_out, msg_buffer->buffer, msg_buffer->length);
+    if (pub_module_id != NULL) {
+        *pub_module_id = msg_buffer->module_id;
+    }
+    len = msg_buffer->length;
+    MSG_CHECK_AND_FREE_MESSAGE(msg_buffer, node);
+#else
+    memcpy(buf_out, node->buffer, node->length);
+    if (pub_module_id != NULL) {
+        *pub_module_id = node->module_id;
+    }
+
+    len = node->length;
+    msg_free(node);
+#endif
+    return len;
+}
+
+unsigned int has_msg_receive_latest(has_module_ID_e module_id, unsigned char *pub_module_id, unsigned char *buf_out) {
+    has_msg_node_t *node;
+    has_static_sub_list_t *sub_list;
+    unsigned int len = 0;
+    if ((module_id >= MODULE_MAX) || (module_id == INVALID_ID) || (buf_out == NULL)) {
+        msg_printf("%s id of buff wrong:%d\n", __func__, module_id);
+        return 0;
+    }
+
+    sub_list = &g_sub_list[g_sub_list_index[module_id]];
+    MSG_CHECK_SUBSCIBE_INFO(sub_list->module_id, 0);
+
+    /* 处理最新的消息 */
+    msg_mutex_lock(&sub_list->msg_mutex);
+    if (!list_empty(&sub_list->msg_list)) {
+        node = list_last_entry(&sub_list->msg_list, has_msg_node_t, list);
+        list_del(&node->list);
+        msg_mutex_unlock(&sub_list->msg_mutex);
+    } else { // 没有消息需要处理
+        msg_mutex_unlock(&sub_list->msg_mutex);
+        return 0;
+    }
+
+#ifdef MSG_OPT_PUB_LIST
+    has_msg_buffer_t *msg_buffer;
+    msg_buffer = node->msg_buff;
+    memcpy(buf_out, msg_buffer->buffer, msg_buffer->length);
+    if (pub_module_id != NULL) {
+        *pub_module_id = msg_buffer->module_id;
+    }
+    len = msg_buffer->length;
+    MSG_CHECK_AND_FREE_MESSAGE(msg_buffer, node);
+#else
+    memcpy(buf_out, node->buffer, node->length);
+    if (pub_module_id != NULL) {
+        *pub_module_id = node->module_id;
+    }
+
+    len = node->length;
+    msg_free(node);
+#endif
+    return len;
+}
+#endif
+
+unsigned int has_msg_is_message_empty(has_module_ID_e module_id) {
+    has_msg_node_t *node, *temp;
+    has_static_sub_list_t *sub_list;
+    unsigned int num = 0;
+    MSG_CHECK_MODULE_ID(module_id, -1);
+
+    sub_list = &g_sub_list[g_sub_list_index[module_id]];
+    MSG_CHECK_SUBSCIBE_INFO(sub_list->module_id, -1);
+
+    msg_mutex_lock(&sub_list->msg_mutex);
+    if (list_empty(&sub_list->msg_list)) {
+        msg_mutex_unlock(&sub_list->msg_mutex);
+        return 0;
+    } else {
+        msg_mutex_unlock(&sub_list->msg_mutex);
+        return -1;
+    }
+}
+
+unsigned int has_msg_get_message_number(has_module_ID_e module_id) {
+    has_msg_node_t *node, *temp;
+    has_static_sub_list_t *sub_list;
+    unsigned int num = 0;
+    MSG_CHECK_MODULE_ID(module_id, 0);
+
+    sub_list = &g_sub_list[g_sub_list_index[module_id]];
+    MSG_CHECK_SUBSCIBE_INFO(sub_list->module_id, 0);
+
+    msg_mutex_lock(&sub_list->msg_mutex);
+    list_for_each_entry_safe(node, temp, &sub_list->msg_list, list) {
+        num ++;
+    }
+    msg_mutex_unlock(&sub_list->msg_mutex);
+    return num;
+}
+
+int has_msg_printf_subscribe(has_module_ID_e module_id) {
+    has_static_sub_list_t *sub_list;
+    MSG_CHECK_MODULE_ID(module_id, -1);
+
+    sub_list = &g_sub_list[g_sub_list_index[module_id]];
+    MSG_CHECK_SUBSCIBE_INFO(sub_list->module_id, -1);
+
+    msg_printf("module:%d subscribes to:\n", module_id);
+    for (int i = 0; i < sub_list->sub_module_cnt; i++) {
+        msg_printf("%d ", sub_list->sub_module[i]);
+    }
+    return 0;
+}
+
+int has_msg_delete_all_message(has_module_ID_e module_id) {
+    has_msg_node_t *node, *temp;
+    has_static_sub_list_t *sub_list;
+    MSG_CHECK_MODULE_ID(module_id, -1);
+
+    sub_list = &g_sub_list[g_sub_list_index[module_id]];
+    MSG_CHECK_SUBSCIBE_INFO(sub_list->module_id, -1);
+
+    msg_mutex_lock(&sub_list->msg_mutex);
+    if (!list_empty(&sub_list->msg_list)) {
+        list_for_each_entry_safe(node, temp, &sub_list->msg_list, list) {
+            list_del(&node->list);
+            msg_free(node);
+        }
+    }
+    INIT_LIST_HEAD(&sub_list->msg_list);
+    msg_mutex_unlock(&sub_list->msg_mutex);
+    return 0;
+}
+
+#ifdef MSG_OPT_DYNAMIC_SUB
+int has_msg_subscribe(has_module_ID_e module_id, has_module_ID_e sub_id) {
+    has_static_sub_list_t *sub_list;
+    int j;
+    if ((module_id >= MODULE_MAX) || (sub_id >= MODULE_MAX)
+        || (module_id == INVALID_ID) || (sub_id == INVALID_ID)) {
+        return -1;
+    }
+    sub_list = &g_sub_list[g_sub_list_index[module_id]];
+    MSG_CHECK_SUBSCIBE_INFO(sub_list->module_id, -1);
+
+    msg_mutex_lock(&sub_list->msg_mutex);
+    if (sub_list->sub_module_cnt < (MODULE_MAX - 1)) {
+        /* 检查是否已经订阅 */
+        for (j = 0; j < sub_list->sub_module_cnt; j++) {
+            if (sub_id == sub_list->sub_module[j]) {
+                msg_mutex_unlock(&sub_list->msg_mutex);
+                msg_printf("ID:%d has been sub by module:%d\n", sub_id, module_id);
+                return -1;
+            }
+        }
+        /* 加入该模块的订阅表中 */
+        sub_list->sub_module[j] = sub_id;
+        sub_list->sub_module_cnt ++;
+        msg_mutex_unlock(&sub_list->msg_mutex);
+        return 0;
+    }
+    else {
+        msg_mutex_unlock(&sub_list->msg_mutex);
+        msg_printf("module:%d subscribe fail\n", module_id);
+        return -1;
+    }
+}
+
+int has_msg_unsubscribe(has_module_ID_e module_id, has_module_ID_e sub_id) {
+    has_static_sub_list_t *sub_list;
+    if ((module_id >= MODULE_MAX) || (sub_id >= MODULE_MAX)
+        || (module_id == INVALID_ID) || (sub_id == INVALID_ID)) {
+        return -1;
+    }
+    sub_list = &g_sub_list[g_sub_list_index[module_id]];
+    MSG_CHECK_SUBSCIBE_INFO(sub_list->module_id, -1);
+
+    msg_mutex_lock(&sub_list->msg_mutex);
+    for (int j = 0; j < sub_list->sub_module_cnt; j++) {
+        if (sub_id == sub_list->sub_module[j]) {
+            sub_list->sub_module[j] = INVALID_ID;
+            sub_list->sub_module_cnt --;
+        }
+    }
+    msg_mutex_unlock(&sub_list->msg_mutex);
+    return 0;
+}
+#endif

mw/has_task_msg_manager/test_msg/has_task_msg_os_port.h

@@ -0,0 +1,162 @@
+#ifndef HAS_TASK_MSG_OS_PROT_H
+#define HAS_TASK_MSG_OS_PROT_H
+
+#define MSG_OPT_OS_LINUX 1
+#define MSG_OPT_OS_freeRTOS 2
+#define MSG_OPT_OS_RT_THREAD 3
+#define MSG_OPT_OS_NONE_OR_OSAL 4
+#define MSG_OPT_OS_USER_DEFINE 5
+
+/* TODO:rtt和freertos未测过 */
+#define MSG_OPT_OS MSG_OPT_OS_LINUX
+// #define MSG_OPT_OS MSG_OPT_OS_freeRTOS
+// #define MSG_OPT_OS MSG_OPT_OS_RT_THREAD
+// #define MSG_OPT_OS MSG_OPT_OS_NONE_OR_OSAL
+// #define MSG_OPT_OS MSG_OPT_OS_USER_DEFINE
+
+#define MSG_OPT_USE_MUTEX // 支持互斥量
+
+
+/* ----------------------- for Linux ----------------------- */
+#if (MSG_OPT_OS == MSG_OPT_OS_LINUX)
+// #include <malloc.h>
+#include <pthread.h>
+#include <stdlib.h>
+#include <stdio.h>
+
+#ifdef MSG_OPT_USE_MUTEX
+#define msg_mutex_t pthread_mutex_t
+#define msg_mutex_init(m) pthread_mutex_init((msg_mutex_t *)m, NULL)
+#define msg_mutex_lock(m) pthread_mutex_lock((msg_mutex_t *)m)
+#define msg_mutex_unlock(m) pthread_mutex_unlock((msg_mutex_t *)m)
+#else
+#define msg_mutex_t 
+#define msg_mutex_init(m)
+#define msg_mutex_lock(m)
+#define msg_mutex_unlock(m)
+#endif
+
+#define msg_malloc malloc
+#define msg_free free
+#define msg_printf printf
+
+/* ----------------------- for freeRTOS ----------------------- */
+#elif (MSG_OPT_OS == MSG_OPT_OS_freeRTOS)
+#include <stdlib.h>
+#include <stdio.h>
+#include "FreeRTOS.h" 
+#include "semphr.h"
+#ifdef MSG_OPT_USE_MUTEX
+#define msg_mutex_t SemaphoreHandle_t
+static inline void msg_mutex_init(void *mutex) {
+	*mutex = xSemaphoreCreateMutex();
+    if (*mutex == NULL) {
+        printf("create mutex err!!\n\n");
+    }
+}
+
+static inline void msg_mutex_lock(void *mutex) {
+    if (xSemaphoreTake(*mutex, portMAX_DELAY) != pdTRUE) {
+        printf("lock mutex err!!\n\n");
+    }
+}
+
+static inline void msg_mutex_unlock(void *mutex) {
+    xSemaphoreGive(*mutex);
+}
+#else
+#define msg_mutex_t 
+#define msg_mutex_init(m)
+#define msg_mutex_lock(m)
+#define msg_mutex_unlock(m)
+#endif
+
+#define msg_malloc pvPortMalloc
+#define msg_free vPortFree
+#define msg_printf printf
+
+/* ----------------------- for RT-thread ----------------------- */
+#elif (MSG_OPT_OS == MSG_OPT_OS_RT_THREAD)
+#include <stdlib.h>
+#include <stdio.h>
+#include <rtthread.h>
+#ifdef MSG_OPT_USE_MUTEX
+#define msg_mutex_t rt_mutex_t
+static inline void msg_mutex_init(void *mutex) {
+	*mutex = rt_mutex_create("has_msg", RT_IPC_FLAG_FIFO);
+    if (*mutex == RT_NULL) {
+        printf("create mutex err!!\n\n");
+    }
+}
+
+static inline void msg_mutex_lock(void *mutex) {
+    if (rt_mutex_take(*mutex, RT_WAITING_FOREVER) != RT_EOK) {
+        printf("lock mutex err!!\n\n");
+    }
+}
+
+static inline void msg_mutex_unlock(void *mutex) {
+    rt_mutex_release(*mutex);
+}
+#else
+#define msg_mutex_t 
+#define msg_mutex_init(m)
+#define msg_mutex_lock(m)
+#define msg_mutex_unlock(m)
+#endif
+
+#define msg_malloc rt_malloc
+#define msg_free rt_free
+#define msg_printf printf
+
+/* ----------------------- for none os ----------------------- */
+#elif (MSG_OPT_OS == MSG_OPT_OS_NONE_OR_OSAL)
+#include <stdlib.h>
+#include <stdio.h>
+#undef MSG_OPT_USE_MUTEX
+
+#define msg_mutex_t 
+#define msg_mutex_init(m)
+#define msg_mutex_lock(m)
+#define msg_mutex_unlock(m)
+
+#define msg_malloc malloc
+#define msg_free free
+#define msg_printf printf
+
+/* ----------------------- for user define ----------------------- */
+#elif (MSG_OPT_OS == MSG_OPT_OS_USER_DEFINE)
+#include <stdlib.h>
+#include <stdio.h>
+
+#ifdef MSG_OPT_USE_MUTEX
+#define msg_mutex_t int/* to be done */
+static inline void msg_mutex_init(void *mutex) {
+	/* to be done */
+}
+
+static inline void msg_mutex_lock(void *mutex) {
+    /* to be done */
+}
+
+static inline void msg_mutex_unlock(void *mutex) {
+    /* to be done */
+}
+#else
+#define msg_mutex_t 
+#define msg_mutex_init(m)
+#define msg_mutex_lock(m)
+#define msg_mutex_unlock(m)
+#endif
+
+
+#define msg_malloc malloc
+#define msg_free free
+#define msg_printf printf
+
+#else
+#error "!Undefined system environment!"
+#endif
+
+
+#endif

mw/has_task_msg_manager/test_msg/list.h

@@ -0,0 +1,208 @@
+/*-
+ * Copyright (c) 2011 Felix Fietkau <nbd@openwrt.org>
+ * Copyright (c) 2010 Isilon Systems, Inc.
+ * Copyright (c) 2010 iX Systems, Inc.
+ * Copyright (c) 2010 Panasas, Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice unmodified, this list of conditions, and the following
+ *    disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in the
+ *    documentation and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
+ * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
+ * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
+ * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
+ * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
+ * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
+ * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+#ifndef _LINUX_LIST_H_
+#define _LINUX_LIST_H_
+
+#include <stddef.h>
+#include <stdbool.h>
+
+#define	prefetch(x)
+
+#ifndef container_of
+#define container_of(ptr, type, member)					\
+	({								\
+		const typeof(((type *) NULL)->member) *__mptr = (ptr);	\
+		(type *) ((char *) __mptr - offsetof(type, member));	\
+	})
+#endif
+
+struct list_head {
+	struct list_head *next;
+	struct list_head *prev;
+};
+
+#define LIST_HEAD_INIT(name) { &(name), &(name) }
+#undef LIST_HEAD
+#define LIST_HEAD(name)	struct list_head name = LIST_HEAD_INIT(name)
+
+static inline void
+INIT_LIST_HEAD(struct list_head *list)
+{
+	list->next = list->prev = list;
+}
+
+static inline bool
+list_empty(const struct list_head *head)
+{
+	return (head->next == head);
+}
+
+static inline bool
+list_is_first(const struct list_head *list,
+	      const struct list_head *head)
+{
+	return list->prev == head;
+}
+
+static inline bool
+list_is_last(const struct list_head *list,
+	     const struct list_head *head)
+{
+	return list->next == head;
+}
+
+static inline void
+_list_del(struct list_head *entry)
+{
+	entry->next->prev = entry->prev;
+	entry->prev->next = entry->next;
+}
+
+static inline void
+list_del(struct list_head *entry)
+{
+	_list_del(entry);
+	entry->next = entry->prev = NULL;
+}
+
+static inline void
+_list_add(struct list_head *_new, struct list_head *prev,
+    struct list_head *next)
+{
+
+	next->prev = _new;
+	_new->next = next;
+	_new->prev = prev;
+	prev->next = _new;
+}
+
+static inline void
+list_del_init(struct list_head *entry)
+{
+	_list_del(entry);
+	INIT_LIST_HEAD(entry);
+}
+
+#define	list_entry(ptr, type, field)	container_of(ptr, type, field)
+#define	list_first_entry(ptr, type, field)	list_entry((ptr)->next, type, field)
+#define	list_last_entry(ptr, type, field)	list_entry((ptr)->prev, type, field)
+
+#define	list_for_each(p, head)						\
+	for (p = (head)->next; p != (head); p = p->next)
+
+#define	list_for_each_safe(p, n, head)					\
+	for (p = (head)->next, n = p->next; p != (head); p = n, n = p->next)
+
+#define list_for_each_entry(p, h, field)				\
+	for (p = list_first_entry(h, typeof(*p), field); &p->field != (h); \
+	    p = list_entry(p->field.next, typeof(*p), field))
+
+#define list_for_each_entry_safe(p, n, h, field)			\
+	for (p = list_first_entry(h, typeof(*p), field),		\
+	    n = list_entry(p->field.next, typeof(*p), field); &p->field != (h);\
+	    p = n, n = list_entry(n->field.next, typeof(*n), field))
+
+#define	list_for_each_entry_reverse(p, h, field)			\
+	for (p = list_last_entry(h, typeof(*p), field); &p->field != (h); \
+	    p = list_entry(p->field.prev, typeof(*p), field))
+
+#define	list_for_each_prev(p, h) for (p = (h)->prev; p != (h); p = p->prev)
+#define	list_for_each_prev_safe(p, n, h) for (p = (h)->prev, n = p->prev; p != (h); p = n, n = p->prev)
+
+static inline void
+list_add(struct list_head *_new, struct list_head *head)
+{
+	_list_add(_new, head, head->next);
+}
+
+static inline void
+list_add_tail(struct list_head *_new, struct list_head *head)
+{
+	_list_add(_new, head->prev, head);
+}
+
+static inline void
+list_move(struct list_head *list, struct list_head *head)
+{
+	_list_del(list);
+	list_add(list, head);
+}
+
+static inline void
+list_move_tail(struct list_head *entry, struct list_head *head)
+{
+	_list_del(entry);
+	list_add_tail(entry, head);
+}
+
+static inline void
+_list_splice(const struct list_head *list, struct list_head *prev,
+    struct list_head *next)
+{
+	struct list_head *first;
+	struct list_head *last;
+
+	if (list_empty(list))
+		return;
+
+	first = list->next;
+	last = list->prev;
+	first->prev = prev;
+	prev->next = first;
+	last->next = next;
+	next->prev = last;
+}
+
+static inline void
+list_splice(const struct list_head *list, struct list_head *head)
+{
+	_list_splice(list, head, head->next);
+}
+
+static inline void
+list_splice_tail(struct list_head *list, struct list_head *head)
+{
+	_list_splice(list, head->prev, head);
+}
+
+static inline void
+list_splice_init(struct list_head *list, struct list_head *head)
+{
+	_list_splice(list, head, head->next);
+	INIT_LIST_HEAD(list);
+}
+
+static inline void
+list_splice_tail_init(struct list_head *list, struct list_head *head)
+{
+	_list_splice(list, head->prev, head);
+	INIT_LIST_HEAD(list);
+}
+
+#endif /* _LINUX_LIST_H_ */