#moon-mqtt-client

#安装与构建

moon add Strangelight-Merser/moon-mqtt-client

moon update
moon check --target native
moon test --target native
moon build --target native

#先运行状态同步演示

python3 -m venv .venv
.venv/bin/pip install -r tests/integration/requirements.txt
./scripts/moon.sh build --target native
.venv/bin/python examples/mqtt_demo/demo.py

• desired：带回差的温控规则所期望的状态。
• result：发送结果，取值为 sent、not_sent 或 unknown。
• reported：设备实际反馈的状态，通过 ID 与本次命令或查询关联。

./scripts/moon.sh run examples/mqtt_demo/cli --target native -- \
  publish --host 127.0.0.1 -t lab/state -m ON --qos 1 --retain --stats
./scripts/moon.sh run examples/mqtt_demo/cli --target native -- \
  subscribe --host 127.0.0.1 -t 'lab/#' --count 1 --timeout-ms 3000

#支持范围

• 原生 TCP/TLS，以及 WS/WSS；TLS 支持系统根证书或自定义 PEM CA。
• 可选双向 TLS（mTLS）：PEM 证书链 + 未加密私钥；Plain 不能搭配客户端身份。
• 默认 MQTT 3.1.1，可选 MQTT 5 应用子集；QoS 0/1、保留消息、遗嘱消息（Last Will）和用户名/密码认证。
• 订阅与取消订阅确认，包括逐主题的订阅拒绝结果。
• 有上限的发送队列、事件队列、报文大小和并发请求数。
• 默认 CleanSession=true：重连创建新会话，恢复已确认的订阅，随后发出 Connected(generation) 事件。可显式选择 ResumeSession，通过投递句柄在同一 进程作用域内恢复未确认的 QoS 1 发布；原生 beta 还可选择 SQLite durable outbox，在进程重启后恢复同一逻辑会话中的显式投递。
• 任务和 socket 的生命周期由回调作用域管理。回调正常返回或调用 disconnect() 时 发送 DISCONNECT；回调异常或被取消时直接关闭传输连接。

#原生 WebSocket

let config = @mqtt.Config::new(
  "broker.example.com", "native-wss-client", port=8084,
  transport=@mqtt.WebSocket("/mqtt"), tls=@mqtt.SystemRoots,
)

./scripts/moon.sh run examples/mqtt_demo/cli --target native -- \
  publish --host broker.example.com --port 8084 --ws-path /mqtt --tls \
  -t lab/state -m ON --qos 1

#API 示例

async fn main {
  let config = @mqtt.Config::new("127.0.0.1", "my-moon-client")
  @mqtt.with_client(config, async fn(client) {
    let results = client.subscribe([
      { topic: "lab/temperature", qos: @mqtt.AtLeastOnce },
    ])
    if results == [@mqtt.Granted(@mqtt.AtLeastOnce)] {
      client.publish("lab/status", @utf8.encode("ready"),
        qos=@mqtt.AtLeastOnce, retain=true)
    }
    // 在这里处理 Connected、Disconnected 和 MessageReceived 事件。
    // 长期运行的订阅者必须持续调用 next_event() 消费事件。
  })
}

let config = @mqtt.Config::new(
  "127.0.0.1", "exclusive-stable-client-id",
  session_policy=@mqtt.ResumeSession,
)
@mqtt.with_client(config, async fn(client) {
  let delivery = client.submit_delivery(
    "command-20260918-001", "lab/command", @utf8.encode("ON"),
  )
  let status = delivery.wait()
  // Acknowledged means a matching broker PUBACK was received. Reconcile
  // TerminalOutcomeUnknown with application state before issuing a new command.
  match status.terminal_cause {
    Some(@mqtt.BrokerSessionLost(_)) => () // establish a new logical session
    Some(@mqtt.DeliveryAttemptsExhausted(_)) => () // reconcile before retry
    _ => ()
  }
})

let outbox = @mqtt.DurableOutboxOptions::new("./commands.sqlite3")
let config = @mqtt.Config::new(
  "127.0.0.1", "exclusive-stable-client-id",
  session_policy=@mqtt.ResumeSession,
)
@mqtt.with_durable_client(outbox, config, async fn(client, recovered) {
  let delivery = if recovered.is_empty() {
    client.submit_durable_delivery(
      "command-20260918-002", "lab/command", @utf8.encode("OFF"),
      @env.now().reinterpret_as_int64() + 30000L,
    )
  } else {
    recovered[0]
  }
  ignore(delivery.wait())
})

#投递结果与失败语义

#MQTT 5 子集

let config = @mqtt.Config::new("127.0.0.1", "requester", protocol=@mqtt.Mqtt5)
@mqtt.with_client(config, async fn(client) {
  let receipt = client.publish_detailed(
    "service/request", b"status", qos=@mqtt.AtLeastOnce,
    properties=@mqtt.PublishProperties::new(
      message_expiry_secs=Some(30L),
      response_topic=Some("service/reply"),
      correlation_data=Some(b"request-1"),
      user_properties=[("source", "moonbit")],
    ),
  )
  match receipt {
    @mqtt.Accepted(reason) => println("PUBACK reason=\{reason.code}")
    @mqtt.Written => println("written")
  }
})

#更多使用场景

1. 带回差、保留状态和在线状态的温度控制。
2. 有界 Frigate 事件去重，对已结束的人员事件发出提醒。
3. ROS 桥接的 JSON/基本类型契约，包含命令校验与回执。

#验证

PYTHON=.venv/bin/python ./tests/integration/run.sh
./scripts/check.sh

# 独立模拟设备的四种状态同步场景
.venv/bin/python examples/mqtt_demo/demo.py

# 从 Mooncakes 安装到全新的临时模块，并完成 QoS 1 消息往返
.venv/bin/python tests/consumer_smoke.py --registry

# 固定版本的官方 EMQX 镜像：收发、拒绝订阅、重启恢复；需要 Docker
./tests/emqx.sh

# 30 分钟重启压力测试：1 KiB QoS 1、16 个并发任务、100 次断线恢复
# broker 默认仅记录警告和错误，日志大小有上限
MOONBIT_ASYNC_CHECK_FD_LEAK=1 .venv/bin/python tests/soak.py \
  --duration 1800 --cycles 100 --artifacts tests/integration/artifacts/soak

#许可证与上游归属