Client Guide

Client Guide

Complete guide to using the MQTT client library.

---

Creating a Client

use mqtt5::MqttClient;

// Create client with client ID
let client = MqttClient::new("my-device-001");

The client ID must be unique per broker connection.

To create a client pre-configured with connect options, use MqttClient::with_options:

use mqtt5::{MqttClient, ConnectOptions};
use std::time::Duration;

let options = ConnectOptions::new("my-device-001")
    .with_clean_start(true)
    .with_keep_alive(Duration::from_secs(60))
    .with_credentials("mqtt_user", b"secret");

let client = MqttClient::with_options(options);

---

Connecting

Simple Connection

// TCP
client.connect("mqtt://localhost:1883").await?;

// TLS
client.connect("mqtts://broker.example.com:8883").await?;

// WebSocket
client.connect("ws://broker.example.com:8080/mqtt").await?;

// QUIC
client.connect("quic://broker.example.com:14567").await?;

connect returns Result<()>. To inspect the session_present flag from CONNACK, use connect_with_options, which returns a ConnectResult { session_present: bool }.

Connection with Options

use mqtt5::{ConnectOptions, QoS};
use std::time::Duration;

let options = ConnectOptions::new("my-device-001")
    .with_credentials("alice", b"secret")
    .with_keep_alive(Duration::from_secs(30))
    .with_clean_start(true)
    .with_session_expiry_interval(3600); // seconds

let result = client.connect_with_options("mqtt://localhost:1883", options).await?;
println!("Session present: {}", result.session_present);

Notable ConnectOptions builder methods:

Method	Description
with_keep_alive(Duration)	Requested keep-alive interval
with_clean_start(bool)	Start a clean session
with_credentials(user, pass_bytes)	Username + password (password is impl AsRef<[u8]>)
with_session_expiry_interval(u32)	Session expiry, in seconds
with_receive_maximum(u16)	Inbound QoS 1/2 receive maximum
with_will(WillMessage)	Last Will and Testament
with_automatic_reconnect(bool)	Enable/disable auto-reconnect
with_reconnect_delay(initial, max)	Backoff bounds
with_max_reconnect_attempts(u32)	Cap reconnect attempts
with_protocol_version(ProtocolVersion)	Force v5.0 or v3.1.1
with_keepalive_config(KeepaliveConfig)	Ping/timeout tuning

Will Message (Last Will and Testament)

Sent by the broker if the client disconnects unexpectedly. The will is built with WillMessage::new(topic, payload) and attached via with_will:

use mqtt5::{ConnectOptions, WillMessage, QoS};

let will = WillMessage::new("status/my-device", b"offline".to_vec())
    .with_qos(QoS::AtLeastOnce)
    .with_retain(true);

let options = ConnectOptions::new("my-device").with_will(will);

client.connect_with_options("mqtt://localhost:1883", options).await?;

---

Keep-Alive Negotiation (MQTT v5)

The value passed to with_keep_alive is only a request. Per [MQTT-3.2.2-22], if the broker returns a ServerKeepAlive property in CONNACK, that value wins — even overriding a requested Duration::ZERO (disabled). After connecting, read the effective interval with keep_alive():

let effective = client.keep_alive().await; // Duration::ZERO means disabled

The negotiated value is also delivered in the ConnectionEvent::Connected { keep_alive, .. } event (see Connection Events).

---

Publishing Messages

publish and its variants return Result<PublishResult>, where PublishResult is either PublishResult::QoS0 or PublishResult::QoS1Or2 { packet_id }.

QoS 0 - At Most Once

Fire-and-forget, no acknowledgment:

client.publish("sensors/temperature", b"25.5").await?;

QoS 1 - At Least Once

Broker acknowledges receipt:

let result = client.publish_qos1("sensors/temperature", b"25.5").await?;
println!("Published with packet ID: {:?}", result.packet_id());

QoS 2 - Exactly Once

Four-way handshake guarantees exactly-once delivery:

