🥮 mooncakes.io

    moon-mqtt-client

    Native asynchronous MQTT 3.1.1 client with QoS 0/1 and clean-session reconnect

    Download zip
    Version
    0.1.0
    License
    Apache-2.0
    Repository
    Last updated
    4 hours ago
    Downloads
    2

    Dependencies

    #moon-mqtt-client

    native client

    A native asynchronous MQTT 3.1.1 client for MoonBit. Connect to an existing MQTT broker, subscribe to device or application events, and publish commands and state without writing a socket loop for each application.

    The packet codec is provided by zbhzs1/moonbit-mqtt; this package supplies the connection lifecycle, TLS transport, request tracking, heartbeat and clean-session reconnection on top of moonbitlang/async. It is an early implementation, not a certified MQTT conformance implementation.

    #Scope

    • Native TCP and server-authenticated TLS (system roots or a custom PEM CA).
    • MQTT 3.1.1, QoS 0 and 1, retained messages, Last Will, username/password.
    • Subscribe and unsubscribe acknowledgements, including per-topic rejection.
    • Bounded send queue, event queue, packet size and concurrent requests.
    • CleanSession=true only: reconnect creates a new session and restores confirmed subscriptions. A Connected(generation) event follows subscription restoration.
    • Callback-scoped tasks and sockets. Normal callback return or disconnect() sends DISCONNECT; callback failure or cancellation closes the transport abruptly.

    No QoS 2, MQTT 5, persistent sessions, offline queue, cross-connection retransmit, client-certificate authentication, WebSocket, browser or microcontroller target.

    #Build from source

    Use the native MoonBit toolchain; this checkout was developed with moon 0.1.20260904 and moonc v0.10.12+1634b282e (2026-09-07).

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

    scripts/moon.sh uses a local .tools/moon installation or MOON_HOME when provided. Toolchains and build outputs are not part of the source distribution. For registry releases, install the package with:

    moon add Strangelight-Merser/moon-mqtt-client

    See GitHub Releases for available versions and their verification results.

    #API

    The following is the callback shape used by the runnable examples. Import this package as @mqtt and moonbitlang/core/encoding/utf8 as @utf8.

    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)
        }
        // Consume Connected, Disconnected and MessageReceived events here.
        // A long-running subscriber must continually drain next_event().
      })
    }

    with_client waits for the first successful connection before invoking the callback. An initial connection/CONNACK/TLS failure is returned to the caller. After a connection has been established, failures trigger bounded retries. wait_connected() can wait through a reconnect; publish/subscribe/unsubscribe while disconnected fail with NotConnected, rather than entering an offline queue.

    Use TlsMode::SystemRoots with port 8883 for public trust or TlsMode::CustomCA("path/to/ca.pem") for a private CA. The configured host is also the verified TLS hostname. There is no option to disable verification.

    #Delivery and failure semantics

    ResultWhat it establishes
    QoS 0 publish returnsThe transport write completed; no broker acknowledgement exists.
    QoS 1 publish returnsA matching PUBACK arrived on that connection. It does not establish downstream processing or a physical action.
    NotSentA queued request failed before its write started.
    OutcomeUnknownA write began but the operation was not confirmed. Partial writes and lost ACKs are included.
    Backpressure from a requestThe send or inflight limit prevented accepting that request.
    Event/control queue overflowThe client terminates with an error instead of silently dropping messages.

    An acknowledgement timeout closes the connection and ends all pending requests. Old identifiers never cross into the new connection. The application chooses whether to retry an uncertain operation; use idempotent state-setting commands or application command IDs where appropriate.

    Incoming QoS 1 is acknowledged after acceptance into the bounded event queue, not after application processing. The queue is volatile. QoS 1 duplicates are possible. Reconnection with a clean session can lose messages during the gap; a broker may replay retained state on resubscription. The library does not promise exactly-once processing, durable delivery or uninterrupted subscriptions.

    Default limits: 64 queued sends, 128 queued events, 32 pending operations, 65,536 bytes per packet, 5-second connect/write/ACK timeout, 30-second keepalive, 10 consecutive reconnect attempts with a delay growing from 250 ms to 5 seconds plus a small deterministic jitter. Confirmed subscription filters remain in memory until unsubscribed; keep their set bounded in the application.

    #Runnable scenarios

    中文入口：从这里开始。

    See docs/SCENARIOS.md for complete inputs, rules, outputs and failure boundaries for three intended uses:

    1. Temperature control with hysteresis, retained state and availability.
    2. A bounded Frigate event deduplicator which alerts on completed person events.
    3. A ROS bridge JSON/primitive contract with command validation and receipts.

    The examples use fixtures. They do not claim a deployed camera, robot or Zigbee integration. The temperature example connects to 127.0.0.1:1883:

    moon run examples/temperature_controller --target native

    #Verification

    python3 -m venv .venv
