#moon-mqtt

#安装

moon add Duckweed/moon-mqtt@0.3.13

import {
  "Duckweed/moon-mqtt@0.3.13",
}

#在线 demo

pwsh scripts/build-demo.ps1   # moon build --target js examples/browser-demo，产物复制到 docs/demo/
python -m http.server 8765    # 在仓库根目录
# 浏览器打开 http://127.0.0.1:8765/docs/demo/

#浏览器用法

// your-app/main.mbt
import {
  "Duckweed/moon-mqtt/browser" @browser,
  "Duckweed/moon-mqtt/protocol" @protocol,
}

fn main() -> Unit {
  let client = @browser.connect_persistent("broker.emqx.io", 8084, "my-browser-client")

  @browser.on_connect(client, fn() {
    println("connected")
    @browser.subscribe(client, "sensors/#", @protocol.qos1())
  })

  // SUBACK 之后再发，broker 已经建好订阅，能收到自己发的消息
  @browser.on_suback(client, fn() {
    @browser.publish(client, "sensors/hello", [0x68, 0x69], @protocol.qos1())
  })

  @browser.on_message(client, fn(topic, payload) {
    println("recv " + topic + " " + payload.length().to_string() + " bytes")
  })
}

[browser] session_present=0, resubscribe 0 topic(s)
[demo] connected, subscribing sensors/# QoS1
[browser] suback
[demo] suback, publishing QoS1 sensors/hello
[demo] recv topic=sensors/hello payload=2 bytes

[browser] reconnect attempt #1 in 1s
[browser] session_present=0, resubscribe 1 topic(s)
[demo] reconnected, rely on core auto-resubscribe
[browser] suback

#运行示例

#pubsub：报文编解码

let qos = match @protocol.qos_from_int(1) { Some(q) => q None => return }
let pkt = @protocol.PublishPacket::new(
  "sensors/temp", [0x32, 0x33], qos, false, false, 7,
)
let bytes = @protocol.encode_publish(pkt)
match @protocol.decode_publish(bytes) { ... }

#resume：QoS2 中断重连

#browser-demo：浏览器端到端

#Native TCP

// 自定义 CONNECT（带认证 + 遗嘱 + 持久会话）
let opts = @protocol.ConnectPacket::new("demo-client")
  .with_keep_alive(30)
  .with_clean_session(false)
  .with_auth("user", [0x70, 0x61, 0x73, 0x73])   // "pass"
  .with_will("status", "offline".to_bytes(), @protocol.qos1(), false)

let client = @client.connect_opts("broker.hivemq.com", 1883, opts)?

client.subscribe_with_handler("sensors/#", @protocol.qos1(), fn(topic, payload) {
  println("recv " + topic)
})?

// 每 1 秒 poll 一次，累计到 keep_alive 自动发 PINGREQ
match client.poll_loop(600, 1000) {
  Ok(_) => ()
  Err(_) => {
    let client = client.reconnect()?
    println("reconnected, session_present=" + client.is_session_present().to_string())
  }
}
client.disconnect()?

#命令行验证

moon build --target js examples/ws-spike
node "_build\js\debug\build\examples\ws-spike\ws-spike.js"

#设计

矢量图在 docs/assets/architecture.svg。颜色区分应用、宿主 API、传输层和纯计算层。

• 浏览器：browser/ 持有 core/ClientCore，JS 胶水层持有 WebSocket，经 mangled 名回调 MoonBit 的 browser_on_message / browser_on_close。core/ 吐事件给应用，同时吐出站字节由 browser_send 写回 broker。
• native：client/ 用 protocol/ + session/ + transport/ 组装阻塞式 MqttClient，走 C FFI TCP。
• protocol/、session/、core/ 不 import 任何 platform 依赖；transport/ 和 client/ 只编 native，browser/ 只编 js。
• QoS2 exactly-once（PUBREC/PUBREL/PUBCOMP 三段握手）和断线重连的状态边界是核心。