let result = client.publish_qos2("commands/critical", b"execute").await?;

Retained Messages

Last message stored by broker, sent to new subscribers:

client.publish_retain("status/online", b"true").await?;

Publish with Custom Options

Use publish_with_options with PublishOptions to set QoS, retain, and MQTT v5 publish properties (including message expiry):

use mqtt5::{PublishOptions, QoS};
use mqtt5::types::PublishProperties;

let options = PublishOptions {
    qos: QoS::AtLeastOnce,
    retain: false,
    properties: PublishProperties {
        message_expiry_interval: Some(60), // expires after 60 seconds
        ..Default::default()
    },
    ..Default::default()
};

client.publish_with_options("alerts/fire", b"Building A", options).await?;

---

Subscribing

The subscribe callback receives a Message. Its fields are:

Field	Type	Notes
topic	String	Topic the message was published to
payload	Vec<u8>	Message bytes
qos	QoS	Delivery QoS
retain	bool	Whether it was a retained message
properties	MessageProperties	MQTT v5 properties
stream_id	Option<u64>	QUIC stream id, if applicable

subscribe returns Result<(u16, QoS)> — the packet id and the granted QoS.

Simple Subscribe

client.subscribe("sensors/#", |msg| {
    println!("Topic: {}", msg.topic);
    println!("Payload: {}", String::from_utf8_lossy(&msg.payload));
}).await?;

Subscribe with QoS

Use subscribe_with_options and SubscribeOptions to request a QoS or other subscription options:

use mqtt5::{SubscribeOptions, QoS};

let options = SubscribeOptions { qos: QoS::AtLeastOnce, ..Default::default() };

let (packet_id, granted_qos) = client
    .subscribe_with_options("sensors/#", options, |msg| {
        println!("{}: {:?}", msg.topic, msg.payload);
    })
    .await?;

Multiple Subscriptions

client.subscribe("sensors/#", handler.clone()).await?;
client.subscribe("commands/#", handler.clone()).await?;

// Or subscribe to several filters at once (callback must be Clone):
client.subscribe_many(
    vec![("sensors/#", QoS::AtMostOnce), ("commands/#", QoS::AtLeastOnce)],
    handler,
).await?;

Unsubscribe

client.unsubscribe("sensors/#").await?;

// Or several at once:
client.unsubscribe_many(vec!["sensors/#", "commands/#"]).await?;

---

TLS Configuration

