# rt_msg

**Repository Path**: mlzhu/rt_msg

## Basic Information

- **Project Name**: rt_msg
- **Description**: 一个面向嵌入式 / 上位机的轻量 C 消息总线，把"事件 - 回调 - 定时器 - 广播频道"统一成一套去耦合机制。支持RTOS和Linux环境，零外部依赖、可移植、多实例、线程安全。
- **Primary Language**: C
- **License**: MIT
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-09-04
- **Last Updated**: 2026-09-04

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# rt_msg — 去耦合消息回调库

一个面向嵌入式 / 上位机的轻量 C 消息总线，把"事件 - 回调 - 定时器 - 广播频道"统一成一套去耦合机制。
**零外部依赖、可移植、多实例、线程安全**。

## 特性

- ✅ **同步消息**：`push` 时立即执行回调
- ✅ **异步消息**：`push` 置位、`pop` 才执行（适合 ISR → 主循环解耦）
- ✅ **定时器**：单次 / 周期，并入同一套消息机制
- ✅ **广播频道**：一对多多线程同步消息，每订阅者独立 FIFO，支持阻塞 recv/wait
- ✅ **多实例**：每个 `rt_msg_inst_t` / `rt_msg_channel_t` 互相隔离
- ✅ **线程安全**：所有公共 API 内部加锁，回调在解锁后执行避免死锁
- ✅ **跨平台**：核心库只依赖 `<stdint.h>` / `<string.h>`，对任何 MCU/RTOS/PC 零依赖
- ✅ **可选模块 `rt_err`**：纯 bitmap 错误码管理，与消息总线完全解耦

## 两套模型对比

| 模型 | 适用场景 | API 风格 | 并发模型 |
|---|---|---|---|
| **rt_msg**（push/pop） | 单线程主循环、ISR→main 解耦、定时器回调 | register/push/pop | 覆盖式（一个 key 一个回调，新 data 覆盖旧 data） |
| **rt_msg_channel**（broadcast/recv） | 多线程生产者-消费者、事件广播 | subscribe/broadcast/recv | FIFO 队列（每订阅者独立队列，消息不丢失） |

## 目录结构

```
rt_msg_open/
├── include/
│   ├── rt_msg.h          push/pop 模型 API
│   ├── rt_msg_channel.h  广播频道 API（多线程一对多）
│   ├── rt_msg_port.h     平台适配接口
│   └── rt_err.h          可选：错误码位图
├── src/
│   ├── rt_msg.c          push/pop 实现
│   ├── rt_msg_channel.c  广播频道实现
│   ├── rt_msg_port.c     libc 默认适配
│   └── rt_err.c          错误码实现
├── ports/
│   ├── freertos/         FreeRTOS 适配（lock + sem）
│   └── linux/            Linux/POSIX 适配（pthread + sem_t）
├── example/
│   ├── example_main.c    push/pop + 定时器 demo
│   └── example_channel.c 多线程广播 demo（1 发布者 + 3 消费者）
├── LICENSE               MIT
└── README.md
```

## 广播频道快速上手

### 1. 配置 port（必须提供 sem 接口）

```c
/* Linux */
rt_msg_port_use_linux();

/* FreeRTOS */
rt_msg_port_use_freertos();

/* libc 默认（PC 跨平台，自带条件变量 sem） */
rt_msg_port_use_libc_default();
```

### 2. 创建频道 + 订阅

```c
rt_msg_channel_t *ch = rt_msg_channel_create(16);  /* 队列深度 16 */

/* 每个消费者线程订阅一个 */
rt_msg_subscriber_t *sub_logger = rt_msg_channel_subscribe(ch);
rt_msg_subscriber_t *sub_ui     = rt_msg_channel_subscribe(ch);
rt_msg_subscriber_t *sub_net    = rt_msg_channel_subscribe(ch);
```

### 3. 发布者广播

```c
const char *event = "socket_connected";
rt_msg_channel_broadcast(ch, (void *)event, NULL);
/* 三个订阅者各自队列都会收到一份 */
```

### 4. 消费者接收（阻塞 / 非阻塞 / 等待）