sequenceDiagram
    participant B as Broker
    participant G as 胶水层 (WebSocket / C FFI)
    participant C as core.ClientCore
    participant A as 应用回调
    G->>C: start()
    C-->>G: CONNECT 字节
    G->>B: send CONNECT
    B-->>G: CONNACK 字节
    G->>C: feed(bytes)
    C-->>G: [Connack] + 出站 []
    G->>A: on_connack(session_present)
    B-->>G: PUBLISH (QoS2) 字节
    G->>C: feed(bytes)
    C-->>G: [Publish] + [PUBREC]
    G->>A: on_publish(topic, payload)
    G->>B: send PUBREC
    B-->>G: PUBREL
    G->>C: feed(bytes)
    C-->>G: [] + [PUBCOMP]
    G->>B: send PUBCOMP

#CONNECT 字段支持

#状态

• protocol/ MQTT 3.1.1 报文编解码
• session/ 会话状态机（QoS2 exactly-once、packet_id 分配、在途持久化）
• core/ ClientCore（start/feed、入站 QoS1/2 自动握手、出站 QoS1/2 在途追踪、reconnect、replay）
• transport/ native TCP（C FFI，跨平台，read 带超时）
• client/ 阻塞式高层 API（poll_loop 自动心跳、reconnect 自动 replay、subscribe_with_handler / unsubscribe）
• browser/ 浏览器高层 API（connect / connect_persistent / on_connect / on_suback / on_message / subscribe / publish / disconnect）
• 会话恢复：core/ 本地订阅表 + session_present=0 时自动重订；browser/ 在 subscribe 时登记待确认订阅，SUBACK 通过后自动落库
• 浏览器端到端实测：2026-11-04 无头 Chrome（?autotest=1），首连订阅、publish QoS1、收消息、强制断开、重连后由 core 自动重订并收到第二次 SUBACK
• async/ 非阻塞外观骨架（AsyncClient：feed_bytes / drain_pending / drain_events 与 subscribe / unsubscribe / publish / reconnect），真实非阻塞 IO 运行时接入属下阶段
• 浏览器 MQTT 保活：browser_tick 由 JS 每秒驱动，每 keep_alive/2 秒发 PINGREQ，连续 3 次无 PINGRESP 主动断开重连
• 事件 → 动作的决策表：core/plan_event 把「该不该关连接、该回调谁」做成纯函数，browser/ 与 async/ 共用同一张表
• 主题校验（§4.7）：publish 的 topic name 拒绝空串 / 通配符 / U+0000；subscribe 的过滤器拒绝 a/#/b、a/x+ 这类越界写法；解码侧同样拒绝 broker 发来的非法 topic
• moon test --target wasm-gc 141/141；moon test --target native 180/180（js 后端本机缺 node，交给 CI 跑）
• moon check --target js --deny-warn 通过，0 警告
• 发布到 mooncakes（0.3.13）

#构建与测试

moon check
moon check --target js --deny-warn   # 0 警告
moon test --target wasm-gc           # 141/141
moon test --target native            # 180/180
moon run examples/pubsub
moon run examples/resume
pwsh scripts/build-demo.ps1          # 浏览器 demo

#手动验证（native 连真 broker）

# 1. 起本地 broker
mosquitto -p 1883 -v

# 2. 另开终端订阅
mosquitto_sub -h 127.0.0.1 -t 'sensors/#' -v

# 3. 跑 MoonBit 程序（connect → publish → subscribe → ping → disconnect）
#    代码见上文「Native TCP」

#AI 工具使用说明

• AI 辅助了报文编解码的样板代码、单测用例生成、README 初稿润色。
• 人把控了选题、分层、QoS 状态机边界条件、C FFI 边界设计、已知限制的如实声明、与 mooncakes 现有项目的查重。
• AI 生成的单测都由人对照 OASIS MQTT 3.1.1 规范核过断言；编解码有往返测试（encode → decode → 字段比对）；CI 四后端全绿才合并。
• 不照搬其他开源项目代码（API 参考 MQTT.js / Paho C，实现是 MoonBit 重写）；不把 AI 生成的测试数据当真理。