The client applies TLS either by storing PEM bytes with set_tls_config (then connecting with a mqtts:// / wss:// / quic:// URL) or by passing a fully-built TlsConfig to connect_with_tls_and_options.

Server Certificate Verification (with client cert / mTLS)

let ca_cert    = std::fs::read("ca.pem")?;
let client_cert = std::fs::read("client.pem")?;
let client_key  = std::fs::read("client-key.pem")?;

// cert_pem, key_pem, ca_cert_pem are all Option<Vec<u8>>
client.set_tls_config(Some(client_cert), Some(client_key), Some(ca_cert)).await;

client.connect("mqtts://broker.example.com:8883").await?;

Pass None for the client cert/key if you only need to trust a custom CA:

let ca_cert = std::fs::read("ca.pem")?;
client.set_tls_config(None, None, Some(ca_cert)).await;
client.connect("mqtts://broker.example.com:8883").await?;

Building a TlsConfig Directly (ALPN, timeouts, etc.)

TlsConfig uses TlsConfig::new(addr, hostname) plus load_* / with_* methods — there is no builder(). The address and hostname are taken from the TlsConfig (not from a URL) when using connect_with_tls_and_options:

use mqtt5::transport::tls::TlsConfig;
use mqtt5::ConnectOptions;
use std::net::ToSocketAddrs;

let host = "broker.example.com";
let addr = format!("{host}:8883").to_socket_addrs()?.next().unwrap();

let mut tls = TlsConfig::new(addr, host);
tls.load_ca_cert_pem("ca.pem")?;
tls.load_client_cert_pem("client.pem")?;
tls.load_client_key_pem("client-key.pem")?;
let tls = tls.with_alpn_protocols(&["mqtt"]);

let options = ConnectOptions::new("my-device");
client.connect_with_tls_and_options(tls, options).await?;

Certificates can also be loaded from bytes: load_ca_cert_pem_bytes, load_client_cert_pem_bytes, load_client_key_pem_bytes, and the DER variants load_client_cert_der_bytes / load_client_key_der_bytes.

Skip Certificate Verification (Development Only)

client.set_insecure_tls(true).await;
client.connect("mqtts://test-broker:8883").await?;

---

Auto-Reconnection

The client automatically reconnects with exponential backoff. Reconnection is enabled by default (ReconnectConfig::default().enabled == true).

use std::time::Duration;

let options = ConnectOptions::new("my-device")
    .with_automatic_reconnect(true)
    .with_reconnect_delay(Duration::from_secs(1), Duration::from_secs(30))
    .with_max_reconnect_attempts(10);

client.connect_with_options("mqtt://localhost:1883", options).await?;

On a successful reconnect the client restores its subscriptions automatically, so message callbacks continue firing without re-subscribing.

A successful explicit disconnect() stops auto-reconnection. After you call disconnect(), the client will not attempt to reconnect until you connect again.

Default backoff (ReconnectConfig):

• initial_delay: 1s
• max_delay: 60s
• backoff_factor: 2.0 (stored as backoff_factor_tenths = 20)
• max_attempts: None (retry forever)

Reconnection Flow

Connected ──> Disconnected ──> Reconnecting ──> Connected
                                  1s  2s  4s ... (backoff, capped at max_delay)

---

Connection Events

Register a single handler for all connection lifecycle events with on_connection_event. (There are no separate on_connect / on_disconnect methods.)

use mqtt5::{ConnectionEvent, DisconnectReason};

client.on_connection_event(|event| {
    match event {
        ConnectionEvent::Connecting => println!("Connecting..."),
        ConnectionEvent::Connected { session_present, keep_alive } => {
            println!("Connected! session_present={session_present}, keep_alive={keep_alive:?}");
        }
        ConnectionEvent::Disconnected { reason } => {
            println!("Disconnected: {reason:?}");
        }
        ConnectionEvent::Reconnecting { attempt } => {
            println!("Reconnecting, attempt {attempt}");
        }
        ConnectionEvent::ReconnectFailed { error } => {
            println!("Reconnect failed: {error}");
        }
    }
}).await?;

// Check connection status at any time
if client.is_connected().await {
    println!("Currently connected");
}

DisconnectReason variants: ClientInitiated, ServerClosed, NetworkError(String), ProtocolError(String), KeepAliveTimeout, AuthFailure.

You can also register an error handler with on_error:

client.on_error(|err| eprintln!("client error: {err}")).await?;

---

Enhanced Authentication

MQTT v5.0 supports challenge-response authentication (e.g. SCRAM-SHA-256). An auth handler is installed on the client with set_auth_handler (not on ConnectOptions).

SCRAM-SHA-256

use mqtt5::ScramSha256AuthHandler;

let handler = ScramSha256AuthHandler::new("username", "password");
client.set_auth_handler(handler).await;

client.connect("mqtt://localhost:1883").await?;

Plain (SASL)

use mqtt5::PlainAuthHandler;

let handler = PlainAuthHandler::new("username", "password")
    .with_authzid("admin"); // Optional authorization identity

client.set_auth_handler(handler).await;

JWT Authentication

use mqtt5::JwtAuthHandler;

let handler = JwtAuthHandler::new("eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...");
client.set_auth_handler(handler).await;

Custom Auth Handler

Implement the AuthHandler trait. handle_challenge returns an AuthResponse (Continue(Vec<u8>), Success, or Abort(String)); override initial_response to send data in the initial CONNECT:

use mqtt5::{AuthHandler, AuthResponse};
use mqtt5::client::auth_handler::AuthFuture;

struct MyAuthHandler;

impl AuthHandler for MyAuthHandler {
    fn handle_challenge<'a>(
        &'a self,
        auth_method: &'a str,
        challenge_data: Option<&'a [u8]>,
    ) -> AuthFuture<'a, AuthResponse> {
        Box::pin(async move {
            let _ = (auth_method, challenge_data);
            Ok(AuthResponse::Success)
        })
    }

    fn initial_response<'a>(
        &'a self,
        _auth_method: &'a str,
    ) -> AuthFuture<'a, Option<Vec<u8>>> {
        Box::pin(async move { Ok(None) })
    }
}

