扩展库 / mqtt
mqtt
网络通信MQTT 3.1.1 QoS 0/1 报文编解码、TCP 逐帧收发与一次性握手。使用 Buffer 处理二进制,使用 hex 检查报文;C 与 xlang 共享 XRT 内存、错误和网络设施。
范围:这是协议与传输工具库,不是完整客户端或 broker。没有自动心跳、重连、重发、去重、订阅路由、持久会话、Will、UNSUBSCRIBE、QoS 2、MQTT 5 或 TLS 集成。构造 QoS 1 报文不等于完整的会话交付保证。本文对应当前源码,不代表旧下载包已包含更新。
快速开始
import mqtt
func main(...) as int
Buffer frame = mqtt.publishFrame(1, "home/temp", "22.5", 42)
var message = mqtt.parseFrame(frame)
println(message["topic"])
println(message["payload"])
println(mqtt.frameToHex(frame))
return 0
end正文含零字节或不是 UTF-8 时,传入 Buffer 并用 mqtt.payloadBytes(frame) 提取独立副本。文本重载通过 C 字符串传参,不保留嵌入 NUL。不要把协议字节当作普通文本。
xlang API
| 入口 | 返回与合同 |
|---|---|
connectFrame(clientId, username = "", password = "", keepalive = 60, cleanSession = true) | Buffer;前三项为 string,keepalive 为 int 秒,0 不开启心跳。password 也可为 Buffer,此时 username 必填。 |
publishFrame(qos, topic, payload, packetId = 0, retain = false, duplicate = false) | Buffer;qos/packetId 为 int,topic 为 string,payload 为 string 或 Buffer。QoS 0 只允许 id=0、DUP=false;QoS 1 的 id 为 1..65535。 |
pubackFrame(packetId) | Buffer;packetId 为 1..65535。 |
subscribeFrame(packetId, topic, qos = 0) | Buffer;单个过滤器,支持独占层级的 + 和末级 #,QoS 0/1。 |
pingreqFrame() / disconnectFrame() | Buffer;仅构帧,不自动操作连接。 |
hexToFrame(string hex) / frameToHex(Buffer frame) | Buffer / 小写 hex string;只转换字节,不验证完整协议。 |
frameSize(Buffer prefix) | int;固定头未齐返回 0,头齐后返回第一帧总长度,正文可尚未收齐。 |
parseFrame(Buffer frame) / parse(string hex) | var 字典;恰好一帧,拒绝截断、尾随数据和无效字段;结果不借用原帧。 |
payloadBytes(Buffer frame) | 严格验证 PUBLISH,返回正文的独立 Buffer。 |
connect、publish、puback、subscribe、pingreq、disconnect | 同名 *Frame 文本重载的 hex 便捷形式,参数相同,返回 string。 |
构造/解析/范围错误抛 ValueError。协议文本必须是无 NUL 的有效 UTF-8,单字段上限 65535 字节。空 username/password 表示不携带字段;非空密码要求非空 username;构造入口不能区分不存在与存在但长度为零的认证字段。
解析字段
所有报文有 type 和 remaining。CONNECT 有 clientId、cleanSession、keepalive、username、password;CONNACK 有 returnCode、sessionPresent;PUBLISH 有 topic、payload、qos、retain、dup,仅 QoS 1 有 packetId。PUBACK 有 packetId。SUBSCRIBE 有 packetId、subscriptions(topic/qos 项);SUBACK 有 packetId、granted(0/1/2/128)。PINGREQ/PINGRESP/DISCONNECT 没有额外字段。
granted=2 可作为协议数据解析,但不表示库实现 QoS 2。解析器不维护 packetId 状态;应用应检查应答是否匹配请求。二进制正文请使用 payloadBytes,不能依赖 var 字典转文本的结果。
TCP 与失败边界
| 入口 | 行为 |
|---|---|
sendFrame(TcpStream stream, Buffer frame, int timeoutMs = 5000, point cancel = null) | 检查固定头及总长度,分片发送并等待队列排空,不重新验证正文。 |
recvFrame(TcpStream stream, int timeoutMs = 5000, int packetLimit = 1048576, point cancel = null) | 返回独立 Buffer,只消费本帧,不吞后续帧;正文语义交给 parseFrame。 |
handshake(TcpStream stream, string clientId, string username = "", string password = "", int keepalive = 60, int timeoutMs = 5000, point cancel = null) | Clean Session CONNECT 与成功 CONNACK 共用一个截止时间;返回解析字典,不接管 stream。 |
- timeoutMs=-1 不限时,0 立即超时,正数为整个操作的总预算,不逐片重置。C API 使用单调微秒绝对截止时间。
- 默认接收限制 1MiB,含固定头;范围 2..268435460,在分配正文前检查。
- 同一连接上的读取与写入分别串行化,握手独占双向操作;阻塞 API 不可在该连接的驱动 worker 回调中使用。
- cancel 是借用的 XRT xcancel*,存活至调用结束;null 不取消,不能传其他句柄。
- 空连接等待超时/取消可以重试;读入半帧后失败会终止连接。CONNECT 发出后,握手失败也会终止连接。
- 发送开始后失败会终止连接,部分数据可能已经到达对端;成功不是业务确认,不应据此盲目重发。
- 网络入口抛 IOError 并保留底层诊断;未开始 I/O 的参数失败不改变连接。
C API
公开头为 extlibs/mqtt/include/xrt_mqtt.h 和 xrt_mqtt_net.h。编解码编译 mqtt_frame.c、mqtt_build.c、mqtt_parse.c,网络增加 mqtt_net.c,链接一份 XRT。不需要 xlang 编译器或 mqtt_native.c。
| 函数 | 拥有权 |
|---|---|
xrtMqttConnect / Publish / Puback / Subscribe / Pingreq / Disconnect | 实际函数均以 xrtMqtt 开头;返回 xbuffer*,失败 NULL,用 xrtBufferDestroy 释放。 |
xrtMqttInspect | ERROR/MORE/READY;MORE 不设置新错误,完整头填写 PacketSize。 |
xrtMqttPublishView | bool;Topic/Payload 借用原帧,原帧修改或销毁后失效。 |
xrtMqttParse | xvalue*,复制内容;失败 NULL,用 xrtValueRelease 释放。 |
xrtMqttSendFrame | bool;借用 stream、frame、cancel。 |
xrtMqttRecvFrame / xrtMqttHandshake | 拥有型 xbuffer* / xvalue*;失败 NULL;不接管连接。 |
以返回值判断本次操作是否失败;成功不清除之前的 xrtGetError 错误槽。语言桥的 16 个 xmq_* 声明单独放在 mqtt_native.h,并与编译器生成的原型进行严格 C 类型核对。
范例与验证
xl run examples/extlibs/mqtt.xl
xl run examples/extlibs/mqtt_c.c
xl run examples/extlibs/mqtt_tcp.xl
xl run examples/extlibs/mqtt_tcp_c.c前两份离线构帧/解析范例输出相同的 15 行;后两份本机随机端口 TCP 范例输出相同的 4 行。无需公共 broker;均可用 xl build 编译。
powershell -NoProfile -ExecutionPolicy Bypass -File tools/check_extlib_mqtt.ps1
bash tools/check_extlib_mqtt.sh2026-09-26 本地验收范围:Windows GCC、C/xlang JIT/AOT、原型与黄金输出对照;Linux 独立 C ASan/UBSan。覆盖逐字节分包、粘包、64KiB 二进制、无效包、OOM、发送背压、异常 CONNACK、超时与取消。构建及源码哈希绑定实际结果。
未覆盖真实 broker 矩阵、Linux xlang JIT/AOT、长时会话、TLS 或公网环境。测试通过不能作为未实现会话能力的证明。仓库门禁结果见 build/test/mqtt_guard 与 mqtt_linux_guard。
协议依据 OASIS MQTT 3.1.1 Errata 01;完整 API 合同位于发行包 extlibs/mqtt/README.md。