```c
/* 阻塞接收（带超时） */
void *data = NULL;
rt_msg_err_t r = rt_msg_sub_recv(sub_logger, &data, 1000);  /* 1s 超时 */
if (r == RT_MSG_OK) {
    printf("got: %s\n", (const char *)data);
}

/* 非阻塞 */
r = rt_msg_sub_try_recv(sub_ui, &data);

/* 等待但不取走（多频道轮询） */
int has = rt_msg_sub_wait(sub_net, 1000);
```

### 5. 销毁

```c
/* 取消单个订阅（调用方需保证消费者线程已退出） */
rt_msg_channel_unsubscribe(sub_logger);

/* 销毁频道，自动清理所有剩余订阅者 */
rt_msg_channel_destroy(ch);
```

## 广播频道 API 一览

| 分类 | 函数 |
|---|---|
| 频道生命周期 | `rt_msg_channel_create` `rt_msg_channel_destroy` |
| 订阅 | `rt_msg_channel_subscribe` `rt_msg_channel_subscribe_ext` `rt_msg_channel_unsubscribe` |
| 发布 | `rt_msg_channel_broadcast` |
| 接收 | `rt_msg_sub_recv` `rt_msg_sub_try_recv` `rt_msg_sub_wait` |
| 统计 | `rt_msg_sub_pending` `rt_msg_sub_dropped` |

## 广播频道设计要点

### data 所有权

广播场景下 `data` 被多订阅者共享，`data_free` **必须为 NULL**。data 生命周期由发送方管理：
- 静态/常量数据：直接用
- 动态数据：发送方在确认所有订阅者都 recv 完后再释放（或用引用计数包装）

### 队列满策略

某订阅者队列满时，**新消息仍入队尾**，**队首最旧消息被覆盖挤掉**，该订阅者 `dropped++`，**不影响其他订阅者**。
返回 `RT_MSG_ERR_FULL`（其他订阅者仍收到，且本订阅者也会收到最新消息，只是丢失了最旧的一条）。
可通过 `rt_msg_sub_dropped()` 查询被覆盖的旧消息数。

策略选择理由：广播场景下"最新数据"通常比"完整历史"更重要（如状态、传感器读数），
覆盖最旧消息能保证消费者始终拿到最新数据，不会因队列满而饿死新消息。

### 阻塞与超时

- `recv(timeout)`：阻塞取走一条消息，超时返回 `RT_MSG_ERR_TIMEOUT`
- `wait(timeout)`：阻塞等待消息到达但不取走（多频道轮询场景）；返回 1=有消息 0=超时
- `try_recv`：非阻塞
- `timeout = (uint32_t)-1`：永久等待

### unsubscribe 安全性

调用方必须保证 `unsubscribe` 时没有其他线程在 `recv` 该订阅者，否则 sem 销毁会导致阻塞线程崩溃。
推荐做法：消费者线程收到特定"退出消息"后自行 return，主控线程再 unsubscribe。

## push/pop 模型快速上手

### 1. 配置 port（任选其一）

```c
/* A. PC 单线程测试 */
rt_msg_port_use_libc_default();

/* B. FreeRTOS 工程 */
rt_msg_port_use_freertos();
```

### 2. 注册消息

```c
static rt_msg_inst_t g_bus;

rt_msg_init(&g_bus);

/* 同步消息 */
rt_msg_register(&g_bus, GROUP_SYS, KEY_REBOOT,
                RT_MSG_TYPE_SYNC, reboot_cb, "user-ctx");

/* 异步消息 */
rt_msg_register(&g_bus, GROUP_NET, KEY_SOCKET,
                RT_MSG_TYPE_ASYNC, socket_cb, NULL);

/* 周期定时器 */
rt_timer_register(&g_bus, KEY_LED, RT_TIMER_PERIODIC, led_cb, NULL);
rt_timer_start  (&g_bus, KEY_LED, 500, NULL, NULL);  /* 500ms */
```

### 3. 发布 / 消费