.venv/bin/pip install -r tests/integration/requirements.txt
# Install Mosquitto and OpenSSL using your OS package manager, then:
PYTHON=.venv/bin/python ./tests/integration/run.sh

    Run every local check with ./scripts/check.sh.

    The integration suite uses independent Mosquitto and Eclipse Paho processes, loopback-only listeners, temporary certificates and packet fault injection. See docs/VALIDATION.md for the actual executed results and remaining gaps. A workflow definition is not evidence of a completed hosted CI run.

    #License and upstream work

    Apache-2.0. See NOTICE. The dependencies remain separate packages with their own attribution and licenses; this project does not claim authorship of MQTT wire encoding, the async runtime or TLS implementation.

    ClientError

    source
    pub(all) suberror ClientError {
      InvalidConfig(String)
      ProtocolError(String)
      NotConnected
      Closed
      Backpressure(String)
      NotSent(String)
      OutcomeUnknown(String)
      ConnectionRefused(String)
      ReconnectExhausted(String)
    } derive(
    Debug    )

    Client

    source
    type Client

    Client::disconnect

    source
    async fn Client::disconnect(self : Client) -> Unit

    Stop reconnect, send DISCONNECT when possible, and await worker cleanup.

    Client::next_event

    source
    async fn Client::next_event(self : Client) -> Event

    Client::publish

    source
    async fn Client::publish(self : Client, topic : String, payload : Bytes, qos? : QoS, retain? : Bool) -> Unit

    Client::subscribe

    source
    async fn Client::subscribe(self : Client, topics : Array[Subscription]) -> Array[SubscriptionResult]

    Client::unsubscribe

    source
    async fn Client::unsubscribe(self : Client, topics : Array[String]) -> Unit

    Client::wait_connected

    source
    async fn Client::wait_connected(self : Client) -> Unit

    Config

    source
    pub(all) struct Config {
      host : String
      port : Int
      client_id : String
      tls : TlsMode
      username : String?
      password : Bytes?
      will : Will?
      keep_alive_secs : Int
      connect_timeout_ms : Int
      ack_timeout_ms : Int
      send_capacity : Int
      receive_capacity : Int
      max_inflight : Int
      max_packet_size : Int
      reconnect_delay_ms : Int
      max_reconnect_delay_ms : Int
      reconnect_attempts : Int
    }

    Config::new

    source
    fn Config::new(host : String, client_id : String, port? : Int, tls? : TlsMode, username? : String?, password? : Bytes?, will? : Will?, keep_alive_secs? : Int, connect_timeout_ms? : Int, ack_timeout_ms? : Int, send_capacity? : Int, receive_capacity? : Int, max_inflight? : Int, max_packet_size? : Int, reconnect_delay_ms? : Int, max_reconnect_delay_ms? : Int, reconnect_attempts? : Int) -> Config

    Event

    source
    pub(all) enum Event {
      Connected(Int)
      Disconnected(Int, String)
      MessageReceived(Message)
    } derive(
    Debug    )

    Message

    source
    pub(all) struct Message {
      topic : String
      payload : Bytes
      qos : QoS
      retain : Bool
      dup : Bool
      generation : Int
    } derive(
    Debug    )

    QoS

    source
    pub(all) enum QoS {
      AtMostOnce
      AtLeastOnce
    } derive(Eq, 
    Debug    )

    Subscription

    source
    pub(all) struct Subscription {
      topic : String
      qos : QoS
    } derive(Eq, 
    Debug    )

    SubscriptionResult

    source
    pub(all) enum SubscriptionResult {
      Granted(QoS)
      Rejected
    } derive(Eq, 
    Debug    )

    TlsMode

    source
    pub(all) enum TlsMode {
      Plain
      SystemRoots
      CustomCA(String)
    } derive(
    Debug    )

    Will

    source
    pub(all) struct Will {
      topic : String
      payload : Bytes
      qos : QoS
      retain : Bool
    } derive(
    Debug    )

    with_client

    source
    async fn[T] with_client(config : Config, action : async (Client) -> T) -> T

    Own all client tasks within the callback's lifetime.

    Source Files

    Powered by MoonBit

    Site sourceReport issuePackagesBuild queueSkillsStatistics

    © 2026 mooncakes.io