第一次提交

2026-05-29 17:36:34 +08:00
commit 5b3333125d
1318 changed files with 722705 additions and 0 deletions
@@ -0,0 +1,117 @@
+#include "message_center.h"
+#include "stdlib.h"
+#include "string.h"
+#include "bsp_log.h"
+
+/* message_center是fake head node,是方便链表编写的技巧,这样就不需要处理链表头的特殊情况 */
+static Publisher_t message_center = {
+    .topic_name = "Message_Manager",
+    .first_subs = NULL,
+    .next_topic_node = NULL};
+
+static void CheckName(char *name)
+{
+    if (strnlen(name, MAX_TOPIC_NAME_LEN + 1) >= MAX_TOPIC_NAME_LEN)
+    {
+        LOGERROR("EVENT NAME TOO LONG:%s", name);
+        while (1)
+            ; // 进入这里说明话题名超出长度限制
+    }
+}
+
+static void CheckLen(uint8_t len1, uint8_t len2)
+{
+    if (len1 != len2)
+    {
+        LOGERROR("EVENT LEN NOT SAME:%d,%d", len1, len2);
+        while (1)
+            ; // 进入这里说明相同话题的消息长度却不同
+    }
+}
+
+Publisher_t *PubRegister(char *name, uint8_t data_len)
+{
+    CheckName(name);
+    Publisher_t *node = &message_center;
+    while (node->next_topic_node) // message_center会直接跳过,不需要特殊处理,这被称作dumb_head(编程技巧)
+    {
+        node = node->next_topic_node;            // 切换到下一个发布者(话题)结点
+        if (strcmp(node->topic_name, name) == 0) // 如果已经注册了相同的话题,直接返回结点指针
+        {
+            CheckLen(data_len, node->data_len);
+            node->pub_registered_flag = 1;
+            return node;
+        }
+    } // 遍历完发现尚未创建name对应的话题
+    // 在链表尾部创建新的话题并初始化
+    node->next_topic_node = (Publisher_t *)malloc(sizeof(Publisher_t));
+    memset(node->next_topic_node, 0, sizeof(Publisher_t));
+    node->next_topic_node->data_len = data_len;
+    strcpy(node->next_topic_node->topic_name, name);
+    node->pub_registered_flag = 1;
+    return node->next_topic_node;
+}
+
+Subscriber_t *SubRegister(char *name, uint8_t data_len)
+{
+    Publisher_t *pub = PubRegister(name, data_len); // 查找或创建该话题的发布者
+    // 创建新的订阅者结点,申请内存,注意要memset因为新空间不一定是空的,可能有之前留存的垃圾值
+    Subscriber_t *ret = (Subscriber_t *)malloc(sizeof(Subscriber_t));
+    memset(ret, 0, sizeof(Subscriber_t));
+    // 对新建的Subscriber进行初始化
+    ret->data_len = data_len; // 设定数据长度
+    for (size_t i = 0; i < QUEUE_SIZE; ++i)
+    { // 给消息队列的每一个元素分配空间,queue里保存的实际上是数据执指针,这样可以兼容不同的数据长度
+        ret->queue[i] = malloc(data_len);
+    }
+    // 如果是第一个订阅者,特殊处理一下,将first_subs指针指向新建的订阅者(详见文档)
+    if (pub->first_subs == NULL)
+    {
+        pub->first_subs = ret;
+        return ret;
+    }
+    // 若该话题已经有订阅者, 遍历订阅者链表,直到到达尾部
+    Subscriber_t *sub = pub->first_subs; // 作为iterator
+    while (sub->next_subs_queue)         // 遍历订阅了该话题的订阅者链表
+    {
+        sub = sub->next_subs_queue; // 移动到下一个订阅者,遇到空指针停下,说明到了链表尾部
+    }
+    sub->next_subs_queue = ret; // 把刚刚创建的订阅者接上
+    return ret;
+}
+
+/* 如果队列为空,会返回0;成功获取数据,返回1;后续可以做更多的修改,比如剩余消息数目等 */
+uint8_t SubGetMessage(Subscriber_t *sub, void *data_ptr)
+{
+    if (sub->temp_size == 0)
+    {
+        return 0;
+    }
+    memcpy(data_ptr, sub->queue[sub->front_idx], sub->data_len);
+    sub->front_idx = (sub->front_idx++) % QUEUE_SIZE; // 队列头索引增加
+    sub->temp_size--;                                 // pop一个数据,长度减1
+    return 1;
+}
+
+uint8_t PubPushMessage(Publisher_t *pub, void *data_ptr)
+{
+    static Subscriber_t *iter;
+    iter = pub->first_subs; // iter作为订阅者指针,遍历订阅该话题的所有订阅者;如果为空说明遍历结束
+    // 遍历订阅了当前话题的所有订阅者,依次填入最新消息
+    while (iter)
+    {
+        if (iter->temp_size == QUEUE_SIZE) // 如果队列已满,则需要删除最老的数据(头部),再填入
+        {
+            // 队列头索引前移动,相当于抛弃前一个位置的数据,被抛弃的位置稍后会被写入新的数据
+            iter->front_idx = (iter->front_idx + 1) % QUEUE_SIZE;
+            iter->temp_size--; // 相当于出队,size-1
+        }
+        // 将Pub的数据复制到队列的尾部(最新)
+        memcpy(iter->queue[iter->back_idx], data_ptr, pub->data_len);
+        iter->back_idx = (iter->back_idx + 1) % QUEUE_SIZE; // 队列尾部前移
+        iter->temp_size++;                                  // 入队,size+1
+
+        iter = iter->next_subs_queue; // 访问下一个订阅者
+    }
+    return 1;
+}
@@ -0,0 +1,85 @@
+/**
+ * @file message_center.h
+ * @author NeoZeng neozng1@hnu.edu.cn
+ * @brief 这是一个伪pubsub机制,仅对应用之间的通信进行了隔离
+ * @version 0.1
+ * @date 2022-11-30
+ *
+ * @copyright Copyright (c) 2022
+ *
+ */
+
+#ifndef PUBSUB_H
+#define PUBSUB_H
+
+#include "stdint.h"
+
+#define MAX_TOPIC_NAME_LEN 32 // 最大的话题名长度,每个话题都有字符串来命名
+#define MAX_TOPIC_COUNT 12    // 最多支持的话题数量
+#define QUEUE_SIZE 1
+
+typedef struct mqt
+{
+    /* 用数组模拟FIFO队列 */
+    void *queue[QUEUE_SIZE];
+    uint8_t data_len;
+    uint8_t front_idx;
+    uint8_t back_idx;
+    uint8_t temp_size; // 当前队列长度
+
+    /* 指向下一个订阅了相同的话题的订阅者的指针 */
+    struct mqt *next_subs_queue; // 使得发布者可以通过链表访问所有订阅了相同话题的订阅者
+} Subscriber_t;
+
+/**
+ * @brief 发布者类型.每个发布者拥有发布者实例,并且可以通过链表访问所有订阅了自己发布的话题的订阅者
+ *
+ */
+typedef struct ent
+{
+    /* 话题名称 */
+    char topic_name[MAX_TOPIC_NAME_LEN + 1]; // 1个字节用于存放字符串结束符 '\0'
+    uint8_t data_len;                        // 该话题的数据长度
+    /* 指向第一个订阅了该话题的订阅者,通过链表访问所有订阅者 */
+    Subscriber_t *first_subs;
+    /* 指向下一个Publisher的指针 */
+    struct ent *next_topic_node;
+    uint8_t pub_registered_flag; // 用于标记该发布者是否已经注册
+} Publisher_t;
+
+/**
+ * @brief 订阅name的话题消息
+ *
+ * @param name 话题名称
+ * @param data_len 消息长度,通过sizeof()获取
+ * @return Subscriber_t* 返回订阅者实例
+ */
+Subscriber_t *SubRegister(char *name, uint8_t data_len);
+
+/**
+ * @brief 注册成为消息发布者
+ *
+ * @param name 发布者发布的话题名称(话题)
+ * @return Publisher_t* 返回发布者实例
+ */
+Publisher_t *PubRegister(char *name, uint8_t data_len);
+
+/**
+ * @brief 获取消息
+ *
+ * @param sub 订阅者实例指针
+ * @param data_ptr 数据指针,接收的消息将会放到此处
+ * @return uint8_t 返回值为0说明没有新的消息(消息队列为空),为1说明获取到了新的消息
+ */
+uint8_t SubGetMessage(Subscriber_t *sub, void *data_ptr);
+
+/**
+ * @brief 发布者给所有订阅了话题的订阅者推送消息
+ *
+ * @param pub 发布者实例指针
+ * @param data_ptr 指向要发布的数据的指针
+ * @return uint8_t 新消息成功推送给几个订阅者
+ */
+uint8_t PubPushMessage(Publisher_t *pub, void *data_ptr);
+
+#endif // !PUBSUB_H
@@ -0,0 +1,180 @@
+# message_center
+
+<p align='right'>neozng1@hnu.edu.cn</p>
+
+> TODO:
+>
+> 支持自定义队列长度，使得订阅者可以自行确定需要的队列长度，适应不同的需求
+
+## 总览和封装说明
+
+**重要定义：**
+
+- 发布者：发布消息的对象。发布者会将自己的消息推送给所有订阅了某个特定**话题**的订阅者。
+- 订阅者：获取消息的对象。订阅者在订阅了某个话题之后，可以通过接口获得该话题的消息。
+- 话题（topic）：用于区分消息来源的对象。可以将一个话题看作一刊杂志，不同的发布者会将文章汇集到杂志上，而订阅者选择订阅一种杂志，然后就可以获取所有写在杂志上的文章。
+
+Message Center不同应用间进行消息传递的中介，它的存在可以在相当大的程度上解耦不同的app，使得不同的应用之间**不存在包含关系**，让代码的自由度更大，将不同模块之间的关系降为**松耦合**。在以往，如果一个.c文件中的数据需要被其他任务/源文件共享，那么其他模块应该要包含前者的头文件，且头文件中应当存在获取该模块数据的接口（即函数，一般是返回数据指针或直接返回数据，**强烈建议不要使用全局变量**）；但现在，不同的应用之间完全隔离，他们不需要了解彼此的存在，而是只能看见一个**消息中心**以及一些**话题**。
+
+需要被共享的消息，将会被**发布者**（publisher）发送到消息中心；要获取消息，则由**订阅者**（subscriber）从消息中心根据订阅的话题获取。在这之前，发布者要在消息中心完成**注册**，将自己要发布的消息类型和话题名称提交到消息中心；订阅者同样要先在消息中心完成订阅，将自己要接收的消息类型和话题名称提交到订阅中心。消息中心会根据**话题名称**，把订阅者绑定到发布相同名称的发布者上。
+
+> 为了节省空间，数据结构上采用了链表+循环数组模拟队列的方式。C没有哈希表，因此让发布者保存所有订阅者的地址（实际上只保存首地址，然后通过链表访问所有订阅者）。
+
+Message Center对外提供了四个接口，所有原本要进行信息交互的应用都应该包含`message_center.h`，并在初始化的时候进行注册。
+
+## 代码结构
+
+.h 文件中包含了外部接口和类型定义，.c中包含了各个接口的具体实现。
+
+## 外部接口
+
+**在代码实现上，话题名实际上就是通过一个字符串体现的。**
+
+```c
+Subscriber_t* SubRegister(char* name,uint8_t data_len);
+
+Publisher_t* PubRegister(char* name,uint8_t data_len);
+
+uint8_t SubGetMessage(Subscriber_t* sub,void* data_ptr);
+
+void PubPushMessage(Publisher_t* pub,void* data_ptr);
+```
+
+### 订阅者
+
+订阅者应该保存一个订阅者类型的指针`Subscriber_t*`，在初始化的时候调用`SubRegister()`并传入要订阅的话题名和该话题对应消息的长度，可以直接输入字符串，示例如下，将从event_name订阅float数据：
+
+```c
+Subscriber_t* my_sub;
+my_sub=SubRegister("event_name",sizeof(float));
+```
+
+订阅完毕后，在应用中通过`SubGetMessage()`获取消息，调用时传入订阅时获得的指针，以及要存放数据的指针。在使用的时候，建议使用强制类型转换将`data_ptr` cast成void*类型（好习惯）。
+
+如果消息队列中有消息，返回值为1；否则，返回值为0，说明没有新的消息可用。
+
+### 发布者
+
+发布者应该保存一个发布者类型的指针，在初始化的时候传入要发布的话题名和该话题对应的消息长度。
+
+完成注册后，通过`PubPushMessage()`发布新的消息。所有订阅了该话题的订阅者都会收到新的消息推送。
+
+### 可修改的宏
+
+```c
+#define MAX_EVENT_NAME_LEN  32  //最大的话题名长度,每个话题都由字符串来命名
+#define QUEUE_SIZE 1            //消息队列的长度
+```
+
+修改第一个可以扩大话题名长度，第二个确定消息队列的长度，数量越大可以保存的消息越多。
+
+## 私有函数和定义
+
+```c
+static Publisher_t message_center = {
+    .event_name = "Message_Manager",
+    .first_subs = NULL,
+    .next_event_node = NULL};
+
+static void CheckName(char* name)
+{
+    if(strnlen(name,MAX_EVENT_NAME_LEN+1)>=MAX_EVENT_NAME_LEN)
+    {
+        while (1); // 进入这里说明话题名超出长度限制
+    }
+}
+```
+
+`message_center`内部保存了指向第一个发布者的指针，可以看作整个消息中心的抽象。通过这个变量，可以访问所有发布者和订阅者。它将会在各个函数中作为dumb_head（哑结点）以简化逻辑，这样不需要对链表头进行特殊处理。
+
+`CheckName()`在发布者/订阅者注册的时候被调用，用于检查话题名是否超过长度限制。超长后会进入死循环，方便开发者检查。
+
+> 四个外部接口的实现都有详细的注释，有兴趣的同学可以自行阅读。下方也提供了流程图。
+
+## 注册、发布、获取消息流程
+
+包含一个结构图和四个流程图。
+
+### Message Center的结构![image-20221201150945052](../../.assets/image-20221201150945052.png)
+
+<center>建议打开原图查看</center>
+
+**多个publisher可以绑定同一个话题，往该话题推送消息。但一个subscriber只能订阅一个话题，如果应用需要订阅多个话题，则要创建对应数量的订阅者。**
+
+> 对于电控程序目前的情况，不存在多个publisher向同一个话题推送消息的情况。
+
+**对于相同话题，其消息长度必须相同**。发布者和订阅者在注册时都会传入消息长度，用`sizof(your_data)`获取。应当保证不同的模块在进行交互式，使用相同的数据长度。
+
+### 发布者和订阅者注册的流程
+
+- **发布者：**
+
+  遍历发布者的话题结点，如果发现相同的话题，直接返回指针即可；遍历完成后发现尚未创建则创建新的话题。
+
+<img src="../../.assets/image-20221201152530558.png" alt="image-20221201152530558" style="zoom: 80%;" />
+
+- **订阅者：**
+
+  需要注意，由于不同应用/模块的初始化顺序不同，可能出现订阅者先于发布者订阅某一消息的情况，所以要进行发布者链表的遍历，判断是否已经存在相同话题名的发布者，不存在则要先创建发布者结点再将新建订阅者结点并挂载到前者上。
+
+<img src="../../.assets/image-20221201152904044.png" alt="image-20221201152904044" style="zoom:80%;" />
+
+### 推送/获取消息的流程
+
+- **数组+头尾索引模拟队列**
+
+<img src="../../.assets/image-20221201155228196.png" alt="image-20221201155228196" style="zoom: 71%;" />
+
+front指向队列头，即最早入队的数据；back指向队列尾，即最新的数据。队列是first in first out（FIFO，先进先出）的结构。back指向的位置是入队数据被写入的位置，front指向的是读取时会出队的位置。当有数据入队，back++；出队则front++。若碰到数组边界，则返回数组头，可以通过取模实现：
+
+```C
+idx=(idx+1)%SIZE_OF_ARRAY; //SIZE_OF_ARRAY是数组大小
+```
+
+我们还需要一个变量用于保存当前队列的元素个数，如果在写入时，队列长度等于上限，应该先将最老的数据出队，再写入新的数据，即：
+
+```c
+back=(back+1)%SIZE_OF_ARRAY;
+size--;
+queue[front]=new_data;
+front=(front+1)%SIZE_OF_ARRAY;
+```
+
+- **发布者推送消息到指定话题**
+
+通过发布者指针，将订阅了该话题的所有订阅者遍历，将新数据入队。
+
+- **订阅者获取消息**
+
+从订阅者指针访问消息队列，取出最先进入队列的数据。注意判断队列是否为空，如果为空则返回0。
+
+## 示例代码
+
+```c
+typedef struct 
+{
+    float a;
+    uint8_t b;
+    uint32_t c;
+}good;
+
+good g1;
+good g2;
+good pub_data={.a=1,.b=2,.c=3};
+// 一个发布者,两个订阅者
+Subscriber_t* s=SubRegister("test",sizeof(good));
+Subscriber_t* ss=SubRegister("test",sizeof(good));
+Publisher_t* p=PubRegister("test",sizeof(good));
+// 推送消息
+PubPushMessage(p,&pub_data);
+pub_data.a=4;
+pub_data.b=5;
+pub_data.c=6;
+// 推送新消息
+PubPushMessage(p,&pub_data);
+volatile uint8_t d= 0; // 确定收到的消息是否有效,可以根据d的值进一步处理
+d=SubGetMessage(s,&g1);
+d=SubGetMessage(s,&g1);
+d=SubGetMessage(s,&g1); // 此时d等于0
+d=SubGetMessage(ss,&g2);
+```