```c
/* ISR 或任意上下文 push */
int stat = 1;
rt_msg_push(&g_bus, KEY_SOCKET, &stat, NULL);

/* 主循环 pop 异步 + 驱动定时器 */
for (;;) {
    rt_msg_pop(&g_bus, KEY_SOCKET);
    rt_timer_tick(&g_bus, 10);   /* 10ms 调一次 */
    your_os_delay(10);
}
```

### 4. 动态数据自动释放

```c
int *p = malloc(sizeof(int));
*p = 42;
rt_msg_push(&g_bus, KEY_SOCKET, p, free);   /* 下次覆盖 / 注销时自动 free */
```

## push/pop API 一览

| 分类 | 函数 |
|---|---|
| 生命周期 | `rt_msg_init` `rt_msg_deinit` |
| 注册 | `rt_msg_register` `rt_msg_unregister` |
| 消息 | `rt_msg_push` `rt_msg_pop` `rt_msg_pending` `rt_msg_clear` |
| 定时器 | `rt_timer_register` `rt_timer_start` `rt_timer_stop` `rt_timer_unregister` `rt_timer_tick` |
| 错误码 | `rt_err_init` `rt_err_set` `rt_err_get_bit` `rt_err_get_word` `rt_err_clear_group` `rt_err_clear_all` |

## 编译示例

```bash
# push/pop demo（单线程）
gcc -I include example/example_main.c src/rt_msg.c src/rt_msg_port.c src/rt_err.c -o demo
./demo

# 广播频道 demo（多线程，Linux/WSL/MinGW）
gcc -Wall -I include example/example_channel.c src/rt_msg_channel.c src/rt_msg_port.c -o ch_demo
./ch_demo
```

广播频道 demo 预期输出（顺序可能交错）：
```
[consumer 0:logger] recv: hello
[consumer 1:ui]     recv: hello
[consumer 2:net]    recv: hello
[consumer 0:logger] recv: world
[consumer 1:ui]     recv: world
...
[consumer 2:net]    recv: done
[consumer 0:logger] exit
[consumer 1:ui]     exit
[consumer 2:net]    exit
[stat] sub 0 pending=0 dropped=0
[stat] sub 1 pending=0 dropped=0
[stat] sub 2 pending=0 dropped=0
== done ==
```

## 容量配置

可在编译时通过 `-D` 覆盖：

| 宏 | 默认 | 含义 |
|---|---|---|
| `RT_MSG_MAX_GROUPS` | 16 | 最多消息组数 |
| `RT_MSG_GROUP_MAX_ITEMS` | 16 | 每组最多 key 数 |
| `RT_MSG_MAX_TIMERS` | 16 | 最多定时器数 |
| `RT_MSG_CHANNEL_DEFAULT_DEPTH` | 16 | 频道默认队列深度 |
| `RT_MSG_CHANNEL_MAX_SUBS` | 16 | 单频道最大订阅者数 |

## 线程安全说明

- 所有公共 API 内部加锁（锁实现由 port 提供）
- **回调在解锁后执行**，回调里可以安全地调用本库其他 API，不会死锁
- port 未配置锁时退化为非线程安全，适合单线程环境
- 广播频道 `unsubscribe` 必须保证无其他线程在 `recv` 该订阅者

## 设计要点

### 为什么不内置 group/key 枚举？

业务相关的 key 由用户在自己代码里定义。库只认 `uint32_t`，`0` 保留为无效。
这避免了库被业务"污染"，保持通用性。

### 定时器为什么并入 rt_msg？

定时器本质是"时间到达时触发一次回调"，和异步消息触发回调是同一种语义。
合并后用户只需学一套 API，且定时器到点自动通过异步消息机制派发，
不需要在主循环里分别处理两套队列。

### data 所有权

`rt_msg_push` 的 `data_free` 参数决定数据生命周期：

| 场景 | `data_free` 传值 |
|---|---|
| 静态/栈上数据 | `NULL`（库不管） |
| 动态数据，回调里自己释放 | `NULL`（约定回调释放） |
| 动态数据，希望库自动管理 | `free` 或自定义释放函数 |

定时器 `rt_timer_start` 同理。

## License

MIT，见 [LICENSE](LICENSE)。