Crate minimq 

Minimq

minimq is a small no_std, no-alloc, async MQTT v5 client for embedded systems.

Use it when your application already has async network I/O and needs one long-lived MQTT session with explicit buffers and reconnect handling.

The main API is Session.

What You Use

• Buffers: caller-owned RX/TX memory
• ConfigBuilder: session configuration
• Io: transport boundary for an established byte stream
• Session: the client you drive
• InboundPublish: output of Session::recv()

Example

use core::net::SocketAddr;
use minimq::{Buffers, ConfigBuilder, ConnectEvent, Error, Session, types::TopicFilter};

async fn run() {
    let rx = &mut [0u8; 256];
    let tx = &mut [0u8; 768];
    let addr: SocketAddr = "127.0.0.1:1883".parse().unwrap();
    let mut session = Session::new(
        ConfigBuilder::new(Buffers::new(rx, tx))
            .client_id("demo")
            .unwrap(),
    );

    loop {
        let io = open_io(addr).await.unwrap();
        match session.connect(io).await.unwrap() {
            ConnectEvent::Connected => {
                session
                    .subscribe(&[TopicFilter::new("demo/in")], &[])
                    .await
                    .unwrap();
            }
            ConnectEvent::Reconnected => {}
        }

        loop {
            match session.recv().await {
                Ok(message) => println!("topic={}", message.topic()),
                Err(Error::Disconnected) => break,
                Err(err) => panic!("{err}"),
            }
        }
    }
}

The attached transport must implement embedded_io_async::Read and embedded_io_async::Write. Ordinary lack of inbound data must keep the read future pending; if the transport returns TimedOut or Interrupted, Session::poll() treats that as transport failure and disconnects the session.

For a TLS connectivity example and for caller-side cooperative driving via external timeouts, see examples/tls_public_broker.rs.

Errors

ConfigBuilder reports setup-time validation failures through ConfigError. Connected session operations report Error:

• broker rejections and invalid inbound MQTT data surface as PeerError
• local buffer and capacity limits surface as ResourceError
• transport failures surface as Error::Transport

Session Model

You provide packet buffers plus an already-established transport, and a loop that explicitly passes that transport into Session::connect() to establish or resume the broker session.

Session::connect() takes ownership of the provided transport and performs the unbounded MQTT CONNECT / CONNACK handshake. Once connected:

• Session::recv() blocks until the next inbound publish arrives or the session is lost.
• Session::poll() blocks until any session progress happens and returns Ok(None) for internal-only progress such as ACK handling, replay, or keepalive traffic.

The session drops the transport again on graceful disconnect, connection failure, or transport/protocol loss.

• ConnectEvent::Connected means the broker created a fresh session. Re-establish subscriptions here.
• ConnectEvent::Reconnected means the broker resumed the existing MQTT session. Existing subscriptions and in-flight QoS state were kept.
• Session::recv() yields one inbound publish.

If Session::recv() or Session::poll() returns Error::Disconnected, the caller decides when to call Session::connect() with a fresh transport again. Other transport/protocol errors already tear down the attached transport locally; callers should handle the error and reconnect rather than retrying recv() or poll() on the same session state.

For cooperative driving:

• use Session::drive() for immediate local progress without waiting for future inbound reads or future session deadlines
• wrap cancel-safe blocking Session::poll() or Session::recv() in an external timeout such as embassy_time::with_timeout() or embassy_time::with_deadline()
• if you need real wall-clock limits, enforce them in the transport’s read, write, and flush futures; using the same budget as minimq’s internal MQTT round-trip timeout keeps keepalive and transport liveness aligned

Buffers

You supply two buffers.

• rx stores one inbound MQTT packet at a time. Size it for the largest inbound publish, including topic, properties, and payload.
• tx stores outbound encodes and retained in-flight state. Size it for the largest outbound packet plus the QoS/session state you want to keep active.

If tx is exhausted, publish() and other outbound operations can return Error::NotReady. Malformed broker varints and undersized local encode buffers are rejected with errors rather than causing panics.

Use Buffers::split() if you prefer one contiguous slab.

Request / Reply

InboundPublish exposes MQTT v5 request/reply properties directly.

• InboundPublish::response_topic()
• InboundPublish::correlation_data()
• InboundPublish::reply()
• InboundPublish::reply_owned()

Transport And Time

minimq uses:

• embedded_io_async for byte I/O
• embassy_time for timing

Re-exports

pub use config::Buffers;

pub use config::ConfigBuilder;

pub use publication::OwnedResponseTarget;

pub use publication::Publication;

Modules

config

Session configuration and caller-owned buffers.

publication

Outbound publish builders and payload adapters.

types

MQTT-specific value types used by the public API. MQTT-specific data types used by the public API.

Structs

InboundPublish

Inbound MQTT PUBLISH surfaced by Session::recv and by Session::drive / Session::poll when they return Some(...).

Op

Handle for one outbound MQTT operation accepted into local session state.

Session

One long-lived MQTT client session.

Will

MQTT will message.

Enums

ConfigError

Configuration errors detected before a session is created.

ConnectEvent

Output of Session::connect.

Error

Possible errors encountered during MQTT operation.

OpStatus

Completion state of a previously accepted outbound operation.

PeerError

Failures caused by broker behavior or invalid inbound MQTT data.

Property

All of the possible properties that MQTT version 5 supports.

PubError

Error returned from Session::publish.

QoS

The quality-of-service for an MQTT message.

ReasonCode

MQTTv5-defined codes that may be returned in response to control packets.

ResourceError

Local capacity and sizing failures.

Retain

The retained status for an MQTT message.

Constants

MQTT_INSECURE_DEFAULT_PORT

Default port number for unencrypted MQTT traffic.

MQTT_SECURE_DEFAULT_PORT

Default port number for encrypted MQTT traffic.

TOPIC_CAPACITY

Fixed-capacity owned MQTT topic storage used by durable configuration.

Traits

Io

Transport trait required by Session.

Type Aliases

PublishError

Publish error type for a specific transport and payload serializer.

SessionError

Session error type for a specific transport.

TopicString

Fixed-capacity owned MQTT topic string.