#moon-mqtt

#安装

moon add Duckweed/moon-mqtt@0.3.3

import {
  "Duckweed/moon-mqtt",
}

#在线 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
[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=1, replay 0 frame(s)
[demo] connected, subscribing sensors/# QoS1

#运行示例

#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）
• browser/ 浏览器高层 API（connect / connect_persistent / on_connect / on_suback / on_message / subscribe / publish / disconnect）
• 浏览器端到端实测：2026-10-22 无头 Chrome，首连、publish QoS1、收消息、强制断开后重连到 session_present=1
• moon test --target wasm-gc 82/82；moon test --target native 107/107
• moon check --target js --deny-warn 通过，0 警告
• 发布到 mooncakes（0.3.3）

#构建与测试

moon check
moon check --target js --deny-warn   # 0 警告
moon test --target wasm-gc           # 82/82
moon test --target native            # 107/107
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 / 边缘遥测；高并发长连接要异步封装，当前没做。
• 同步握手期间收到 PUBLISH：publish 等 PUBACK/PUBREC/PUBCOMP、subscribe 等 SUBACK 时，如果 broker 推来订阅主题的 PUBLISH，当前代码报 "expected XXX" 并丢掉这帧。边发边收请用 poll_loop + subscribe_with_handler；poll_once 处理入站 QoS2 已不再嵌套等 PUBREL。
• 坏协议帧卡住缓冲：core/Framer 遇到认不出的帧（剩余长度超 4 字节、保留报文类型等）会把这帧永久留在内部缓冲，后面的帧也就出不来，且不 reset。正常 broker 不会发这种帧；要防对端异常，得在调用方加超时。
• 读超时：单次 socket 读有 5 秒超时（wait_read），CONNACK/PUBACK/PINGRESP 等同步等待不会永久挂起；极慢的分段帧仍可能超时返回 Err。
• 浏览器自动重连上限：browser/ 断线后 1 秒重连，连续失败最多 3 次；每次连上成功即把计数归零，长期在线不会耗尽配额。连续 3 次失败后刷新页面才会重新建立连接。
• client/ 层测试：25 条单测（client/client_wbtest.mbt），经 Conn 接缝注入脚本化假连接，覆盖 CONNECT/CONNACK 各分支、QoS0/1/2 发布握手、SUBACK、poll 派发与 QoS2 去重、自动心跳、断开、分帧上下限（wasm-gc 82 / native 107）。native + C FFI 无法在 wasm 后端实例化。
• browser/ 层没有自动测试：浏览器路径依赖 extern "js" 与 WebSocket，只能靠 docs/demo/ 的手工验证与 docs/CASES.md 的 I/O 记录；改 browser/ 后必须重跑 scripts/build-demo.ps1 并手工回归 demo 页。
• MQTT 5.0 本期不做：CONNECT/PUBLISH properties、CONNACK reason code 等基础字段列入 11–1 月维护计划；完整 5.0 与 native TLS 视工具链在 2027 评估（浏览器侧的 TLS 由浏览器负责，不属本项目范围）。

#许可证