mqtt

网络通信

1.0.0C / xlang二进制报文

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.sh

2026-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。