MQTT 3.1.1 client library for MoonBit with TCP and TLS transports
| 能力 | 当前状态 |
|---|---|
| CONNECT/CONNACK、拒绝原因、用户名/密码、clean session | 已支持;clean_session 默认 true,可设为 false;公开 session_present() |
| SUBSCRIBE/SUBACK、UNSUBSCRIBE/UNSUBACK | 已支持;每次一个过滤器,订阅可请求 QoS 0/1/2 |
| QoS 0 双向发布 | 已支持;发送在 TCP 写入完成后返回 |
| QoS 1 双向发布 | 已支持;发送等待匹配的 PUBACK;接收在入队后发 PUBACK |
| QoS 2 双向发布 | 已支持四步握手;发送等待 PUBCOMP;接收采用规范的 Method B,持久存储可保留去重状态 |
| 心跳、正常断开 | PINGREQ/PINGRESP、DISCONNECT 已支持 |
| 流式读取 | 支持 TCP 分片与合包、最大报文限制 |
| 待处理请求与队列 | 报文 ID 预留、响应类型/ID 检查、超时和取消清理;接收队列有容量上限 |
| 同进程持久会话与显式重连 | clean_session=false 时保留 QoS 1/2 在途状态及入站 QoS 2 去重;按原 ID 重发未确认的 PUBLISH 或 PUBREL |
| 可选自动重连 | AutoClient 稳定句柄;连接中、已连接、重试中、永久失败和停止状态;临时故障使用有上限指数退避,支持停止与取消 |
| native 进程重启恢复 | 可选 SessionStore;保存出站 QoS 1/2 阶段及原 ID、入站 QoS 2 状态和未消费消息;支持文件存储或应用自定义存储 |
| MQTT 3.1.1 遗嘱 | 可配置主题、二进制负载、QoS 0/1/2 和 retain;每代自动重连的 CONNECT 都重新携带遗嘱 |
| native MQTT over TLS | 可选;默认验证系统信任链与连接主机名,并以该主机名设置 SNI;可指定 PEM CA;握手受连接超时限制 |
| WebSocket、MQTT 5 | 未实现 |
moon add youyong5/moonmqtt@0.1.0
moon add moonbitlang/async@0.22.5
moon add zbhzs1/moonbit-mqtt@0.1.0import {
"youyong5/moonmqtt" @client,
"moonbitlang/async" @async,
"zbhzs1/moonbit-mqtt" @packet,
}async fn main {
let config = @client.Config::new("127.0.0.1", "my-client", port=1883)
config.inbox_capacity = 128
@client.with_client(config, client => {
let _ = client.subscribe("demo/#", @packet.QoS2)
client.publish("demo/hello", b"hello", qos=@packet.QoS2)
let message = @async.with_timeout(5000, () => client.receive())
println(message.topic)
client.unsubscribe("demo/#")
})
}let config = @client.Config::new("127.0.0.1", "stable-client-id")
config.will = Some(@client.Will::new(
"devices/alpha/status",
Bytes::from_array([0x00, 0xff, 0x41]),
qos=@packet.QoS2,
retain=true,
))let config = @client.Config::new("broker.example.com", "stable-client-id", port=8883)
config.tls = Some(@client.TlsConfig::new()) // 系统可信根
// 私有 CA:config.tls = Some(@client.TlsConfig::new(ca_file="./ca.pem"))let config = @client.Config::new("127.0.0.1", "stable-client-id")
config.clean_session = false
@async.with_task_group(group => {
let first = @client.Client::connect(config, group)
let _ = first.subscribe("demo/#", @packet.QoS1)
// 断线后,或主动调用 first.close()/first.disconnect() 后:
let second = first.reconnect(group)
if second.session_present() {
println("resumed broker session")
}
second.disconnect()
})let config = @client.Config::new("127.0.0.1", "stable-auto-id")
config.clean_session = false
@async.with_task_group(group => {
let client = @client.AutoClient::start(
config, group,
policy=@client.ReconnectPolicy::new(initial_delay_ms=100, max_delay_ms=5000),
)
client.wait_connected()
let _ = client.subscribe("demo/#", @packet.QoS1)
let message = client.receive() // 可跨临时断线继续等待
println(message.topic)
client.stop()
})try {
client.stop()
} catch {
@client.DisconnectUncertain => println("broker 可能已发布遗嘱")
}let config = @client.Config::new("127.0.0.1", "stable-client-id")
config.clean_session = false
config.session_store = Some(@client.SessionStore::file("./mqtt-session.bin"))
@async.with_task_group(group => {
let client = @client.Client::connect(config, group)
println("broker session present: \{client.session_present()}")
// subscribe / publish / receive ...
client.disconnect()
})cd D:\Projects\MoonMQTT
moon info
moon fmt
moon check --target native
moon build --target native
moon test --target native
moon run examples/basic --target nativepowershell -ExecutionPolicy Bypass -File tests\interop.ps1
# 若 Mosquitto 不在默认安装目录:
powershell -ExecutionPolicy Bypass -File tests\interop.ps1 -MosquittoDirectory 'D:\path\to\mosquitto\bin'
# 单跑 TLS:powershell -ExecutionPolicy Bypass -File tests\tls_interop.ps1 -MosquittoDirectory 'D:\path\to\mosquitto\bin'pub suberror MqttError {
InvalidConfig(String)
Codec(String)
Protocol(String)
Rejected(ConnackReturnCode)
Closed
TimedOut
InboxFull
Storage(String)
CorruptSession(String)
IncompatibleSessionVersion(Int)
SessionClientIdMismatch(String)
RecoveryUncertain
DisconnectUncertain
TlsValidation(String)
} derive(Debug)pub struct AutoClient {
config : Config
policy : ReconnectPolicy
session : Session
current : Client?
status : AutoStatus
last_error : Error?
stopped : Bool
will_connect_attempted : Bool
earlier_will_uncertain : Bool
task : Task[Unit]?
receive_lock : Mutex
}async fn AutoClient::publish(self : AutoClient, topic : String, payload : Bytes, qos? : QoS, retain? : Bool) -> Unitasync fn[X] AutoClient::start(config : Config, group : TaskGroup[X], policy? : ReconnectPolicy) -> AutoClientpub(all) enum AutoStatus {
Connecting
Connected(Bool)
Retrying(Int, Int)
Failed
Stopped
} derive(Debug)pub struct Client {
config : Config
session : Session
tcp : Tcp
tls : Tls?
write_lock : Mutex
receive_lock : Mutex
messages : Queue[Message]
pending : Map[Int, Pending]
ping_reply : Queue[Unit]
session_present : Bool
closed : Bool
last_error : Error?
disposed : Bool
reader_task : Task[Unit]?
heartbeat_task : Task[Unit]?
}pub(all) struct Config {
host : String
port : Int
client_id : String
username : String?
password : Bytes?
keep_alive_seconds : Int
timeout_ms : Int
max_packet_size : Int
inbox_capacity : Int
clean_session : Bool
session_store : SessionStore?
will : Will?
tls : TlsConfig?
}pub(all) struct ReconnectPolicy {
initial_delay_ms : Int
max_delay_ms : Int
}pub struct SessionStore {
load_data : async () -> Bytes?
save_data : async (Bytes) -> Unit
}pub(all) struct TlsConfig {
ca_file : String?
}