diff --git a/Documents/api-reference.md b/Documents/api-reference.md index 4b91078..8a71715 100644 --- a/Documents/api-reference.md +++ b/Documents/api-reference.md @@ -190,7 +190,6 @@ PID 接口是旧版控制器工具。Motor 新 controller 不依赖该公共 API | `motor_can_sched_register_can(can_dev)` | 注册一个 CAN 设备。 | `0` 成功，负 errno 失败。 | | `motor_can_sched_send(can_dev, frame, param, handle_out)` | 通用发送入口，可配置周期、应答追踪与优先级。 | `0` 成功，负 errno 失败。 | | `motor_can_sched_send_with_priority(can_dev, frame, priority, tag)` | 按明确优先级发送单帧。 | `0` 成功，负 errno 失败。 | -| `motor_can_sched_send_prio(can_dev, frame, high_priority, tag)` | 发送单帧，高优先级布尔入口。 | `0` 成功，负 errno 失败。 | | `motor_can_sched_send_reply(can_dev, frame, reply_id, reply_mask, timeout_ms, tag)` | 发送请求回复帧。 | `0` 成功，负 errno 失败。 | | `motor_can_sched_update(handle, frame)` | 更新周期帧内容。 | `0` 成功，负 errno 失败。 | | `motor_can_sched_remove(handle)` | 删除周期帧。 | `0` 成功，负 errno 失败。 | @@ -306,6 +305,29 @@ PID 接口是旧版控制器工具。Motor 新 controller 不依赖该公共 API Dual Protocol 支持函数帧、同步帧、错误帧与回复帧。协议常量定义在同一头文件中。 +### MQTT-like 协议 + +头文件：`include/ares/protocol/mqttlite/mqttlite_protocol.h` + +| 接口 | 说明 | +| --- | --- | +| `ares_mqttlite_register_topic(protocol, topic_id, topic)` | 注册本地 topic id 映射。 | +| `ares_mqttlite_subscribe(protocol, topic_filter, cb, user_data)` | 注册本地 topic 订阅回调。 | +| `ares_mqttlite_subscribe_id(protocol, topic_id, cb, user_data)` | 按已约定 topic id 注册本地订阅回调。 | +| `ares_mqttlite_unsubscribe(protocol, topic_filter)` | 移除本地 topic 订阅。 | +| `ares_mqttlite_unsubscribe_id(protocol, topic_id)` | 移除本地 topic id 订阅。 | +| `ares_mqttlite_publish(protocol, topic, payload, payload_len, qos, cb, user_data)` | 发布消息，QoS1/QoS2 成功时返回 packet id。 | +| `ares_mqttlite_publish_id(protocol, topic_id, payload, payload_len, qos, cb, user_data)` | 使用 2 字节 topic id 发布消息。 | +| `ares_mqttlite_publish_prepare_id(protocol, topic_id, payload_len, pub)` | 准备 QoS0 topic-id 零拷贝发布缓冲。 | +| `ares_mqttlite_publish_commit(protocol, pub)` | 发送已填充的零拷贝发布缓冲。 | +| `ares_mqttlite_publish_abort(pub)` | 释放未提交的零拷贝发布缓冲。 | +| `ares_mqttlite_ping(protocol)` | 发送 PING 帧。 | +| `ARES_MQTTLITE_PROTOCOL_DEFINE(name)` | 定义 MQTT-like 协议实例。 | + +协议支持 `ARES_MQTTLITE_QOS0`、`ARES_MQTTLITE_QOS1` 与 `ARES_MQTTLITE_QOS2`。QoS1 使用 +`PUBLISH/PUBACK`，QoS2 使用 `PUBLISH/PUBREC/PUBREL/PUBCOMP`。payload 是变长字段，长度由每帧 +header 携带，并受 `CONFIG_ARES_MQTTLITE_MAX_PAYLOAD_SIZE` 限制。 + ### Plotter 协议 头文件：`include/ares/protocol/plotter/aresplot_protocol.h` diff --git a/Documents/communication/interface.md b/Documents/communication/interface.md index 74b85fd..4dc7d62 100644 --- a/Documents/communication/interface.md +++ b/Documents/communication/interface.md @@ -147,7 +147,7 @@ UART 异步回调收到 `UART_RX_RDY` 后： 这一设计保证了单通道串口发送有统一仲裁点，避免多个上下文直接打到底层驱动。 `ares_uart_send_raw()` 则是例外，它直接调用 `uart_tx()`，更适合 plotter/VOFA 这种自带帧组装、 -不依赖 `net_buf` 的快速路径。 +不依赖 `net_buf` 的快速路径。UART 接口没有实现 `alloc_buf_with_data()`。 ### 配置项 @@ -200,7 +200,10 @@ USB bulk 接口适合： - `ares_usbd_init(struct AresInterface *interface)` - `ares_usbd_write(struct AresInterface *interface, struct net_buf *buf)` -- `ares_usbd_write_with_lock(struct AresInterface *interface, struct net_buf *buf, struct k_mutex *mutex)` +- `ares_usbd_write_with_callback(struct AresInterface *interface, struct net_buf *buf, + ares_interface_tx_done_cb_t cb, void *user_data)` +- `ares_usbd_caps(struct AresInterface *interface)` +- `ares_usbd_mtu(struct AresInterface *interface)` - `ares_interface_alloc_buf(struct AresInterface *interface)` - `ares_interface_alloc_buf_with_data(struct AresInterface *interface, void *data, size_t len)` @@ -248,20 +251,24 @@ OUT 端点启用后会持续预投递读请求。收到数据后： #### 发送 -发送调用 `ares_usbd_write()` 或 `ares_usbd_write_with_lock()`： +发送调用 `ares_usbd_write()` 或 `ares_usbd_write_with_callback()`： 1. 检查 USB class 和接口状态。 2. 检查 IN 端点是否已有在飞事务。 -3. 在 `net_buf` 用户区写入端点与互斥信息。 +3. 在 `net_buf` 用户区写入端点与完成回调信息。 4. 调用 `usbd_ep_enqueue()`。 -发送完成后，request handler 清除 IN engaged 状态，并通过 buffer 回调释放 `net_buf`； -若附带互斥锁，还会在缓冲释放回调中解锁。 +发送完成后，request handler 清除 IN engaged 状态，调用发送完成回调并释放 `net_buf`。同步入队失败时， +USB 层也会调用完成回调并释放缓冲。 + +USB bulk 接口没有实现 `send_raw()` 与 `connect()` / `disconnect()` / `is_connected()`。连接状态通过 +USBD 消息转换为协议事件，而不是通过 `AresInterfaceAPI` 查询。 ### `alloc_buf_with_data()` 的作用 -双向协议会优先使用 `alloc_buf_with_data()`。对 USB 来说，这允许直接把现成帧数据包装成 -发送缓冲，减少一次显式拷贝。维护者修改协议发送路径时，应保留这种能力协商。 +双向协议会优先使用 `alloc_buf_with_data()`。对 USB 来说，这允许用现成帧数据创建发送缓冲，减少协议层 +显式拷贝。维护者修改协议发送路径时，应保留这种能力协商，并为没有该能力的接口保留 +`alloc_buf()` + `net_buf_add_mem()` fallback。 ### 协议事件 diff --git a/Documents/communication/protocol.md b/Documents/communication/protocol.md index ad28e51..abd2757 100644 --- a/Documents/communication/protocol.md +++ b/Documents/communication/protocol.md @@ -183,7 +183,7 @@ SYNC 帧长度不能仅靠帧头得知，必须在收到 ID 后查同步表长 双向协议的发送始终通过已绑定接口： -- 优先使用 `alloc_buf_with_data()` 包装现成帧 +- 优先使用 `alloc_buf_with_data()` 用现成帧数据创建发送缓冲 - 否则分配 `net_buf` 后复制数据 - 需要等待 TX 完成的 FUNC/同步帧使用 `send_with_callback()` 清理在飞状态 @@ -227,6 +227,124 @@ dual_func_add(&link_proto, 0x10, my_rpc); dual_sync_add(&link_proto, 0x20, sync_buf, sizeof(sync_buf), sync_cb); ``` +## MQTT-like 协议 + +### 角色 + +MQTT-like 协议面向 USB bulk 这类可靠包传输接口上的轻量发布/订阅。它复用 +`AresInterface`，不直接依赖 USB 类实现，因此接口层仍负责收发 `net_buf`，协议层只处理 topic、 +QoS 握手和消息分发。 + +实现位于： + +- 头文件：`include/ares/protocol/mqttlite/mqttlite_protocol.h` +- 实现：`lib/ares/protocol/mqttlite/mqttlite_protocol.c` + +### 帧类型 + +协议帧包含固定 12 字节头： + +- magic +- version +- frame type +- flags +- packet id +- topic length +- payload length + +当前定义的帧类型： + +- `PUBLISH` +- `PUBACK` +- `PUBREC` +- `PUBREL` +- `PUBCOMP` +- `PING` +- `PONG` + +topic 与 payload 紧跟在固定头之后。payload 是变长字段，长度由 frame header 携带，不是定长。 +最大 topic、payload、frame、订阅数量、topic id 注册数量和在飞 QoS 包数量由 +`CONFIG_ARES_MQTTLITE_*` 控制。 + +### Topic ID + +普通 `PUBLISH` 帧携带完整 topic 字符串。性能敏感路径可以先用 +`ares_mqttlite_register_topic()` 在本地注册 `uint16_t topic_id -> topic` 映射，再使用： + +- `ares_mqttlite_subscribe_id()` +- `ares_mqttlite_unsubscribe_id()` +- `ares_mqttlite_publish_id()` + +topic-id 帧会设置 frame flags 中的 topic-id 位，并把 topic 字段压缩为 2 字节 little-endian id。 +接收端若注册了对应 id，会把 id 反查为字符串传给原有消息回调；未注册时 id 订阅仍可匹配，但回调中的 +topic 字符串为空。 + +### QoS + +协议支持三档 QoS： + +- `ARES_MQTTLITE_QOS0`：最多一次，发送后不等待确认。 +- `ARES_MQTTLITE_QOS1`：至少一次，发送端等待 `PUBACK`，超时重发 `PUBLISH`。 +- `ARES_MQTTLITE_QOS2`：恰好一次语义，使用 `PUBLISH/PUBREC/PUBREL/PUBCOMP`，接收端在 + `PUBREL` 阶段交付。 + +QoS1/QoS2 的发送端会记录在飞包，按 `CONFIG_ARES_MQTTLITE_RETRY_INTERVAL_MS` 重试，超过 +`CONFIG_ARES_MQTTLITE_MAX_RETRIES` 后通过发布回调返回超时。 + +### Topic 匹配 + +订阅使用 `ares_mqttlite_subscribe()` 注册本地回调。topic filter 支持： + +- 精确匹配 +- 单层通配符 `+` +- 末尾多层通配符 `#` + +协议当前不向远端发送 SUBSCRIBE 控制帧；订阅表是本地分发表。两端需要按应用契约分别注册各自要接收的 +topic。 + +topic-id 订阅不做通配符匹配，只比较 `uint16_t topic_id`。这条路径适合高频 telemetry 和固定控制 +topic。 + +### 发送路径 + +发布消息使用： + +```c +ARES_MQTTLITE_PROTOCOL_DEFINE(link_proto); + +ares_bind_interface(&usb_bulk_interface, &link_proto); +ares_mqttlite_subscribe(&link_proto, "robot/+/state", state_cb, NULL); +ares_mqttlite_publish(&link_proto, "robot/chassis/state", payload, payload_len, + ARES_MQTTLITE_QOS1, publish_cb, NULL); +``` + +协议通过已绑定接口分配 `net_buf` 并调用 `send()`。因此它可以跑在 USB bulk 上，也可以跑在其他实现了 +`AresInterfaceAPI` 的块式接口上。 + +QoS0 的 topic-id 快速路径可以避免 payload 复制： + +```c +struct ares_mqttlite_publish_buffer pub; + +ares_mqttlite_register_topic(&link_proto, 1, "robot/chassis/state"); +ares_mqttlite_publish_prepare_id(&link_proto, 1, payload_len, &pub); +memcpy(pub.payload, payload, payload_len); +ares_mqttlite_publish_commit(&link_proto, &pub); +``` + +这条路径直接让调用者写入最终 `net_buf` 的 payload 区域。QoS1/QoS2 仍会保留 payload 副本用于重试。 + +### 错误边界 + +MQTT-like 协议的主要错误边界： + +- topic 或 payload 超过配置上限时，发布返回 `-EINVAL`。 +- frame 超过接收缓冲上限时，解析器丢弃该帧。 +- topic id 为 0 或未按应用契约双边约定时，远端无法按预期分发消息。 +- QoS 在飞表满时，QoS1/QoS2 发布返回 `-ENOMEM`。 +- QoS 重试耗尽时，发布回调收到 `ARES_MQTTLITE_PUBLISH_TIMEOUT`。 +- 断连事件会清空在飞 QoS 状态和已暂存的 QoS2 接收状态。 + ## 绘图协议 ### 角色 diff --git a/Documents/modules.md b/Documents/modules.md index 1a443ea..f3eb14c 100644 --- a/Documents/modules.md +++ b/Documents/modules.md @@ -156,7 +156,7 @@ Graph 当前承载 INSLink 相关代码。该目录并非通用图形库，维 - `include/ares/protocol/` - `lib/ares/protocol/` -协议层是帧解析和业务分发层，目前包含 Dual Protocol 与 Plotter Protocol。 +协议层是帧解析和业务分发层，目前包含 Dual Protocol、MQTT-like Protocol 与 Plotter Protocol。 ### ARES 通信（ARES Comm） diff --git a/Documents/motor/architecture.md b/Documents/motor/architecture.md index 8f28b8e..ee8ae1a 100644 --- a/Documents/motor/architecture.md +++ b/Documents/motor/architecture.md @@ -196,7 +196,6 @@ stateDiagram-v2 | 接口 | 用途 | | --- | --- | | `motor_can_sched_register_can()` | 注册 CAN 设备。 | -| `motor_can_sched_send_prio()` | 发送普通帧，可选高优先级。 | | `motor_can_sched_send_with_priority()` | 以明确优先级发送单帧。 | | `motor_can_sched_send_reply()` | 发送会触发回复的帧，并跟踪回复超时。 | | `motor_can_sched_send()` | 通用入口，支持周期帧和回复跟踪。 | diff --git a/Documents/motor/can-scheduler.md b/Documents/motor/can-scheduler.md index 27c4f52..8fda756 100644 --- a/Documents/motor/can-scheduler.md +++ b/Documents/motor/can-scheduler.md @@ -44,7 +44,7 @@ int motor_can_sched_register_can(const struct device *can_dev); - 周期控制帧默认使用 `NORMAL`。 - 不要把所有帧都提升到高优先级，否则调度器失去意义。 -`motor_can_sched_send_prio()` 是较窄的便捷接口，只暴露“普通”与“高优先级”的常用路径。需要更细粒度控制时，使用 `motor_can_sched_send_with_priority()` 或通用 `motor_can_sched_send()`。 +单帧发送应使用 `motor_can_sched_send_with_priority()` 明确声明优先级；需要周期帧或回复跟踪时，使用通用 `motor_can_sched_send()`。 ## 发送模型 diff --git a/Documents/samples/communication.md b/Documents/samples/communication.md index 2090de8..a6ec28a 100644 --- a/Documents/samples/communication.md +++ b/Documents/samples/communication.md @@ -1,11 +1,12 @@ # Communication 样例 -通信样例分为 `ARES 通信` 与 `Plotter 协议` 两条线，覆盖 USB Bulk、UART 接口和日志输出行为。 +通信样例分为 `ARES 通信`、`MQTT-like USB` 与 `Plotter 协议` 几条线，覆盖 USB Bulk、UART +接口和日志输出行为。 ## 共同边界 - `CONFIG_UART_INTERFACE` 与 `CONFIG_USB_BULK_INTERFACE` 在各样例中的组合不同，须按 `boards/*` 与 `prj.conf` 一起确认。 -- 示例流程以 `CONFIG_DUAL_PROPOSE_PROTOCOL` 或 `CONFIG_PLOTTER` 启用为入口，接口初始化失败时通常会跳过失败通道并继续运行。 +- 示例流程以 `CONFIG_DUAL_PROPOSE_PROTOCOL`、`CONFIG_ARES_MQTTLITE_PROTOCOL` 或 `CONFIG_PLOTTER` 启用为入口，接口初始化失败时通常会跳过失败通道并继续运行。 - 上位机联调前先确认 `usart6` 与 USB OTG 物理连接。 ## samples/communication/ares_communication @@ -24,6 +25,20 @@ - 若新增板适配，先确认 `boards/.conf` 是否覆盖了 `CONFIG_UART_INTERFACE`。 - 回调函数语义（`func_cb` / `sync_cb` / `func_ret_cb`）变化需同步示例文档与接收端联调脚本。 +## samples/communication/mqttlite_usb + +- 用途：MQTT-like 协议的 USB Bulk 发布/订阅演示，周期通过 topic-id 零拷贝路径发布 QoS0 heartbeat，并按 id 订阅 `host/command`。 +- 构建： + - `west build -b dm_mc02 samples/communication/mqttlite_usb --pristine` + - `west flash` +- 适用 board：`dm_mc02` +- 硬件依赖： + - USB 设备能力 + - 兼容 MQTT-like 帧格式的上位机 +- 维护规则： + - QoS 行为变化需同步 `include/ares/protocol/mqttlite/mqttlite_protocol.h` 和协议文档。 + - topic 名称变化需同步上位机接收端。 + ## samples/communication/plotter_auto - 用途：自动上报变量的 Plotter 示例。 diff --git a/Documents/zephyr/kconfig.md b/Documents/zephyr/kconfig.md index 940ceb4..0869202 100644 --- a/Documents/zephyr/kconfig.md +++ b/Documents/zephyr/kconfig.md @@ -61,6 +61,8 @@ Motor 子系统的调度和节拍选项： | `CONFIG_ARES` | 启用 ARES 库菜单。 | | `CONFIG_UART_INTERFACE` | 启用 UART 接口。 | | `CONFIG_USB_BULK_INTERFACE` | 启用 USB Bulk 接口。 | +| `CONFIG_ARES_MQTTLITE_PROTOCOL` | 启用 MQTT-like 发布/订阅协议。 | +| `CONFIG_ARES_MQTTLITE_MAX_TOPICS` | 设置 MQTT-like topic id 注册数量上限。 | | `CONFIG_ARES_BOARD_STATUS_LED` | 启用板级状态 LED 服务。 | | `CONFIG_IMU_PWM_TEMP_CTRL` | 启用 IMU PWM 温控。 | | `CONFIG_AUTO_PROBE_GYRO_BIAS` | 启用陀螺偏置自动探测。 | diff --git a/drivers/motor/common/motor_can_sched.c b/drivers/motor/common/motor_can_sched.c index d675ca1..5d8e896 100644 --- a/drivers/motor/common/motor_can_sched.c +++ b/drivers/motor/common/motor_can_sched.c @@ -1010,17 +1010,6 @@ int motor_can_sched_send(const struct device *can_dev, const struct can_frame *f return 0; } -int motor_can_sched_send_prio(const struct device *can_dev, const struct can_frame *frame, - bool high_priority, const char *tag) -{ - const struct motor_can_sched_tx_param param = { - .high_priority = high_priority, - .tag = tag, - }; - - return motor_can_sched_send(can_dev, frame, ¶m, NULL); -} - int motor_can_sched_send_with_priority(const struct device *can_dev, const struct can_frame *frame, enum motor_can_sched_prio priority, const char *tag) { diff --git a/drivers/motor/common/motor_can_sched.h b/drivers/motor/common/motor_can_sched.h index 641b3b1..8d55f09 100644 --- a/drivers/motor/common/motor_can_sched.h +++ b/drivers/motor/common/motor_can_sched.h @@ -104,8 +104,6 @@ int motor_can_sched_send(const struct device *can_dev, const struct can_frame *f motor_can_sched_handle_t *handle_out); int motor_can_sched_send_with_priority(const struct device *can_dev, const struct can_frame *frame, enum motor_can_sched_prio priority, const char *tag); -int motor_can_sched_send_prio(const struct device *can_dev, const struct can_frame *frame, - bool high_priority, const char *tag); int motor_can_sched_send_reply(const struct device *can_dev, const struct can_frame *frame, uint32_t reply_id, uint32_t reply_mask, uint16_t timeout_ms, const char *tag); diff --git a/drivers/motor/dji/motor_dji.c b/drivers/motor/dji/motor_dji.c index 74ca5d7..a979ab8 100644 --- a/drivers/motor/dji/motor_dji.c +++ b/drivers/motor/dji/motor_dji.c @@ -342,7 +342,9 @@ void dji_control(const struct device *dev, enum motor_cmd cmd) frame.data[1] = (cfg->common.rx_id - 0x200) >> 8; frame.data[2] = 0x55; frame.data[3] = 0x3C; - motor_can_sched_send_prio(cfg->common.phy, &frame, true, "dji-set-zero"); + motor_can_sched_send_with_priority(cfg->common.phy, &frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, + "dji-set-zero"); } break; case CLEAR_CONTROLLER: @@ -357,7 +359,9 @@ void dji_control(const struct device *dev, enum motor_cmd cmd) frame.data[1] = (cfg->common.rx_id - 0x200) >> 8; frame.data[2] = 0x55; frame.data[3] = 0x50; - motor_can_sched_send_prio(cfg->common.phy, &frame, true, "dji-clear-error"); + motor_can_sched_send_with_priority(cfg->common.phy, &frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, + "dji-clear-error"); } break; } @@ -790,8 +794,9 @@ void dji_tx_handler(struct k_work *work) txframe.dlc = 8; txframe.flags = 0; const struct device *can_dev = ctrl_struct->can_dev; - motor_can_sched_send_prio(can_dev, &txframe, true, - "dji-feedback-control"); + motor_can_sched_send_with_priority(can_dev, &txframe, + MOTOR_CAN_SCHED_PRIO_CRITICAL, + "dji-feedback-control"); } } } diff --git a/drivers/motor/dm/motor_dm.c b/drivers/motor/dm/motor_dm.c index 9570a0c..273ba47 100644 --- a/drivers/motor/dm/motor_dm.c +++ b/drivers/motor/dm/motor_dm.c @@ -93,7 +93,8 @@ static int dm_send_cmd_frame(const struct device *dev, const uint8_t data[8], co }; memcpy(frame.data, data, 8); - return motor_can_sched_send_prio(cfg->common.phy, &frame, true, tag); + return motor_can_sched_send_with_priority(cfg->common.phy, &frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, tag); } static int dm_send_cmd_frame_reply(const struct device *dev, const uint8_t data[8], const char *tag) @@ -301,7 +302,7 @@ static void dm_edit_reg_value(const struct device *dev, uint16_t can_id, uint8_t frame.data[5] = reg_value >> 8; frame.data[6] = reg_value >> 16; frame.data[7] = reg_value >> 24; - motor_can_sched_send_prio(dev, &frame, true, "dm-reg"); + motor_can_sched_send_with_priority(dev, &frame, MOTOR_CAN_SCHED_PRIO_CRITICAL, "dm-reg"); } static void dm_edit_reg_float(const struct device *dev, uint16_t can_id, uint8_t reg_addr, diff --git a/drivers/motor/lk/motor_lk.c b/drivers/motor/lk/motor_lk.c index 4a4158d..54d29c9 100644 --- a/drivers/motor/lk/motor_lk.c +++ b/drivers/motor/lk/motor_lk.c @@ -62,13 +62,15 @@ void lk_motor_control(const struct device *dev, enum motor_cmd cmd) switch (cmd) { case ENABLE_MOTOR: frame.data[0] = LK_CMD_MOTOR_RUN; // 0x88 - motor_can_sched_send_prio(cfg->common.phy, &frame, true, "lk-enable"); + motor_can_sched_send_with_priority(cfg->common.phy, &frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, "lk-enable"); k_msleep(10); motor_link_request_enable(&data->common.link); break; case DISABLE_MOTOR: frame.data[0] = LK_CMD_MOTOR_OFF; // 0x80 - motor_can_sched_send_prio(cfg->common.phy, &frame, true, "lk-disable"); + motor_can_sched_send_with_priority(cfg->common.phy, &frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, "lk-disable"); motor_link_request_disable(&data->common.link); break; case SET_ZERO: @@ -82,11 +84,13 @@ void lk_motor_control(const struct device *dev, enum motor_cmd cmd) frame.data[5] = (((uint32_t)(set_angle * LK_POS_FACTOR)) >> 8) & 0xFF; frame.data[6] = (((uint32_t)(set_angle * LK_POS_FACTOR)) >> 16) & 0xFF; frame.data[7] = (((uint32_t)(set_angle * LK_POS_FACTOR)) >> 24) & 0xFF; - motor_can_sched_send_prio(cfg->common.phy, &frame, true, "lk-set-zero"); + motor_can_sched_send_with_priority(cfg->common.phy, &frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, "lk-set-zero"); break; case CLEAR_ERROR: frame.data[0] = LK_CMD_CLEAR_ERR; // 0x9B - motor_can_sched_send_prio(cfg->common.phy, &frame, true, "lk-clear-error"); + motor_can_sched_send_with_priority(cfg->common.phy, &frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, "lk-clear-error"); break; case CLEAR_CONTROLLER: break; @@ -360,8 +364,9 @@ void lk_tx_data_handler(struct k_work *work) tx_frame.dlc = 8; tx_frame.flags = 0; tx_frame.data[0] = LK_CMD_MOTOR_RUN; - motor_can_sched_send_prio(cfg->common.phy, &tx_frame, true, - "lk-retry-enable"); + motor_can_sched_send_with_priority( + cfg->common.phy, &tx_frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, "lk-retry-enable"); } continue; } @@ -371,8 +376,9 @@ void lk_tx_data_handler(struct k_work *work) tx_frame.dlc = 8; tx_frame.flags = 0; tx_frame.data[0] = LK_CMD_MOTOR_RUN; - motor_can_sched_send_prio(cfg->common.phy, &tx_frame, true, - "lk-reenable"); + motor_can_sched_send_with_priority(cfg->common.phy, &tx_frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, + "lk-reenable"); data->params_update[0] = true; data->params_update[1] = true; data->params_update[2] = true; @@ -420,8 +426,9 @@ void lk_tx_params_data_handler(struct k_work *work) int16_t kd_val = (int16_t)(data->params[0].k_d); tx_frame.data[6] = kd_val & 0xFF; tx_frame.data[7] = (kd_val >> 8) & 0xFF; - motor_can_sched_send_prio(cfg->common.phy, &tx_frame, true, - "lk-param-angle"); + motor_can_sched_send_with_priority(cfg->common.phy, &tx_frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, + "lk-param-angle"); } k_sleep(K_USEC(120)); if (data->params_update[1]) { @@ -441,8 +448,9 @@ void lk_tx_params_data_handler(struct k_work *work) int16_t kd_val = (int16_t)(data->params[1].k_d); tx_frame.data[6] = kd_val & 0xFF; tx_frame.data[7] = (kd_val >> 8) & 0xFF; - motor_can_sched_send_prio(cfg->common.phy, &tx_frame, true, - "lk-param-speed"); + motor_can_sched_send_with_priority(cfg->common.phy, &tx_frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, + "lk-param-speed"); } k_sleep(K_USEC(120)); if (data->params_update[2]) { @@ -462,8 +470,9 @@ void lk_tx_params_data_handler(struct k_work *work) int16_t kd_val = (int16_t)(data->params[2].k_d); tx_frame.data[6] = kd_val & 0xFF; tx_frame.data[7] = (kd_val >> 8) & 0xFF; - motor_can_sched_send_prio(cfg->common.phy, &tx_frame, true, - "lk-param-torque"); + motor_can_sched_send_with_priority(cfg->common.phy, &tx_frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, + "lk-param-torque"); } k_sleep(K_USEC(120)); } diff --git a/drivers/motor/mi/motor_mi.c b/drivers/motor/mi/motor_mi.c index 63b5a3b..c1fdaa8 100644 --- a/drivers/motor/mi/motor_mi.c +++ b/drivers/motor/mi/motor_mi.c @@ -75,12 +75,14 @@ void mi_motor_control(const struct device *dev, enum motor_cmd cmd) switch (cmd) { case ENABLE_MOTOR: mi_can_id->mi_msg_mode = Communication_Type_MotorEnable; - motor_can_sched_send_prio(cfg->common.phy, &frame, true, "mi-enable"); + motor_can_sched_send_with_priority(cfg->common.phy, &frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, "mi-enable"); motor_link_request_enable(&data->common.link); break; case DISABLE_MOTOR: mi_can_id->mi_msg_mode = Communication_Type_MotorStop; - motor_can_sched_send_prio(cfg->common.phy, &frame, true, "mi-disable"); + motor_can_sched_send_with_priority(cfg->common.phy, &frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, "mi-disable"); motor_link_request_disable(&data->common.link); break; case SET_ZERO: @@ -88,7 +90,8 @@ void mi_motor_control(const struct device *dev, enum motor_cmd cmd) frame.data[0] = 0x01; data->delta_deg_sum = 0; data->common.angle = 0; - motor_can_sched_send_prio(cfg->common.phy, &frame, true, "mi-set-zero"); + motor_can_sched_send_with_priority(cfg->common.phy, &frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, "mi-set-zero"); break; case CLEAR_CONTROLLER: @@ -221,7 +224,8 @@ static int mi_apply_controller_mode(const struct device *dev, enum motor_mode mo memcpy(&frame.data[0], &index, 2); frame.data[4] = (uint8_t)mode; - motor_can_sched_send_prio(cfg->common.phy, &frame, true, "mi-set-mode"); + motor_can_sched_send_with_priority(cfg->common.phy, &frame, MOTOR_CAN_SCHED_PRIO_CRITICAL, + "mi-set-mode"); for (int i = 0; i < motor_get_controller_count(dev); i++) { const struct motor_controller_config *ctrl_cfg = &cfg->common.controllers[i]; @@ -347,8 +351,9 @@ void mi_tx_data_handler(struct k_work *work) 0x1F00FF00, 5U, "mi-control"); if ((data->common.mode == PV) || (data->common.mode == VO)) { - motor_can_sched_send_prio(cfg->common.phy, &tx_frame[1], true, - "mi-follow"); + motor_can_sched_send_with_priority(cfg->common.phy, &tx_frame[1], + MOTOR_CAN_SCHED_PRIO_CRITICAL, + "mi-follow"); } } if (i % 2 == 1) { diff --git a/drivers/motor/robstride/motor_rs.c b/drivers/motor/robstride/motor_rs.c index 2ff15a4..dd5125e 100644 --- a/drivers/motor/robstride/motor_rs.c +++ b/drivers/motor/robstride/motor_rs.c @@ -158,12 +158,14 @@ static void rs_send_enable_frame(const struct device *dev, const char *tag_prefi frame.id = rs_pack_ext_id(Communication_Type_MotorStop, cfg->common.rx_id & 0xFF, cfg->common.tx_id & 0xFF, 0); frame.data[0] = 0x01; - motor_can_sched_send_prio(cfg->common.phy, &frame, true, tag_prefix); + motor_can_sched_send_with_priority(cfg->common.phy, &frame, MOTOR_CAN_SCHED_PRIO_CRITICAL, + tag_prefix); frame.id = rs_pack_ext_id(Communication_Type_MotorEnable, cfg->common.rx_id & 0xFF, cfg->common.tx_id & 0xFF, 0); frame.data[0] = 0x0; - motor_can_sched_send_prio(cfg->common.phy, &frame, true, "rs-enable"); + motor_can_sched_send_with_priority(cfg->common.phy, &frame, MOTOR_CAN_SCHED_PRIO_CRITICAL, + "rs-enable"); } static void rs_send_auto_report_frame(const struct device *dev, const char *tag) @@ -177,7 +179,8 @@ static void rs_send_auto_report_frame(const struct device *dev, const char *tag) frame.id = rs_pack_ext_id(Communication_Type_MotorReport, cfg->common.rx_id & 0xFF, cfg->common.tx_id & 0xFF, 0); - motor_can_sched_send_prio(cfg->common.phy, &frame, true, tag); + motor_can_sched_send_with_priority(cfg->common.phy, &frame, MOTOR_CAN_SCHED_PRIO_CRITICAL, + tag); } int rs_init(const struct device *dev) @@ -241,7 +244,8 @@ void rs_motor_control(const struct device *dev, enum motor_cmd cmd) case DISABLE_MOTOR: frame.id = rs_pack_ext_id(Communication_Type_MotorStop, cfg->common.rx_id & 0xFF, cfg->common.tx_id & 0xFF, 0); - motor_can_sched_send_prio(cfg->common.phy, &frame, true, "rs-disable"); + motor_can_sched_send_with_priority(cfg->common.phy, &frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, "rs-disable"); motor_link_request_disable(&data->common.link); break; case SET_ZERO: @@ -249,13 +253,15 @@ void rs_motor_control(const struct device *dev, enum motor_cmd cmd) cfg->common.tx_id & 0xFF, 0); frame.data[0] = 0x01; data->common.angle = 0; - motor_can_sched_send_prio(cfg->common.phy, &frame, true, "rs-set-zero"); + motor_can_sched_send_with_priority(cfg->common.phy, &frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, "rs-set-zero"); break; case CLEAR_ERROR: frame.id = rs_pack_ext_id(Communication_Type_MotorStop, cfg->common.rx_id & 0xFF, cfg->common.tx_id & 0xFF, 0); frame.data[0] = 0x01; - motor_can_sched_send_prio(cfg->common.phy, &frame, true, "rs-clear-error"); + motor_can_sched_send_with_priority(cfg->common.phy, &frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, "rs-clear-error"); motor_link_request_disable(&data->common.link); break; case CLEAR_CONTROLLER: @@ -323,7 +329,8 @@ static int rs_send_control_frame(const struct device *dev) } rs_motor_pack(dev, &tx_frame); - ret = motor_can_sched_send_prio(cfg->common.phy, &tx_frame, true, "rs-control"); + ret = motor_can_sched_send_with_priority(cfg->common.phy, &tx_frame, + MOTOR_CAN_SCHED_PRIO_CRITICAL, "rs-control"); if (ret < 0) { motor_stats_inc(MOTOR_STAT_TX_ERROR); } diff --git a/drivers/motor/vesc/motor_vesc.c b/drivers/motor/vesc/motor_vesc.c index 5e23c36..e9dd118 100644 --- a/drivers/motor/vesc/motor_vesc.c +++ b/drivers/motor/vesc/motor_vesc.c @@ -109,7 +109,8 @@ static void vesc_send_ping(const struct device *dev) frame.dlc = 1; frame.data[0] = cfg->common.rx_id & 0xFF; - motor_can_sched_send_prio(cfg->common.phy, &frame, true, "vesc-ping"); + motor_can_sched_send_with_priority(cfg->common.phy, &frame, MOTOR_CAN_SCHED_PRIO_CRITICAL, + "vesc-ping"); } static int vesc_send_control(const struct device *dev) diff --git a/include/ares/protocol/mqttlite/mqttlite_protocol.h b/include/ares/protocol/mqttlite/mqttlite_protocol.h new file mode 100644 index 0000000..8d02e9c --- /dev/null +++ b/include/ares/protocol/mqttlite/mqttlite_protocol.h @@ -0,0 +1,163 @@ +#ifndef ARES_MQTTLITE_PROTOCOL_H +#define ARES_MQTTLITE_PROTOCOL_H + +#include +#include +#include +#include + +#include + +#include + +#ifdef __cplusplus +extern "C" { +#endif + +enum ares_mqttlite_qos { + ARES_MQTTLITE_QOS0 = 0, + ARES_MQTTLITE_QOS1 = 1, + ARES_MQTTLITE_QOS2 = 2, +}; + +enum ares_mqttlite_publish_status { + ARES_MQTTLITE_PUBLISH_ACKED = 0, + ARES_MQTTLITE_PUBLISH_TIMEOUT = -ETIMEDOUT, +}; + +typedef void (*ares_mqttlite_msg_cb_t)(struct AresProtocol *protocol, const char *topic, + const uint8_t *payload, uint16_t payload_len, + enum ares_mqttlite_qos qos, void *user_data); + +typedef void (*ares_mqttlite_publish_cb_t)(struct AresProtocol *protocol, uint16_t packet_id, + int status, void *user_data); + +struct ares_mqttlite_publish_buffer { + struct net_buf *buf; + uint8_t *payload; + uint16_t payload_len; +}; + +int ares_mqttlite_register_topic(struct AresProtocol *protocol, uint16_t topic_id, + const char *topic); + +int ares_mqttlite_subscribe(struct AresProtocol *protocol, const char *topic_filter, + ares_mqttlite_msg_cb_t cb, void *user_data); + +int ares_mqttlite_subscribe_id(struct AresProtocol *protocol, uint16_t topic_id, + ares_mqttlite_msg_cb_t cb, void *user_data); + +int ares_mqttlite_unsubscribe(struct AresProtocol *protocol, const char *topic_filter); + +int ares_mqttlite_unsubscribe_id(struct AresProtocol *protocol, uint16_t topic_id); + +int ares_mqttlite_publish(struct AresProtocol *protocol, const char *topic, const uint8_t *payload, + uint16_t payload_len, enum ares_mqttlite_qos qos, + ares_mqttlite_publish_cb_t cb, void *user_data); + +int ares_mqttlite_publish_id(struct AresProtocol *protocol, uint16_t topic_id, + const uint8_t *payload, uint16_t payload_len, + enum ares_mqttlite_qos qos, ares_mqttlite_publish_cb_t cb, + void *user_data); + +int ares_mqttlite_publish_prepare_id(struct AresProtocol *protocol, uint16_t topic_id, + uint16_t payload_len, + struct ares_mqttlite_publish_buffer *pub); + +int ares_mqttlite_publish_commit(struct AresProtocol *protocol, + struct ares_mqttlite_publish_buffer *pub); + +void ares_mqttlite_publish_abort(struct ares_mqttlite_publish_buffer *pub); + +int ares_mqttlite_ping(struct AresProtocol *protocol); + +void ares_mqttlite_protocol_handle(struct AresProtocol *protocol, struct net_buf *buf); +void ares_mqttlite_protocol_handle_byte(struct AresProtocol *protocol, uint8_t byte); +void ares_mqttlite_protocol_event(struct AresProtocol *protocol, enum AresProtocolEvent event); +int ares_mqttlite_protocol_init(struct AresProtocol *protocol); + +struct ares_mqttlite_subscription { + bool use_topic_id; + uint16_t topic_id; + char topic_filter[CONFIG_ARES_MQTTLITE_MAX_TOPIC_LEN + 1]; + ares_mqttlite_msg_cb_t cb; + void *user_data; +}; + +struct ares_mqttlite_topic_entry { + bool used; + uint16_t topic_id; + char topic[CONFIG_ARES_MQTTLITE_MAX_TOPIC_LEN + 1]; +}; + +enum ares_mqttlite_tx_state { + ARES_MQTTLITE_TX_UNUSED, + ARES_MQTTLITE_TX_WAIT_PUBACK, + ARES_MQTTLITE_TX_WAIT_PUBREC, + ARES_MQTTLITE_TX_WAIT_PUBCOMP, +}; + +struct ares_mqttlite_tx_inflight { + enum ares_mqttlite_tx_state state; + uint16_t packet_id; + enum ares_mqttlite_qos qos; + uint8_t retries; + int64_t last_send_ms; + ares_mqttlite_publish_cb_t cb; + void *user_data; + bool use_topic_id; + uint16_t topic_id; + uint16_t topic_len; + uint16_t payload_len; + char topic[CONFIG_ARES_MQTTLITE_MAX_TOPIC_LEN + 1]; + uint8_t payload[CONFIG_ARES_MQTTLITE_MAX_PAYLOAD_SIZE]; +}; + +struct ares_mqttlite_rx_qos2 { + bool used; + uint16_t packet_id; + bool use_topic_id; + uint16_t topic_id; + uint16_t topic_len; + uint16_t payload_len; + char topic[CONFIG_ARES_MQTTLITE_MAX_TOPIC_LEN + 1]; + uint8_t payload[CONFIG_ARES_MQTTLITE_MAX_PAYLOAD_SIZE]; +}; + +struct ares_mqttlite_protocol_data { + const char *name; + bool online; + uint16_t next_packet_id; + struct k_mutex lock; + struct k_work_delayable retry_work; + struct AresProtocol *protocol; + struct ares_mqttlite_topic_entry topics[CONFIG_ARES_MQTTLITE_MAX_TOPICS]; + struct ares_mqttlite_subscription subscriptions[CONFIG_ARES_MQTTLITE_MAX_SUBSCRIPTIONS]; + struct ares_mqttlite_tx_inflight tx[CONFIG_ARES_MQTTLITE_MAX_INFLIGHT]; + struct ares_mqttlite_rx_qos2 rx_qos2[CONFIG_ARES_MQTTLITE_MAX_INFLIGHT]; + uint8_t rx_buf[CONFIG_ARES_MQTTLITE_MAX_FRAME_SIZE]; + uint16_t rx_pos; + uint16_t rx_expected; +}; + +#define ARES_MQTTLITE_PROTOCOL_DEFINE(Protocol_name) \ + struct AresProtocolAPI Protocol_name##_api = { \ + .handle = ares_mqttlite_protocol_handle, \ + .handle_byte = ares_mqttlite_protocol_handle_byte, \ + .event = ares_mqttlite_protocol_event, \ + .init = ares_mqttlite_protocol_init, \ + }; \ + struct ares_mqttlite_protocol_data Protocol_name##_data = { \ + .name = #Protocol_name, \ + }; \ + struct AresProtocol Protocol_name = { \ + .name = #Protocol_name, \ + .api = &Protocol_name##_api, \ + .priv_data = &Protocol_name##_data, \ + } + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/lib/ares/protocol/CMakeLists.txt b/lib/ares/protocol/CMakeLists.txt index e3005f8..5f44ded 100755 --- a/lib/ares/protocol/CMakeLists.txt +++ b/lib/ares/protocol/CMakeLists.txt @@ -1,2 +1,3 @@ add_subdirectory_ifdef(CONFIG_DUAL_PROPOSE_PROTOCOL dual) add_subdirectory_ifdef(CONFIG_ARES_PLOTTER_PROTOCOL plotter) +add_subdirectory_ifdef(CONFIG_ARES_MQTTLITE_PROTOCOL mqttlite) diff --git a/lib/ares/protocol/Kconfig b/lib/ares/protocol/Kconfig index 72a1673..f74c6f5 100755 --- a/lib/ares/protocol/Kconfig +++ b/lib/ares/protocol/Kconfig @@ -7,3 +7,4 @@ config DUAL_PROPOSE_PROTOCOL select ARES_COMM_LIB rsource "plotter/Kconfig" +rsource "mqttlite/Kconfig" diff --git a/lib/ares/protocol/mqttlite/CMakeLists.txt b/lib/ares/protocol/mqttlite/CMakeLists.txt new file mode 100644 index 0000000..bcd1997 --- /dev/null +++ b/lib/ares/protocol/mqttlite/CMakeLists.txt @@ -0,0 +1,2 @@ +zephyr_library() +zephyr_library_sources(mqttlite_protocol.c) diff --git a/lib/ares/protocol/mqttlite/Kconfig b/lib/ares/protocol/mqttlite/Kconfig new file mode 100644 index 0000000..6308b07 --- /dev/null +++ b/lib/ares/protocol/mqttlite/Kconfig @@ -0,0 +1,55 @@ +config ARES_MQTTLITE_PROTOCOL + bool "ARES MQTT-like protocol" + select ARES_COMM_LIB + select NET_BUF + help + Enable a small MQTT-like publish/subscribe protocol for ARES interfaces. + +if ARES_MQTTLITE_PROTOCOL + +config ARES_MQTTLITE_MAX_TOPIC_LEN + int "Maximum topic/filter length" + range 1 255 + default 64 + +config ARES_MQTTLITE_MAX_PAYLOAD_SIZE + int "Maximum payload size" + range 1 65535 + default 192 + +config ARES_MQTTLITE_MAX_FRAME_SIZE + int "Maximum received frame size" + range 16 65535 + default 272 + +config ARES_MQTTLITE_MAX_SUBSCRIPTIONS + int "Maximum local subscriptions" + range 1 32 + default 8 + +config ARES_MQTTLITE_MAX_TOPICS + int "Maximum registered topic ids" + range 1 64 + default 8 + +config ARES_MQTTLITE_MAX_INFLIGHT + int "Maximum QoS inflight packets" + range 1 32 + default 4 + +config ARES_MQTTLITE_RETRY_INTERVAL_MS + int "QoS retry interval in milliseconds" + range 10 60000 + default 200 + +config ARES_MQTTLITE_MAX_RETRIES + int "Maximum QoS retries before timeout" + range 1 255 + default 5 + +config ARES_MQTTLITE_LOG_LEVEL + int "ARES MQTT-like protocol log level" + range 0 4 + default 3 + +endif diff --git a/lib/ares/protocol/mqttlite/mqttlite_protocol.c b/lib/ares/protocol/mqttlite/mqttlite_protocol.c new file mode 100644 index 0000000..dc9e8e9 --- /dev/null +++ b/lib/ares/protocol/mqttlite/mqttlite_protocol.c @@ -0,0 +1,1029 @@ +#include +#include +#include +#include + +#include +#include +#include +#include + +#include +#include + +LOG_MODULE_REGISTER(ares_mqttlite, CONFIG_ARES_MQTTLITE_LOG_LEVEL); + +#define MQTL_MAGIC 0x4d51 +#define MQTL_VERSION 1 + +#define MQTL_MAGIC_IDX 0 +#define MQTL_VERSION_IDX 2 +#define MQTL_TYPE_IDX 3 +#define MQTL_FLAGS_IDX 4 +#define MQTL_RESERVED_IDX 5 +#define MQTL_PACKET_ID_IDX 6 +#define MQTL_TOPIC_LEN_IDX 8 +#define MQTL_PAYLOAD_LEN_IDX 10 +#define MQTL_HEADER_LEN 12 + +#define MQTL_FLAG_QOS_MASK 0x03 +#define MQTL_FLAG_DUP BIT(2) +#define MQTL_FLAG_TOPIC_ID BIT(3) + +#define MQTL_TOPIC_ID_LEN 2 + +enum mqttlite_frame_type { + MQTL_FRAME_PUBLISH = 1, + MQTL_FRAME_PUBACK = 2, + MQTL_FRAME_PUBREC = 3, + MQTL_FRAME_PUBREL = 4, + MQTL_FRAME_PUBCOMP = 5, + MQTL_FRAME_PING = 6, + MQTL_FRAME_PONG = 7, +}; + +static uint16_t frame_topic_len(const uint8_t *frame) +{ + return sys_get_le16(&frame[MQTL_TOPIC_LEN_IDX]); +} + +static uint16_t frame_payload_len(const uint8_t *frame) +{ + return sys_get_le16(&frame[MQTL_PAYLOAD_LEN_IDX]); +} + +static uint16_t frame_packet_id(const uint8_t *frame) +{ + return sys_get_le16(&frame[MQTL_PACKET_ID_IDX]); +} + +static enum ares_mqttlite_qos frame_qos(const uint8_t *frame) +{ + return frame[MQTL_FLAGS_IDX] & MQTL_FLAG_QOS_MASK; +} + +static bool frame_uses_topic_id(const uint8_t *frame) +{ + return (frame[MQTL_FLAGS_IDX] & MQTL_FLAG_TOPIC_ID) != 0; +} + +static bool valid_qos(enum ares_mqttlite_qos qos) +{ + return qos == ARES_MQTTLITE_QOS0 || qos == ARES_MQTTLITE_QOS1 || qos == ARES_MQTTLITE_QOS2; +} + +static const char *find_topic_name_locked(struct ares_mqttlite_protocol_data *data, + uint16_t topic_id) +{ + for (int i = 0; i < CONFIG_ARES_MQTTLITE_MAX_TOPICS; i++) { + if (data->topics[i].used && data->topics[i].topic_id == topic_id) { + return data->topics[i].topic; + } + } + + return NULL; +} + +static bool topic_matches(const char *filter, const char *topic) +{ + const char *fp = filter; + const char *tp = topic; + + if (strcmp(filter, "#") == 0) { + return true; + } + + while (*fp != '\0' && *tp != '\0') { + if (*fp == '#') { + return fp[1] == '\0' && (fp == filter || fp[-1] == '/'); + } + if (*fp == '+') { + while (*tp != '\0' && *tp != '/') { + tp++; + } + fp++; + continue; + } + if (*fp != *tp) { + return false; + } + fp++; + tp++; + } + + return *fp == *tp || (strcmp(fp, "/#") == 0); +} + +static int alloc_packet_id_locked(struct ares_mqttlite_protocol_data *data) +{ + uint16_t packet_id = data->next_packet_id; + + if (packet_id == 0) { + packet_id = 1; + } + data->next_packet_id = packet_id + 1; + if (data->next_packet_id == 0) { + data->next_packet_id = 1; + } + + return packet_id; +} + +static struct ares_mqttlite_tx_inflight *find_tx_locked(struct ares_mqttlite_protocol_data *data, + uint16_t packet_id) +{ + for (int i = 0; i < CONFIG_ARES_MQTTLITE_MAX_INFLIGHT; i++) { + if (data->tx[i].state != ARES_MQTTLITE_TX_UNUSED && + data->tx[i].packet_id == packet_id) { + return &data->tx[i]; + } + } + + return NULL; +} + +static struct ares_mqttlite_tx_inflight *alloc_tx_locked(struct ares_mqttlite_protocol_data *data) +{ + for (int i = 0; i < CONFIG_ARES_MQTTLITE_MAX_INFLIGHT; i++) { + if (data->tx[i].state == ARES_MQTTLITE_TX_UNUSED) { + return &data->tx[i]; + } + } + + return NULL; +} + +static struct ares_mqttlite_rx_qos2 *find_rx_qos2_locked(struct ares_mqttlite_protocol_data *data, + uint16_t packet_id) +{ + for (int i = 0; i < CONFIG_ARES_MQTTLITE_MAX_INFLIGHT; i++) { + if (data->rx_qos2[i].used && data->rx_qos2[i].packet_id == packet_id) { + return &data->rx_qos2[i]; + } + } + + return NULL; +} + +static struct ares_mqttlite_rx_qos2 *alloc_rx_qos2_locked(struct ares_mqttlite_protocol_data *data) +{ + for (int i = 0; i < CONFIG_ARES_MQTTLITE_MAX_INFLIGHT; i++) { + if (!data->rx_qos2[i].used) { + return &data->rx_qos2[i]; + } + } + + return NULL; +} + +static int send_frame(struct AresProtocol *protocol, enum mqttlite_frame_type type, + enum ares_mqttlite_qos qos, uint8_t extra_flags, uint16_t packet_id, + bool use_topic_id, uint16_t topic_id, const char *topic, uint16_t topic_len, + const uint8_t *payload, uint16_t payload_len) +{ + struct AresInterface *interface = protocol->interface; + struct net_buf *buf; + uint16_t wire_topic_len = use_topic_id ? MQTL_TOPIC_ID_LEN : topic_len; + size_t frame_len = MQTL_HEADER_LEN + wire_topic_len + payload_len; + + if (interface == NULL || interface->api == NULL || interface->api->send == NULL || + interface->api->alloc_buf == NULL) { + return -ENODEV; + } + if (frame_len > CONFIG_ARES_MQTTLITE_MAX_FRAME_SIZE) { + return -EMSGSIZE; + } + + buf = interface->api->alloc_buf(interface); + if (buf == NULL) { + return -ENOMEM; + } + if (net_buf_tailroom(buf) < frame_len) { + net_buf_unref(buf); + return -EMSGSIZE; + } + + net_buf_add(buf, frame_len); + sys_put_le16(MQTL_MAGIC, &buf->data[MQTL_MAGIC_IDX]); + buf->data[MQTL_VERSION_IDX] = MQTL_VERSION; + buf->data[MQTL_TYPE_IDX] = type; + buf->data[MQTL_FLAGS_IDX] = + (qos & MQTL_FLAG_QOS_MASK) | extra_flags | (use_topic_id ? MQTL_FLAG_TOPIC_ID : 0); + buf->data[MQTL_RESERVED_IDX] = 0; + sys_put_le16(packet_id, &buf->data[MQTL_PACKET_ID_IDX]); + sys_put_le16(wire_topic_len, &buf->data[MQTL_TOPIC_LEN_IDX]); + sys_put_le16(payload_len, &buf->data[MQTL_PAYLOAD_LEN_IDX]); + + if (use_topic_id) { + sys_put_le16(topic_id, &buf->data[MQTL_HEADER_LEN]); + } else if (topic_len > 0) { + memcpy(&buf->data[MQTL_HEADER_LEN], topic, topic_len); + } + if (payload_len > 0) { + memcpy(&buf->data[MQTL_HEADER_LEN + wire_topic_len], payload, payload_len); + } + + return interface->api->send(interface, buf); +} + +static int send_ack(struct AresProtocol *protocol, enum mqttlite_frame_type type, + uint16_t packet_id) +{ + return send_frame(protocol, type, ARES_MQTTLITE_QOS0, 0, packet_id, false, 0, NULL, 0, NULL, + 0); +} + +static void schedule_retry(struct ares_mqttlite_protocol_data *data) +{ + k_work_schedule(&data->retry_work, K_MSEC(CONFIG_ARES_MQTTLITE_RETRY_INTERVAL_MS)); +} + +static void dispatch_publish(struct AresProtocol *protocol, const char *topic, bool use_topic_id, + uint16_t topic_id, const uint8_t *payload, uint16_t payload_len, + enum ares_mqttlite_qos qos) +{ + struct ares_mqttlite_protocol_data *data = protocol->priv_data; + ares_mqttlite_msg_cb_t callbacks[CONFIG_ARES_MQTTLITE_MAX_SUBSCRIPTIONS]; + void *user_data[CONFIG_ARES_MQTTLITE_MAX_SUBSCRIPTIONS]; + int count = 0; + + k_mutex_lock(&data->lock, K_FOREVER); + for (int i = 0; i < CONFIG_ARES_MQTTLITE_MAX_SUBSCRIPTIONS; i++) { + if (data->subscriptions[i].cb == NULL) { + continue; + } + if (data->subscriptions[i].use_topic_id) { + if (!use_topic_id || data->subscriptions[i].topic_id != topic_id) { + continue; + } + } else if (!topic_matches(data->subscriptions[i].topic_filter, topic)) { + continue; + } + { + callbacks[count] = data->subscriptions[i].cb; + user_data[count] = data->subscriptions[i].user_data; + count++; + } + } + k_mutex_unlock(&data->lock); + + for (int i = 0; i < count; i++) { + callbacks[i](protocol, topic, payload, payload_len, qos, user_data[i]); + } +} + +static void clear_tx(struct ares_mqttlite_tx_inflight *tx) +{ + memset(tx, 0, sizeof(*tx)); + tx->state = ARES_MQTTLITE_TX_UNUSED; +} + +static void complete_tx(struct AresProtocol *protocol, uint16_t packet_id, int status) +{ + struct ares_mqttlite_protocol_data *data = protocol->priv_data; + ares_mqttlite_publish_cb_t cb = NULL; + void *user_data = NULL; + + k_mutex_lock(&data->lock, K_FOREVER); + struct ares_mqttlite_tx_inflight *tx = find_tx_locked(data, packet_id); + if (tx != NULL) { + cb = tx->cb; + user_data = tx->user_data; + clear_tx(tx); + } + k_mutex_unlock(&data->lock); + + if (cb != NULL) { + cb(protocol, packet_id, status, user_data); + } +} + +static void handle_publish(struct AresProtocol *protocol, const uint8_t *frame) +{ + struct ares_mqttlite_protocol_data *data = protocol->priv_data; + uint16_t packet_id = frame_packet_id(frame); + uint16_t topic_len = frame_topic_len(frame); + uint16_t payload_len = frame_payload_len(frame); + enum ares_mqttlite_qos qos = frame_qos(frame); + bool use_topic_id = frame_uses_topic_id(frame); + uint16_t topic_id = 0; + char topic[CONFIG_ARES_MQTTLITE_MAX_TOPIC_LEN + 1]; + const uint8_t *payload = &frame[MQTL_HEADER_LEN + topic_len]; + + topic[0] = '\0'; + + if (!valid_qos(qos) || topic_len == 0 || + payload_len > CONFIG_ARES_MQTTLITE_MAX_PAYLOAD_SIZE) { + LOG_WRN("%s rejected malformed publish", data->name); + return; + } + if (use_topic_id && topic_len != MQTL_TOPIC_ID_LEN) { + LOG_WRN("%s rejected malformed topic-id publish", data->name); + return; + } + if (!use_topic_id && topic_len > CONFIG_ARES_MQTTLITE_MAX_TOPIC_LEN) { + LOG_WRN("%s rejected oversized topic publish", data->name); + return; + } + if (qos != ARES_MQTTLITE_QOS0 && packet_id == 0) { + LOG_WRN("%s rejected QoS publish without packet id", data->name); + return; + } + + if (use_topic_id) { + const char *registered_topic; + + topic_id = sys_get_le16(&frame[MQTL_HEADER_LEN]); + k_mutex_lock(&data->lock, K_FOREVER); + registered_topic = find_topic_name_locked(data, topic_id); + if (registered_topic != NULL) { + strncpy(topic, registered_topic, sizeof(topic) - 1); + topic[sizeof(topic) - 1] = '\0'; + } + k_mutex_unlock(&data->lock); + } else { + memcpy(topic, &frame[MQTL_HEADER_LEN], topic_len); + topic[topic_len] = '\0'; + } + + if (qos == ARES_MQTTLITE_QOS0) { + dispatch_publish(protocol, topic, use_topic_id, topic_id, payload, payload_len, + qos); + return; + } + + if (qos == ARES_MQTTLITE_QOS1) { + dispatch_publish(protocol, topic, use_topic_id, topic_id, payload, payload_len, + qos); + (void)send_ack(protocol, MQTL_FRAME_PUBACK, packet_id); + return; + } + + k_mutex_lock(&data->lock, K_FOREVER); + struct ares_mqttlite_rx_qos2 *rx = find_rx_qos2_locked(data, packet_id); + if (rx == NULL) { + rx = alloc_rx_qos2_locked(data); + if (rx != NULL) { + rx->used = true; + rx->packet_id = packet_id; + rx->use_topic_id = use_topic_id; + rx->topic_id = topic_id; + rx->topic_len = strlen(topic); + rx->payload_len = payload_len; + memcpy(rx->topic, topic, rx->topic_len + 1); + memcpy(rx->payload, payload, payload_len); + } + } + k_mutex_unlock(&data->lock); + + if (rx == NULL) { + LOG_WRN("%s has no room for QoS2 packet %u", data->name, packet_id); + return; + } + + (void)send_ack(protocol, MQTL_FRAME_PUBREC, packet_id); +} + +static void handle_pubrel(struct AresProtocol *protocol, uint16_t packet_id) +{ + struct ares_mqttlite_protocol_data *data = protocol->priv_data; + char topic[CONFIG_ARES_MQTTLITE_MAX_TOPIC_LEN + 1]; + uint8_t payload[CONFIG_ARES_MQTTLITE_MAX_PAYLOAD_SIZE]; + uint16_t payload_len = 0; + bool use_topic_id = false; + uint16_t topic_id = 0; + bool found = false; + + k_mutex_lock(&data->lock, K_FOREVER); + struct ares_mqttlite_rx_qos2 *rx = find_rx_qos2_locked(data, packet_id); + if (rx != NULL) { + memcpy(topic, rx->topic, rx->topic_len + 1); + memcpy(payload, rx->payload, rx->payload_len); + payload_len = rx->payload_len; + use_topic_id = rx->use_topic_id; + topic_id = rx->topic_id; + memset(rx, 0, sizeof(*rx)); + found = true; + } + k_mutex_unlock(&data->lock); + + if (found) { + dispatch_publish(protocol, topic, use_topic_id, topic_id, payload, payload_len, + ARES_MQTTLITE_QOS2); + } + (void)send_ack(protocol, MQTL_FRAME_PUBCOMP, packet_id); +} + +static void handle_ack_frame(struct AresProtocol *protocol, enum mqttlite_frame_type type, + uint16_t packet_id) +{ + struct ares_mqttlite_protocol_data *data = protocol->priv_data; + + if (packet_id == 0) { + return; + } + + if (type == MQTL_FRAME_PUBACK || type == MQTL_FRAME_PUBCOMP) { + complete_tx(protocol, packet_id, ARES_MQTTLITE_PUBLISH_ACKED); + return; + } + + if (type == MQTL_FRAME_PUBREC) { + bool send_pubrel = false; + + k_mutex_lock(&data->lock, K_FOREVER); + struct ares_mqttlite_tx_inflight *tx = find_tx_locked(data, packet_id); + if (tx != NULL && tx->state == ARES_MQTTLITE_TX_WAIT_PUBREC) { + tx->state = ARES_MQTTLITE_TX_WAIT_PUBCOMP; + tx->retries = 0; + tx->last_send_ms = k_uptime_get(); + send_pubrel = true; + schedule_retry(data); + } + k_mutex_unlock(&data->lock); + + if (send_pubrel) { + (void)send_ack(protocol, MQTL_FRAME_PUBREL, packet_id); + } + } +} + +static void process_frame(struct AresProtocol *protocol, const uint8_t *frame, uint16_t len) +{ + uint16_t topic_len; + uint16_t payload_len; + + if (len < MQTL_HEADER_LEN || sys_get_le16(&frame[MQTL_MAGIC_IDX]) != MQTL_MAGIC || + frame[MQTL_VERSION_IDX] != MQTL_VERSION) { + return; + } + + topic_len = frame_topic_len(frame); + payload_len = frame_payload_len(frame); + if ((uint32_t)MQTL_HEADER_LEN + topic_len + payload_len != len) { + return; + } + + switch (frame[MQTL_TYPE_IDX]) { + case MQTL_FRAME_PUBLISH: + handle_publish(protocol, frame); + break; + case MQTL_FRAME_PUBACK: + case MQTL_FRAME_PUBREC: + case MQTL_FRAME_PUBCOMP: + handle_ack_frame(protocol, frame[MQTL_TYPE_IDX], frame_packet_id(frame)); + break; + case MQTL_FRAME_PUBREL: + handle_pubrel(protocol, frame_packet_id(frame)); + break; + case MQTL_FRAME_PING: + (void)send_ack(protocol, MQTL_FRAME_PONG, 0); + break; + case MQTL_FRAME_PONG: + break; + default: + break; + } +} + +static void retry_work_handler(struct k_work *work) +{ + struct k_work_delayable *dwork = k_work_delayable_from_work(work); + struct ares_mqttlite_protocol_data *data = + CONTAINER_OF(dwork, struct ares_mqttlite_protocol_data, retry_work); + struct AresProtocol *protocol = data->protocol; + int64_t now = k_uptime_get(); + bool has_pending = false; + + k_mutex_lock(&data->lock, K_FOREVER); + for (int i = 0; i < CONFIG_ARES_MQTTLITE_MAX_INFLIGHT; i++) { + struct ares_mqttlite_tx_inflight *tx = &data->tx[i]; + enum ares_mqttlite_tx_state state; + enum ares_mqttlite_qos qos; + bool use_topic_id; + uint16_t packet_id; + uint16_t topic_id; + uint16_t topic_len; + uint16_t payload_len; + char topic[CONFIG_ARES_MQTTLITE_MAX_TOPIC_LEN + 1]; + uint8_t payload[CONFIG_ARES_MQTTLITE_MAX_PAYLOAD_SIZE]; + + if (tx->state == ARES_MQTTLITE_TX_UNUSED) { + continue; + } + has_pending = true; + if (now - tx->last_send_ms < CONFIG_ARES_MQTTLITE_RETRY_INTERVAL_MS) { + continue; + } + if (tx->retries >= CONFIG_ARES_MQTTLITE_MAX_RETRIES) { + uint16_t packet_id = tx->packet_id; + ares_mqttlite_publish_cb_t cb = tx->cb; + void *user_data = tx->user_data; + + clear_tx(tx); + k_mutex_unlock(&data->lock); + if (cb != NULL) { + cb(protocol, packet_id, ARES_MQTTLITE_PUBLISH_TIMEOUT, user_data); + } + k_mutex_lock(&data->lock, K_FOREVER); + has_pending = false; + for (int j = 0; j < CONFIG_ARES_MQTTLITE_MAX_INFLIGHT; j++) { + if (data->tx[j].state != ARES_MQTTLITE_TX_UNUSED) { + has_pending = true; + break; + } + } + continue; + } + + state = tx->state; + qos = tx->qos; + use_topic_id = tx->use_topic_id; + packet_id = tx->packet_id; + topic_id = tx->topic_id; + topic_len = tx->topic_len; + payload_len = tx->payload_len; + memcpy(topic, tx->topic, topic_len + 1); + memcpy(payload, tx->payload, payload_len); + tx->retries++; + tx->last_send_ms = now; + k_mutex_unlock(&data->lock); + if (state == ARES_MQTTLITE_TX_WAIT_PUBCOMP) { + (void)send_ack(protocol, MQTL_FRAME_PUBREL, packet_id); + } else { + (void)send_frame(protocol, MQTL_FRAME_PUBLISH, qos, MQTL_FLAG_DUP, + packet_id, use_topic_id, topic_id, topic, topic_len, + payload, payload_len); + } + k_mutex_lock(&data->lock, K_FOREVER); + } + k_mutex_unlock(&data->lock); + + if (has_pending) { + schedule_retry(data); + } +} + +int ares_mqttlite_register_topic(struct AresProtocol *protocol, uint16_t topic_id, + const char *topic) +{ + struct ares_mqttlite_protocol_data *data; + size_t topic_len; + int empty = -1; + + if (protocol == NULL || topic_id == 0 || topic == NULL) { + return -EINVAL; + } + + data = protocol->priv_data; + topic_len = strlen(topic); + if (topic_len == 0 || topic_len > CONFIG_ARES_MQTTLITE_MAX_TOPIC_LEN) { + return -EINVAL; + } + + k_mutex_lock(&data->lock, K_FOREVER); + for (int i = 0; i < CONFIG_ARES_MQTTLITE_MAX_TOPICS; i++) { + if (!data->topics[i].used && empty < 0) { + empty = i; + continue; + } + if (data->topics[i].used && data->topics[i].topic_id == topic_id) { + memcpy(data->topics[i].topic, topic, topic_len + 1); + k_mutex_unlock(&data->lock); + return 0; + } + } + if (empty < 0) { + k_mutex_unlock(&data->lock); + return -ENOMEM; + } + + data->topics[empty].used = true; + data->topics[empty].topic_id = topic_id; + memcpy(data->topics[empty].topic, topic, topic_len + 1); + k_mutex_unlock(&data->lock); + + return 0; +} + +int ares_mqttlite_subscribe(struct AresProtocol *protocol, const char *topic_filter, + ares_mqttlite_msg_cb_t cb, void *user_data) +{ + struct ares_mqttlite_protocol_data *data; + size_t topic_len; + int empty = -1; + + if (protocol == NULL || topic_filter == NULL || cb == NULL) { + return -EINVAL; + } + + data = protocol->priv_data; + topic_len = strlen(topic_filter); + if (topic_len == 0 || topic_len > CONFIG_ARES_MQTTLITE_MAX_TOPIC_LEN) { + return -EINVAL; + } + + k_mutex_lock(&data->lock, K_FOREVER); + for (int i = 0; i < CONFIG_ARES_MQTTLITE_MAX_SUBSCRIPTIONS; i++) { + if (data->subscriptions[i].cb == NULL && empty < 0) { + empty = i; + continue; + } + if (!data->subscriptions[i].use_topic_id && + strcmp(data->subscriptions[i].topic_filter, topic_filter) == 0) { + data->subscriptions[i].cb = cb; + data->subscriptions[i].user_data = user_data; + k_mutex_unlock(&data->lock); + return 0; + } + } + if (empty < 0) { + k_mutex_unlock(&data->lock); + return -ENOMEM; + } + + memcpy(data->subscriptions[empty].topic_filter, topic_filter, topic_len + 1); + data->subscriptions[empty].cb = cb; + data->subscriptions[empty].user_data = user_data; + k_mutex_unlock(&data->lock); + + return 0; +} + +int ares_mqttlite_subscribe_id(struct AresProtocol *protocol, uint16_t topic_id, + ares_mqttlite_msg_cb_t cb, void *user_data) +{ + struct ares_mqttlite_protocol_data *data; + int empty = -1; + + if (protocol == NULL || topic_id == 0 || cb == NULL) { + return -EINVAL; + } + + data = protocol->priv_data; + k_mutex_lock(&data->lock, K_FOREVER); + for (int i = 0; i < CONFIG_ARES_MQTTLITE_MAX_SUBSCRIPTIONS; i++) { + if (data->subscriptions[i].cb == NULL && empty < 0) { + empty = i; + continue; + } + if (data->subscriptions[i].use_topic_id && + data->subscriptions[i].topic_id == topic_id) { + data->subscriptions[i].cb = cb; + data->subscriptions[i].user_data = user_data; + k_mutex_unlock(&data->lock); + return 0; + } + } + if (empty < 0) { + k_mutex_unlock(&data->lock); + return -ENOMEM; + } + + data->subscriptions[empty].use_topic_id = true; + data->subscriptions[empty].topic_id = topic_id; + data->subscriptions[empty].topic_filter[0] = '\0'; + data->subscriptions[empty].cb = cb; + data->subscriptions[empty].user_data = user_data; + k_mutex_unlock(&data->lock); + + return 0; +} + +int ares_mqttlite_unsubscribe(struct AresProtocol *protocol, const char *topic_filter) +{ + struct ares_mqttlite_protocol_data *data; + + if (protocol == NULL || topic_filter == NULL) { + return -EINVAL; + } + + data = protocol->priv_data; + k_mutex_lock(&data->lock, K_FOREVER); + for (int i = 0; i < CONFIG_ARES_MQTTLITE_MAX_SUBSCRIPTIONS; i++) { + if (!data->subscriptions[i].use_topic_id && + strcmp(data->subscriptions[i].topic_filter, topic_filter) == 0) { + memset(&data->subscriptions[i], 0, sizeof(data->subscriptions[i])); + k_mutex_unlock(&data->lock); + return 0; + } + } + k_mutex_unlock(&data->lock); + + return -ENOENT; +} + +int ares_mqttlite_unsubscribe_id(struct AresProtocol *protocol, uint16_t topic_id) +{ + struct ares_mqttlite_protocol_data *data; + + if (protocol == NULL || topic_id == 0) { + return -EINVAL; + } + + data = protocol->priv_data; + k_mutex_lock(&data->lock, K_FOREVER); + for (int i = 0; i < CONFIG_ARES_MQTTLITE_MAX_SUBSCRIPTIONS; i++) { + if (data->subscriptions[i].use_topic_id && + data->subscriptions[i].topic_id == topic_id) { + memset(&data->subscriptions[i], 0, sizeof(data->subscriptions[i])); + k_mutex_unlock(&data->lock); + return 0; + } + } + k_mutex_unlock(&data->lock); + + return -ENOENT; +} + +static int publish_common(struct AresProtocol *protocol, bool use_topic_id, uint16_t topic_id, + const char *topic, uint16_t topic_len, const uint8_t *payload, + uint16_t payload_len, enum ares_mqttlite_qos qos, + ares_mqttlite_publish_cb_t cb, void *user_data) +{ + struct ares_mqttlite_protocol_data *data; + uint16_t packet_id = 0; + int ret; + + if (protocol == NULL || !valid_qos(qos)) { + return -EINVAL; + } + if (payload == NULL && payload_len > 0) { + return -EINVAL; + } + + data = protocol->priv_data; + if (use_topic_id) { + if (topic_id == 0) { + return -EINVAL; + } + topic_len = 0; + topic = ""; + } else if (topic == NULL || topic_len == 0 || + topic_len > CONFIG_ARES_MQTTLITE_MAX_TOPIC_LEN) { + return -EINVAL; + } + if (payload_len > CONFIG_ARES_MQTTLITE_MAX_PAYLOAD_SIZE) { + return -EINVAL; + } + + if (qos != ARES_MQTTLITE_QOS0) { + k_mutex_lock(&data->lock, K_FOREVER); + struct ares_mqttlite_tx_inflight *tx = alloc_tx_locked(data); + if (tx == NULL) { + k_mutex_unlock(&data->lock); + return -ENOMEM; + } + packet_id = alloc_packet_id_locked(data); + tx->state = qos == ARES_MQTTLITE_QOS1 ? ARES_MQTTLITE_TX_WAIT_PUBACK + : ARES_MQTTLITE_TX_WAIT_PUBREC; + tx->packet_id = packet_id; + tx->qos = qos; + tx->retries = 0; + tx->last_send_ms = k_uptime_get(); + tx->cb = cb; + tx->user_data = user_data; + tx->use_topic_id = use_topic_id; + tx->topic_id = topic_id; + tx->topic_len = topic_len; + tx->payload_len = payload_len; + memcpy(tx->topic, topic, topic_len + 1); + if (payload_len > 0) { + memcpy(tx->payload, payload, payload_len); + } + k_mutex_unlock(&data->lock); + } + + ret = send_frame(protocol, MQTL_FRAME_PUBLISH, qos, 0, packet_id, use_topic_id, topic_id, + topic, topic_len, payload, payload_len); + if (ret != 0 && qos != ARES_MQTTLITE_QOS0) { + complete_tx(protocol, packet_id, ret); + return ret; + } + + if (qos != ARES_MQTTLITE_QOS0) { + schedule_retry(data); + return packet_id; + } + + return ret; +} + +int ares_mqttlite_publish(struct AresProtocol *protocol, const char *topic, const uint8_t *payload, + uint16_t payload_len, enum ares_mqttlite_qos qos, + ares_mqttlite_publish_cb_t cb, void *user_data) +{ + if (topic == NULL) { + return -EINVAL; + } + + return publish_common(protocol, false, 0, topic, strlen(topic), payload, payload_len, qos, + cb, user_data); +} + +int ares_mqttlite_publish_id(struct AresProtocol *protocol, uint16_t topic_id, + const uint8_t *payload, uint16_t payload_len, + enum ares_mqttlite_qos qos, ares_mqttlite_publish_cb_t cb, + void *user_data) +{ + return publish_common(protocol, true, topic_id, NULL, 0, payload, payload_len, qos, cb, + user_data); +} + +int ares_mqttlite_publish_prepare_id(struct AresProtocol *protocol, uint16_t topic_id, + uint16_t payload_len, struct ares_mqttlite_publish_buffer *pub) +{ + struct AresInterface *interface; + size_t frame_len = MQTL_HEADER_LEN + MQTL_TOPIC_ID_LEN + payload_len; + + if (protocol == NULL || topic_id == 0 || pub == NULL || + payload_len > CONFIG_ARES_MQTTLITE_MAX_PAYLOAD_SIZE || + frame_len > CONFIG_ARES_MQTTLITE_MAX_FRAME_SIZE) { + return -EINVAL; + } + + interface = protocol->interface; + if (interface == NULL || interface->api == NULL || interface->api->send == NULL || + interface->api->alloc_buf == NULL) { + return -ENODEV; + } + + memset(pub, 0, sizeof(*pub)); + pub->buf = interface->api->alloc_buf(interface); + if (pub->buf == NULL) { + return -ENOMEM; + } + if (net_buf_tailroom(pub->buf) < frame_len) { + net_buf_unref(pub->buf); + memset(pub, 0, sizeof(*pub)); + return -EMSGSIZE; + } + + net_buf_add(pub->buf, frame_len); + sys_put_le16(MQTL_MAGIC, &pub->buf->data[MQTL_MAGIC_IDX]); + pub->buf->data[MQTL_VERSION_IDX] = MQTL_VERSION; + pub->buf->data[MQTL_TYPE_IDX] = MQTL_FRAME_PUBLISH; + pub->buf->data[MQTL_FLAGS_IDX] = ARES_MQTTLITE_QOS0 | MQTL_FLAG_TOPIC_ID; + pub->buf->data[MQTL_RESERVED_IDX] = 0; + sys_put_le16(0, &pub->buf->data[MQTL_PACKET_ID_IDX]); + sys_put_le16(MQTL_TOPIC_ID_LEN, &pub->buf->data[MQTL_TOPIC_LEN_IDX]); + sys_put_le16(payload_len, &pub->buf->data[MQTL_PAYLOAD_LEN_IDX]); + sys_put_le16(topic_id, &pub->buf->data[MQTL_HEADER_LEN]); + pub->payload = &pub->buf->data[MQTL_HEADER_LEN + MQTL_TOPIC_ID_LEN]; + pub->payload_len = payload_len; + + return 0; +} + +int ares_mqttlite_publish_commit(struct AresProtocol *protocol, + struct ares_mqttlite_publish_buffer *pub) +{ + struct AresInterface *interface; + struct net_buf *buf; + + if (protocol == NULL || pub == NULL || pub->buf == NULL) { + return -EINVAL; + } + + interface = protocol->interface; + if (interface == NULL || interface->api == NULL || interface->api->send == NULL) { + ares_mqttlite_publish_abort(pub); + return -ENODEV; + } + + buf = pub->buf; + memset(pub, 0, sizeof(*pub)); + return interface->api->send(interface, buf); +} + +void ares_mqttlite_publish_abort(struct ares_mqttlite_publish_buffer *pub) +{ + if (pub != NULL && pub->buf != NULL) { + net_buf_unref(pub->buf); + memset(pub, 0, sizeof(*pub)); + } +} + +int ares_mqttlite_ping(struct AresProtocol *protocol) +{ + return send_ack(protocol, MQTL_FRAME_PING, 0); +} + +void ares_mqttlite_protocol_handle_byte(struct AresProtocol *protocol, uint8_t byte) +{ + struct ares_mqttlite_protocol_data *data = protocol->priv_data; + + if (data->rx_pos >= sizeof(data->rx_buf)) { + data->rx_pos = 0; + data->rx_expected = 0; + } + + data->rx_buf[data->rx_pos++] = byte; + + if (data->rx_pos == MQTL_HEADER_LEN) { + if (sys_get_le16(&data->rx_buf[MQTL_MAGIC_IDX]) != MQTL_MAGIC || + data->rx_buf[MQTL_VERSION_IDX] != MQTL_VERSION) { + memmove(data->rx_buf, &data->rx_buf[1], data->rx_pos - 1); + data->rx_pos--; + return; + } + data->rx_expected = MQTL_HEADER_LEN + frame_topic_len(data->rx_buf) + + frame_payload_len(data->rx_buf); + if (data->rx_expected > sizeof(data->rx_buf)) { + data->rx_pos = 0; + data->rx_expected = 0; + return; + } + } + + if (data->rx_expected != 0 && data->rx_pos >= data->rx_expected) { + uint8_t frame[CONFIG_ARES_MQTTLITE_MAX_FRAME_SIZE]; + uint16_t frame_len = data->rx_expected; + + memcpy(frame, data->rx_buf, frame_len); + data->rx_pos = 0; + data->rx_expected = 0; + process_frame(protocol, frame, frame_len); + } +} + +void ares_mqttlite_protocol_handle(struct AresProtocol *protocol, struct net_buf *buf) +{ + struct ares_mqttlite_protocol_data *data = protocol->priv_data; + uint16_t pos = 0; + + if (data->rx_pos == 0) { + while (buf->len - pos >= MQTL_HEADER_LEN) { + uint16_t frame_len; + + if (sys_get_le16(&buf->data[pos + MQTL_MAGIC_IDX]) != MQTL_MAGIC || + buf->data[pos + MQTL_VERSION_IDX] != MQTL_VERSION) { + break; + } + + frame_len = MQTL_HEADER_LEN + + sys_get_le16(&buf->data[pos + MQTL_TOPIC_LEN_IDX]) + + sys_get_le16(&buf->data[pos + MQTL_PAYLOAD_LEN_IDX]); + if (frame_len > CONFIG_ARES_MQTTLITE_MAX_FRAME_SIZE) { + pos = buf->len; + break; + } + if (frame_len > buf->len - pos) { + break; + } + + process_frame(protocol, &buf->data[pos], frame_len); + pos += frame_len; + } + } + + for (uint16_t i = pos; i < buf->len; i++) { + ares_mqttlite_protocol_handle_byte(protocol, buf->data[i]); + } +} + +void ares_mqttlite_protocol_event(struct AresProtocol *protocol, enum AresProtocolEvent event) +{ + struct ares_mqttlite_protocol_data *data = protocol->priv_data; + + k_mutex_lock(&data->lock, K_FOREVER); + data->online = event == ARES_PROTOCOL_EVENT_CONNECTED; + if (!data->online) { + (void)k_work_cancel_delayable(&data->retry_work); + for (int i = 0; i < CONFIG_ARES_MQTTLITE_MAX_INFLIGHT; i++) { + clear_tx(&data->tx[i]); + memset(&data->rx_qos2[i], 0, sizeof(data->rx_qos2[i])); + } + data->rx_pos = 0; + data->rx_expected = 0; + } + k_mutex_unlock(&data->lock); +} + +int ares_mqttlite_protocol_init(struct AresProtocol *protocol) +{ + struct ares_mqttlite_protocol_data *data; + + if (protocol == NULL || protocol->priv_data == NULL) { + return -EINVAL; + } + + data = protocol->priv_data; + k_mutex_init(&data->lock); + k_work_init_delayable(&data->retry_work, retry_work_handler); + data->protocol = protocol; + data->next_packet_id = 1; + data->rx_pos = 0; + data->rx_expected = 0; + data->online = false; + memset(data->topics, 0, sizeof(data->topics)); + memset(data->subscriptions, 0, sizeof(data->subscriptions)); + memset(data->rx_qos2, 0, sizeof(data->rx_qos2)); + + for (int i = 0; i < CONFIG_ARES_MQTTLITE_MAX_INFLIGHT; i++) { + clear_tx(&data->tx[i]); + } + + LOG_INF("%s MQTT-like protocol init", data->name); + return 0; +} diff --git a/samples/communication/mqttlite_usb/CMakeLists.txt b/samples/communication/mqttlite_usb/CMakeLists.txt new file mode 100644 index 0000000..d31c8c9 --- /dev/null +++ b/samples/communication/mqttlite_usb/CMakeLists.txt @@ -0,0 +1,6 @@ +cmake_minimum_required(VERSION 3.20.0) + +find_package(Zephyr REQUIRED HINTS $ENV{ZEPHYR_BASE}) +project(ares_mqttlite_usb) + +target_sources(app PRIVATE src/main.c) diff --git a/samples/communication/mqttlite_usb/README.md b/samples/communication/mqttlite_usb/README.md new file mode 100644 index 0000000..82b7665 --- /dev/null +++ b/samples/communication/mqttlite_usb/README.md @@ -0,0 +1,22 @@ +# ARES MQTT-like USB Bulk 示例 + +这个示例演示如何把 MQTT-like 协议绑定到 USB bulk 接口上。 + +## 功能 + +- 使用 `ARES_BULK_INTERFACE_DEFINE()` 定义 USB bulk 接口。 +- 使用 `ARES_MQTTLITE_PROTOCOL_DEFINE()` 定义协议实例。 +- 注册 `robot/mqttlite/heartbeat` 与 `host/command` 的 topic id。 +- 按 id 订阅 `host/command`。 +- 周期通过 QoS0 zero-copy 路径发布 `robot/mqttlite/heartbeat`。 + +## 构建 + +```sh +west build -b dm_mc02 samples/communication/mqttlite_usb --pristine +``` + +## 上位机约定 + +设备会发送带 topic-id 标志的 MQTT-like `PUBLISH` 帧，heartbeat 的 topic id 为 `1`， +`host/command` 的 topic id 为 `2`。当前示例 heartbeat 使用 QoS0，不等待上位机确认。 diff --git a/samples/communication/mqttlite_usb/boards/dm_mc02.conf b/samples/communication/mqttlite_usb/boards/dm_mc02.conf new file mode 100644 index 0000000..d9b3aa9 --- /dev/null +++ b/samples/communication/mqttlite_usb/boards/dm_mc02.conf @@ -0,0 +1,9 @@ +# dm_mc02 uses USART10 as the debug console, so keep this sample focused on USB bulk. +CONFIG_UART_INTERFACE=n + +CONFIG_CONSOLE=y +CONFIG_UART_CONSOLE=y +CONFIG_LOG_BACKEND_UART=y + +CONFIG_SENSOR=n +CONFIG_BMI08X=n diff --git a/samples/communication/mqttlite_usb/boards/dm_mc02.overlay b/samples/communication/mqttlite_usb/boards/dm_mc02.overlay new file mode 100644 index 0000000..c6cd80d --- /dev/null +++ b/samples/communication/mqttlite_usb/boards/dm_mc02.overlay @@ -0,0 +1,5 @@ +zephyr_udc0: &usbotg_hs { + pinctrl-0 = <&usb_otg_hs_dm_pa11 &usb_otg_hs_dp_pa12>; + pinctrl-names = "default"; + status = "okay"; +}; diff --git a/samples/communication/mqttlite_usb/prj.conf b/samples/communication/mqttlite_usb/prj.conf new file mode 100644 index 0000000..4d3387c --- /dev/null +++ b/samples/communication/mqttlite_usb/prj.conf @@ -0,0 +1,12 @@ +# ARES MQTT-like protocol over USB Bulk + +CONFIG_USB_BULK_INTERFACE=y +CONFIG_ARES_MQTTLITE_PROTOCOL=y +CONFIG_NET_BUF=y + +CONFIG_MAIN_THREAD_PRIORITY=10 + +CONFIG_LOG=y +CONFIG_LOG_DEFAULT_LEVEL=3 +CONFIG_UDC_DRIVER_LOG_LEVEL_INF=y +CONFIG_USBD_LOG_LEVEL_INF=y diff --git a/samples/communication/mqttlite_usb/sample.yaml b/samples/communication/mqttlite_usb/sample.yaml new file mode 100644 index 0000000..d597b3d --- /dev/null +++ b/samples/communication/mqttlite_usb/sample.yaml @@ -0,0 +1,14 @@ +# SPDX-License-Identifier: Apache-2.0 + +sample: + name: ARES MQTT-like USB Bulk +tests: + sample.communication.mqttlite_usb: + tags: + - ares + - communication + - usb + platform_allow: + - dm_mc02 + integration_platforms: + - dm_mc02 diff --git a/samples/communication/mqttlite_usb/src/main.c b/samples/communication/mqttlite_usb/src/main.c new file mode 100644 index 0000000..51a7b2d --- /dev/null +++ b/samples/communication/mqttlite_usb/src/main.c @@ -0,0 +1,96 @@ +/* + * Copyright (c) 2026 ttwards + * + * SPDX-License-Identifier: Apache-2.0 + */ + +#include + +#include +#include +#include + +#include +#include +#include + +LOG_MODULE_REGISTER(mqttlite_usb_sample, LOG_LEVEL_INF); + +#define TOPIC_HEARTBEAT 1 +#define TOPIC_HOST_COMMAND 2 + +ARES_BULK_INTERFACE_DEFINE(usb_bulk_interface); +ARES_MQTTLITE_PROTOCOL_DEFINE(mqttlite_protocol); + +static void command_cb(struct AresProtocol *protocol, const char *topic, const uint8_t *payload, + uint16_t payload_len, enum ares_mqttlite_qos qos, void *user_data) +{ + char text[65]; + size_t len = MIN(payload_len, sizeof(text) - 1); + + ARG_UNUSED(protocol); + ARG_UNUSED(user_data); + + memcpy(text, payload, len); + text[len] = '\0'; + + LOG_INF("RX topic=%s qos=%u payload=%s", topic, qos, text); +} + +int main(void) +{ + uint32_t counter = 0; + int ret; + + LOG_INF("ARES MQTT-like USB Bulk sample starting"); + + ret = ares_bind_interface(&usb_bulk_interface, &mqttlite_protocol); + if (ret != 0) { + LOG_ERR("Failed to bind USB bulk interface (%d)", ret); + return ret; + } + + ret = ares_mqttlite_register_topic(&mqttlite_protocol, TOPIC_HEARTBEAT, + "robot/mqttlite/heartbeat"); + if (ret != 0) { + LOG_ERR("Failed to register heartbeat topic (%d)", ret); + return ret; + } + + ret = ares_mqttlite_register_topic(&mqttlite_protocol, TOPIC_HOST_COMMAND, "host/command"); + if (ret != 0) { + LOG_ERR("Failed to register command topic (%d)", ret); + return ret; + } + + ret = ares_mqttlite_subscribe_id(&mqttlite_protocol, TOPIC_HOST_COMMAND, command_cb, NULL); + if (ret != 0) { + LOG_ERR("Failed to subscribe host/command (%d)", ret); + return ret; + } + + while (1) { + char payload[32]; + struct ares_mqttlite_publish_buffer pub; + size_t payload_len; + + snprintk(payload, sizeof(payload), "heartbeat:%u", counter++); + payload_len = strlen(payload); + + ret = ares_mqttlite_publish_prepare_id(&mqttlite_protocol, TOPIC_HEARTBEAT, + payload_len, &pub); + if (ret == 0) { + memcpy(pub.payload, payload, payload_len); + ret = ares_mqttlite_publish_commit(&mqttlite_protocol, &pub); + } + if (ret < 0) { + LOG_WRN("Publish failed (%d)", ret); + } else { + LOG_INF("QoS0 heartbeat published"); + } + + k_sleep(K_SECONDS(1)); + } + + return 0; +} diff --git a/tests/native_sim/mqttlite/CMakeLists.txt b/tests/native_sim/mqttlite/CMakeLists.txt new file mode 100644 index 0000000..49e3892 --- /dev/null +++ b/tests/native_sim/mqttlite/CMakeLists.txt @@ -0,0 +1,7 @@ +# SPDX-License-Identifier: Apache-2.0 + +cmake_minimum_required(VERSION 3.20.0) +find_package(Zephyr REQUIRED HINTS $ENV{ZEPHYR_BASE}) +project(ares_native_sim_mqttlite) + +target_sources(app PRIVATE src/main.c) diff --git a/tests/native_sim/mqttlite/prj.conf b/tests/native_sim/mqttlite/prj.conf new file mode 100644 index 0000000..439a995 --- /dev/null +++ b/tests/native_sim/mqttlite/prj.conf @@ -0,0 +1,9 @@ +CONFIG_ZTEST=y +CONFIG_ARES_MQTTLITE_PROTOCOL=y +CONFIG_ARES_MQTTLITE_MAX_TOPIC_LEN=32 +CONFIG_ARES_MQTTLITE_MAX_PAYLOAD_SIZE=64 +CONFIG_ARES_MQTTLITE_MAX_FRAME_SIZE=128 +CONFIG_ARES_MQTTLITE_MAX_INFLIGHT=4 +CONFIG_ARES_MQTTLITE_MAX_SUBSCRIPTIONS=8 +CONFIG_ARES_MQTTLITE_MAX_TOPICS=8 +CONFIG_ARES_MQTTLITE_RETRY_INTERVAL_MS=50 diff --git a/tests/native_sim/mqttlite/src/main.c b/tests/native_sim/mqttlite/src/main.c new file mode 100644 index 0000000..66082b2 --- /dev/null +++ b/tests/native_sim/mqttlite/src/main.c @@ -0,0 +1,238 @@ +/* + * Copyright (c) 2026 ttwards + * + * SPDX-License-Identifier: Apache-2.0 + */ + +#include +#include + +#include +#include + +#include +#include +#include + +NET_BUF_POOL_DEFINE(fake_pool, 16, 512, 0, NULL); + +struct fake_link { + struct AresInterface *peer; +}; + +static int fake_init(struct AresInterface *interface) +{ + ARG_UNUSED(interface); + return 0; +} + +static struct net_buf *fake_alloc_buf(struct AresInterface *interface) +{ + ARG_UNUSED(interface); + return net_buf_alloc(&fake_pool, K_NO_WAIT); +} + +static int fake_send(struct AresInterface *interface, struct net_buf *buf) +{ + struct fake_link *link = interface->priv_data; + + zassert_not_null(link->peer, "fake link peer must be set"); + zassert_not_null(link->peer->protocol, "fake link peer protocol must be bound"); + + link->peer->protocol->api->handle(link->peer->protocol, buf); + net_buf_unref(buf); + + return 0; +} + +static const struct AresInterfaceAPI fake_api = { + .init = fake_init, + .send = fake_send, + .alloc_buf = fake_alloc_buf, +}; + +static struct fake_link client_link; +static struct fake_link server_link; + +static struct AresInterface client_if = { + .name = "client", + .api = &fake_api, + .priv_data = &client_link, +}; + +static struct AresInterface server_if = { + .name = "server", + .api = &fake_api, + .priv_data = &server_link, +}; + +ARES_MQTTLITE_PROTOCOL_DEFINE(client_protocol); +ARES_MQTTLITE_PROTOCOL_DEFINE(server_protocol); + +static int rx_count; +static enum ares_mqttlite_qos last_qos; +static char last_topic[CONFIG_ARES_MQTTLITE_MAX_TOPIC_LEN + 1]; +static char last_payload[CONFIG_ARES_MQTTLITE_MAX_PAYLOAD_SIZE + 1]; +static int publish_status; +static uint16_t publish_packet_id; + +static void reset_observed(void) +{ + rx_count = 0; + last_qos = ARES_MQTTLITE_QOS0; + memset(last_topic, 0, sizeof(last_topic)); + memset(last_payload, 0, sizeof(last_payload)); + publish_status = INT_MIN; + publish_packet_id = 0; +} + +static void message_cb(struct AresProtocol *protocol, const char *topic, const uint8_t *payload, + uint16_t payload_len, enum ares_mqttlite_qos qos, void *user_data) +{ + ARG_UNUSED(protocol); + ARG_UNUSED(user_data); + + rx_count++; + last_qos = qos; + strncpy(last_topic, topic, sizeof(last_topic) - 1); + last_topic[sizeof(last_topic) - 1] = '\0'; + memcpy(last_payload, payload, payload_len); + last_payload[payload_len] = '\0'; +} + +static void publish_cb(struct AresProtocol *protocol, uint16_t packet_id, int status, + void *user_data) +{ + ARG_UNUSED(protocol); + ARG_UNUSED(user_data); + + publish_packet_id = packet_id; + publish_status = status; +} + +static void bind_pair(void) +{ + client_link.peer = &server_if; + server_link.peer = &client_if; + + zassert_ok(ares_bind_interface(&client_if, &client_protocol)); + zassert_ok(ares_bind_interface(&server_if, &server_protocol)); + client_protocol.api->event(&client_protocol, ARES_PROTOCOL_EVENT_CONNECTED); + server_protocol.api->event(&server_protocol, ARES_PROTOCOL_EVENT_CONNECTED); +} + +static void *suite_setup(void) +{ + bind_pair(); + return NULL; +} + +ZTEST(ares_mqttlite, test_qos0_publish_delivers_to_subscriber) +{ + const uint8_t payload[] = "ready"; + + reset_observed(); + + zassert_ok(ares_mqttlite_subscribe(&server_protocol, "robot/state", message_cb, NULL)); + zassert_ok(ares_mqttlite_publish(&client_protocol, "robot/state", payload, + sizeof(payload) - 1, ARES_MQTTLITE_QOS0, NULL, NULL)); + + zassert_equal(rx_count, 1, "rx_count=%d", rx_count); + zassert_equal(last_qos, ARES_MQTTLITE_QOS0); + zassert_mem_equal(last_payload, "ready", sizeof("ready")); +} + +ZTEST(ares_mqttlite, test_qos1_publish_gets_puback) +{ + const uint8_t payload[] = "armed"; + int packet_id; + + reset_observed(); + + zassert_ok(ares_mqttlite_subscribe(&server_protocol, "robot/state", message_cb, NULL)); + packet_id = + ares_mqttlite_publish(&client_protocol, "robot/state", payload, sizeof(payload) - 1, + ARES_MQTTLITE_QOS1, publish_cb, NULL); + + zassert_true(packet_id > 0, "QoS1 publish should return a packet id"); + zassert_equal(rx_count, 1, "rx_count=%d", rx_count); + zassert_equal(last_qos, ARES_MQTTLITE_QOS1); + zassert_equal(publish_packet_id, packet_id); + zassert_equal(publish_status, ARES_MQTTLITE_PUBLISH_ACKED); +} + +ZTEST(ares_mqttlite, test_qos2_publish_completes_exactly_once) +{ + const uint8_t payload[] = "shoot"; + int packet_id; + + reset_observed(); + + zassert_ok(ares_mqttlite_subscribe(&server_protocol, "robot/command", message_cb, NULL)); + packet_id = + ares_mqttlite_publish(&client_protocol, "robot/command", payload, + sizeof(payload) - 1, ARES_MQTTLITE_QOS2, publish_cb, NULL); + + zassert_true(packet_id > 0, "QoS2 publish should return a packet id"); + zassert_equal(rx_count, 1, "rx_count=%d", rx_count); + zassert_equal(last_qos, ARES_MQTTLITE_QOS2); + zassert_mem_equal(last_payload, "shoot", sizeof("shoot")); + zassert_equal(publish_packet_id, packet_id); + zassert_equal(publish_status, ARES_MQTTLITE_PUBLISH_ACKED); +} + +ZTEST(ares_mqttlite, test_topic_wildcards_match) +{ + const uint8_t payload[] = "42"; + + reset_observed(); + + zassert_ok(ares_mqttlite_subscribe(&server_protocol, "robot/+/rpm", message_cb, NULL)); + zassert_ok(ares_mqttlite_publish(&client_protocol, "robot/front/rpm", payload, + sizeof(payload) - 1, ARES_MQTTLITE_QOS0, NULL, NULL)); + + zassert_equal(rx_count, 1); + zassert_mem_equal(last_payload, "42", sizeof("42")); +} + +ZTEST(ares_mqttlite, test_topic_id_qos1_publish_gets_puback) +{ + const uint8_t payload[] = "fast"; + int packet_id; + + reset_observed(); + + zassert_ok(ares_mqttlite_register_topic(&server_protocol, 7, "robot/fast")); + zassert_ok(ares_mqttlite_subscribe_id(&server_protocol, 7, message_cb, NULL)); + packet_id = ares_mqttlite_publish_id(&client_protocol, 7, payload, sizeof(payload) - 1, + ARES_MQTTLITE_QOS1, publish_cb, NULL); + + zassert_true(packet_id > 0, "QoS1 id publish should return a packet id"); + zassert_equal(rx_count, 1, "rx_count=%d", rx_count); + zassert_equal(last_qos, ARES_MQTTLITE_QOS1); + zassert_mem_equal(last_topic, "robot/fast", sizeof("robot/fast")); + zassert_mem_equal(last_payload, "fast", sizeof("fast")); + zassert_equal(publish_packet_id, packet_id); + zassert_equal(publish_status, ARES_MQTTLITE_PUBLISH_ACKED); +} + +ZTEST(ares_mqttlite, test_topic_id_qos0_prepare_commit_is_zero_copy) +{ + struct ares_mqttlite_publish_buffer pub; + + reset_observed(); + + zassert_ok(ares_mqttlite_register_topic(&server_protocol, 8, "robot/zero")); + zassert_ok(ares_mqttlite_subscribe_id(&server_protocol, 8, message_cb, NULL)); + zassert_ok(ares_mqttlite_publish_prepare_id(&client_protocol, 8, 4, &pub)); + zassert_not_null(pub.payload); + memcpy(pub.payload, "zero", 4); + zassert_ok(ares_mqttlite_publish_commit(&client_protocol, &pub)); + + zassert_equal(rx_count, 1, "rx_count=%d", rx_count); + zassert_equal(last_qos, ARES_MQTTLITE_QOS0); + zassert_mem_equal(last_topic, "robot/zero", sizeof("robot/zero")); + zassert_mem_equal(last_payload, "zero", sizeof("zero")); +} + +ZTEST_SUITE(ares_mqttlite, NULL, suite_setup, NULL, NULL, NULL); diff --git a/tests/native_sim/mqttlite/testcase.yaml b/tests/native_sim/mqttlite/testcase.yaml new file mode 100644 index 0000000..fcfa0c3 --- /dev/null +++ b/tests/native_sim/mqttlite/testcase.yaml @@ -0,0 +1,14 @@ +# SPDX-License-Identifier: Apache-2.0 + +tests: + ares.native_sim.mqttlite: + tags: + - ares + - native_sim + - protocol + platform_allow: + - native_sim + - native_sim/native/64 + integration_platforms: + - native_sim + - native_sim/native/64