To trigger MQTT v5 re-authentication on an established connection:

client.reauthenticate().await?;

---

Session Persistence

Resume a session after reconnection:

let options = ConnectOptions::new("my-device")
    .with_clean_start(false)              // Resume existing session
    .with_session_expiry_interval(3600);  // seconds

client.connect_with_options("mqtt://localhost:1883", options).await?;

With clean_start = false and a non-zero session expiry:

• Previous subscriptions are restored
• Queued QoS 1/2 messages are delivered
• In-flight messages are resumed

---

QUIC Transport Features

When connected over QUIC (quic://), extra capabilities are available (require the transport-quic feature).

Connection Migration

Rebinds the QUIC endpoint to a new local UDP socket — useful for mobile clients that change networks:

client.migrate().await?;

0-RTT Early Data

Enable 0-RTT before connecting; after a successful resumption you can check whether 0-RTT data was accepted:

client.set_quic_early_data(true).await;
client.connect("quic://broker.example.com:14567").await?;

if client.was_zero_rtt().await {
    println!("0-RTT early data was accepted");
}

Other QUIC tuning setters: set_quic_stream_strategy, set_quic_datagrams, set_quic_flow_headers, set_quic_max_streams, set_quic_connect_timeout.

---

Error Handling

use mqtt5::MqttError;

match client.publish_qos1("topic", b"data").await {
    Ok(result) => println!("Published: {result:?}"),
    Err(MqttError::NotConnected) => println!("Not connected to broker"),
    Err(MqttError::PublishFailed(reason)) => println!("Publish rejected: {reason:?}"),
    Err(MqttError::Timeout) => println!("Operation timed out"),
    Err(e) => println!("Other error: {e:?}"),
}

MqttError::PublishFailed and MqttError::ConnectionRefused each carry a ReasonCode. NotConnected, AlreadyConnected, Timeout, and KeepAliveTimeout are unit variants.

---

Graceful Disconnect

// Sends a DISCONNECT packet and stops auto-reconnection
client.disconnect().await?;

---

Complete Example

use mqtt5::{MqttClient, ConnectOptions, WillMessage, QoS};
use std::time::Duration;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = MqttClient::new("sensor-device-001");

    let will = WillMessage::new("status/sensor-device-001", b"offline".to_vec())
        .with_qos(QoS::AtLeastOnce);

    let options = ConnectOptions::new("sensor-device-001")
        .with_keep_alive(Duration::from_secs(30))
        .with_automatic_reconnect(true)
        .with_will(will);

    client.connect_with_options("mqtt://localhost:1883", options).await?;

    client.subscribe("commands/sensor-device-001/#", |msg| {
        println!("Command received: {}", String::from_utf8_lossy(&msg.payload));
    }).await?;

    loop {
        let temp = read_temperature();
        client.publish_qos1("sensors/temperature", temp.as_bytes()).await?;
        tokio::time::sleep(Duration::from_secs(10)).await;
    }
}

fn read_temperature() -> String {
    "25.5".to_string()
}

---

Next Steps

• QUIC Transport - Modern UDP-based transport
• AWS IoT Integration - Cloud connectivity
• Mock Testing - Unit test your MQTT code