#已知限制

• 会话快照落盘：SessionStore 提供 snapshot() / restore() 字节级原语，写盘由调用方自己做，库不碰文件。
• transport/ 和 client/ 仅 native：依赖 C FFI socket。protocol/、session/、core/ 四后端可测；浏览器走 browser/ + WebSocket 胶水层，已端到端验证。browser/ 仅 js 后端。
• 阻塞式 API：client/ 是阻塞的，适合 CLI / 边缘遥测；高并发长连接要异步封装。async/ 目前是非阻塞外观骨架（AsyncClient 由调用方驱动 feed_bytes / drain_pending / drain_events），还没有接真实非阻塞 IO 运行时。
• 同步握手期间收到 PUBLISH（已修，2026-11-07）：publish 等 PUBACK/PUBREC/PUBCOMP、subscribe 等 SUBACK、ping 等 PINGRESP 的等待统一走 read_control，途中到达的业务 PUBLISH 照常按订阅表派发、入站 PUBREL 照常回 PUBCOMP，不再当成「非目标帧」丢弃。10 月轮验收未通过的直接原因就是这条。
• 坏协议帧（已修，2026-11-10）：core/Framer 现在区分半包与畸形头。畸形头（保留报文类型、剩余长度超 4 字节）或缓冲超过协议最大帧时，清空缓冲并返回 Err；ClientCore::feed 把它转成 FramerError 事件，调用方据此断开连接。半包仍按原语义等下一批字节。
• 读超时：单次 socket 读有 5 秒超时（wait_read），CONNACK/PUBACK/PINGRESP 等同步等待不会永久挂起；极慢的分段帧仍可能超时返回 Err。
• 浏览器自动重连上限：browser/ 断线后 1 秒重连，连续失败最多 3 次；每次连上成功即把计数归零，长期在线不会耗尽配额。连续 3 次失败后刷新页面才会重新建立连接。
• client/ 层测试：28 条白盒单测（client/client_wbtest.mbt），经 Conn 接缝注入脚本化假连接，覆盖 CONNECT/CONNACK 各分支、QoS0/1/2 发布握手、SUBACK、poll 派发与 QoS2 去重、自动心跳、断开、分帧上下限、握手等待期间（含 QoS2 等 PUBREC、等 PUBCOMP 两个等待点）的插帧派发（当前实测 wasm-gc 107 / native 137）。native + C FFI 无法在 wasm 后端实例化。
• native 重连的入站 QoS2 去重：reconnect 在 session_present=1 时保留旧客户端的 inbound_qos2 表，避免 broker 重发「已派发、PUBREL 尚未完成」的 PUBLISH 被重复派发；session_present=0 时清空。这条路径需要真实 TCP，未纳入假连接单测。
• browser/ 层：WebSocket 的 IO 仍然没有自动测试（依赖 extern "js"，只能靠 docs/demo/ 手工验证与 docs/CASES.md 的 I/O 记录；改 browser/ 后必须重跑 scripts/build-demo.ps1 并手工回归 demo 页）。但决策层已有单测：事件 → 动作的映射统一走 core/plan_event（core/dispatch_wbtest.mbt 7 条），拒绝 CONNACK 与畸形帧都必须走「关连接」分支，不会再被 _ => () 静默吞掉。
• 畸形帧后的收尾（2026-11-10 补）：ClientCore::feed 产出 FramerError 之后，browser/ 现在会关 WebSocket 并回调 on_error，async/ 会冻结实例并让 AsyncClient::fatal() 返回原因；此前两层都把它当普通事件忽略，与「调用方应断开」的约定不符。
• MQTT 5.0 本期不做：CONNECT/PUBLISH properties、CONNACK reason code 等基础字段列入 11–1 月维护计划；完整 5.0 与 native TLS 视工具链在 2027 评估（浏览器侧的 TLS 由浏览器负责，不属本项目范围）。

#许可证