Embedded Guide

Embedded Guide (no_std)

Use the mqtt5-protocol crate on bare-metal embedded systems.

---

Overview

mqtt5-protocol is a platform-agnostic MQTT v5.0 and v3.1.1 implementation. It is the shared foundation used by both mqtt5 (native) and mqtt5-wasm (browser), and it compiles on no_std targets (requires alloc).

It provides:

• Full MQTT v5.0 (and v3.1.1) packet encoding/decoding
• Session building blocks — flow-control config, message queues, subscription manager, topic-alias manager, and limits
• Topic validation and topic matching
• Platform-agnostic time types with an injectable time source

It does not include:

• Networking (you provide the transport)
• Async runtime (integrate with your embedded runtime)

┌─────────────────────────────────────────────────────────────────┐
│                     Your Embedded Application                    │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌───────────────┐    ┌───────────────┐    ┌───────────────┐   │
│  │ mqtt5-protocol│    │  Your Network │    │ Your Runtime  │   │
│  │ (packets)     │<──>│  Driver       │<──>│ (embassy/RTIC)│   │
│  └───────────────┘    └───────────────┘    └───────────────┘   │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

---

Supported Targets

Target	Platform	Notes
x86_64-*, aarch64-*	Desktop/Server	Full std support
wasm32-unknown-unknown	WebAssembly	Browser / Node.js
thumbv7em-none-eabihf	ARM Cortex-M4F	STM32F4, nRF52, etc.
riscv32imc-unknown-none-elf	RISC-V	ESP32-C3, BL602, etc.

Other no_std targets that provide alloc (e.g. thumbv6m-none-eabi for Cortex-M0+) also work; see the single-core note below.

---

Installation

Cargo.toml

[dependencies]
mqtt5-protocol = { version = "0.14", default-features = false }

The only feature flag is std (default on); disabling default features gives you the no_std build. alloc is always required.

For Single-Core MCUs

Single-core MCUs need portable-atomic's single-core assumption. Add to .cargo/config.toml:

[target.riscv32imc-unknown-none-elf]
rustflags = ["--cfg", "portable_atomic_unsafe_assume_single_core"]

[target.thumbv6m-none-eabi]
rustflags = ["--cfg", "portable_atomic_unsafe_assume_single_core"]

---

Time Source

On no_std targets there is no OS clock, so Instant::now() and SystemTime::now() rely on an externally supplied time source. The time module exposes free functions (there is no trait to implement):

Function	Purpose
set_time_source(monotonic, epoch)	Initialize both time sources at startup
update_monotonic_time(millis)	Update the monotonic clock (for Instant)
update_epoch_time(millis)	Update the wall clock (for SystemTime)

Initialize at Startup

#![no_std]
extern crate alloc;

use mqtt5_protocol::time::set_time_source;

fn init_time() {
    // monotonic_millis, epoch_millis — usually 0 at boot
    set_time_source(0, 0);
}

Update Periodically

Call from your timer interrupt or main loop:

use mqtt5_protocol::time::update_monotonic_time;

#[interrupt]
fn TIM2() {
    static mut MILLIS: u64 = 0;
    unsafe {
        *MILLIS += 1;
        update_monotonic_time(*MILLIS);
    }
}

Using Time

use mqtt5_protocol::time::{Instant, Duration};

let start = Instant::now();
// ... do work ...
if start.elapsed() > Duration::from_secs(5) {
    // Timeout handling
}

Implementation notes: the time source uses AtomicU32 pairs (32-bit target friendly) with SeqCst ordering, and returns 0 until initialized.

---

Packet Encoding/Decoding

The concrete per-packet types (ConnectPacket, PublishPacket, …) implement the MqttPacket trait, which provides encode/encode_body. A decoded wire packet is returned as the Packet enum.

Encode a CONNECT Packet

use mqtt5_protocol::packet::connect::ConnectPacket;
use mqtt5_protocol::packet::MqttPacket; // trait providing encode()
use mqtt5_protocol::types::ConnectOptions;
use mqtt5_protocol::time::Duration;
use bytes::BytesMut;

fn create_connect_packet() -> BytesMut {
    let options = ConnectOptions::new("my-device-001")
        .with_keep_alive(Duration::from_secs(60))
        .with_clean_start(true);

    let connect = ConnectPacket::new(options);

    let mut buf = BytesMut::with_capacity(128);
    connect.encode(&mut buf).unwrap();
    buf
}

Note: with_keep_alive takes a Duration, not a raw integer.

Decode an Incoming Packet

Packet::decode reads from anything implementing bytes::Buf and advances it:

use mqtt5_protocol::packet::Packet;
use mqtt5_protocol::ReasonCode;
use bytes::Bytes;

fn handle_packet(data: &[u8]) {
    let mut bytes = Bytes::copy_from_slice(data);

    match Packet::decode(&mut bytes) {
        Ok(Packet::ConnAck(connack)) => {
            if connack.reason_code == ReasonCode::Success {
                // Connected successfully
            }
        }
        Ok(Packet::Publish(publish)) => {
            let topic = &publish.topic_name;
            let payload = &publish.payload;
            let _ = (topic, payload);
        }
        Ok(Packet::PubAck(_puback)) => {
            // QoS 1 acknowledged
        }
        Ok(_) => {}
        Err(_e) => {
            // Handle decode error
        }
    }
}

The full-packet enum is Packet (variants ConnAck, Publish, PubAck, …). MqttPacket is the trait, not the enum. reason_code is a ReasonCode, compared against variants such as ReasonCode::Success — not an integer.

---

Session Building Blocks

mqtt5_protocol::session provides reusable pieces rather than a single "session" object: SubscriptionManager, MessageQueue, TopicAliasManager, LimitsManager, and FlowControlConfig.

Track Subscriptions

use mqtt5_protocol::session::{SubscriptionManager, Subscription};

let mut subs = SubscriptionManager::new();
subs.add("sensors/#".to_string(), Subscription::default())?;

let matches = subs.matching_subscriptions("sensors/temp");

Flow-Control Config

use mqtt5_protocol::session::FlowControlConfig;
use mqtt5_protocol::time::Duration;

let flow = FlowControlConfig {
    enable_backpressure: true,
    backpressure_timeout: Some(Duration::from_secs(30)),
    max_pending_queue_size: 1000,
    in_flight_timeout: Duration::from_secs(60),
};

---

Embassy Integration Example

Using with the Embassy async runtime:

#![no_std]
#![no_main]

extern crate alloc;

use embassy_executor::Spawner;
use embassy_net::tcp::TcpSocket;
use mqtt5_protocol::packet::{connect::ConnectPacket, Packet, MqttPacket};
use mqtt5_protocol::types::ConnectOptions;
use mqtt5_protocol::ReasonCode;
use bytes::{Bytes, BytesMut};

#[embassy_executor::main]
async fn main(_spawner: Spawner) {
    // Initialize hardware, network stack, socket ...
    let mut socket = TcpSocket::new(stack, rx_buffer, tx_buffer);
    socket.connect(broker_addr).await.unwrap();

    // Build and send CONNECT
    let connect = ConnectPacket::new(ConnectOptions::new("embassy-device"));
    let mut buf = BytesMut::with_capacity(128);
    connect.encode(&mut buf).unwrap();
    socket.write_all(&buf).await.unwrap();

    // Read and decode the response
    let mut rx_buf = [0u8; 256];
    let n = socket.read(&mut rx_buf).await.unwrap();
    let mut response = Bytes::copy_from_slice(&rx_buf[..n]);

    if let Ok(Packet::ConnAck(connack)) = Packet::decode(&mut response) {
        if connack.reason_code == ReasonCode::Success {
            // Connected!
        }
    }
}

---

RTIC Integration Example

Using with the RTIC framework — drive the time source from a periodic timer:

#![no_std]
#![no_main]

use rtic::app;
use mqtt5_protocol::time::{set_time_source, update_monotonic_time};

#[app(device = stm32f4xx_hal::pac)]
mod app {
    use super::*;

    #[shared]
    struct Shared {}

    #[local]
    struct Local {
        timer: Timer<TIM2>,
    }

    #[init]
    fn init(ctx: init::Context) -> (Shared, Local) {
        set_time_source(0, 0);

        let mut timer = Timer::new(ctx.device.TIM2, 1.kHz(), &clocks);
        timer.listen(Event::Update);

        (Shared {}, Local { timer })
    }

    #[task(binds = TIM2, local = [timer, ticks: u64 = 0])]
    fn timer_tick(ctx: timer_tick::Context) {
        ctx.local.timer.clear_interrupt(Event::Update);
        *ctx.local.ticks += 1;
        update_monotonic_time(*ctx.local.ticks);
    }
}

---

Memory Considerations

Heap Usage

The crate uses alloc for topic strings, payload buffers, and properties. Configure an allocator:

use embedded_alloc::Heap;

#[global_allocator]
static HEAP: Heap = Heap::empty();

fn init_heap() {
    unsafe { HEAP.init(HEAP_START, 4096) };
}

Stack Usage

Packet encoding/decoding uses stack buffers — ensure adequate stack. Small control packets need a few dozen bytes; CONNECT is on the order of a few hundred; PUBLISH scales with payload size.

---

Build Commands

# ARM Cortex-M4F (hardware FPU)
cargo build --target thumbv7em-none-eabihf -p mqtt5-protocol --no-default-features

# ARM Cortex-M0+ (no FPU)
cargo build --target thumbv6m-none-eabi -p mqtt5-protocol --no-default-features

# RISC-V (ESP32-C3)
cargo build --target riscv32imc-unknown-none-elf -p mqtt5-protocol --no-default-features

---

Troubleshooting

Linker Errors with Atomics

undefined symbol: __atomic_*

Add the single-core portable-atomic cfg to .cargo/config.toml:

rustflags = ["--cfg", "portable_atomic_unsafe_assume_single_core"]

Out of Memory

• Reduce buffer sizes and payloads.
• Grow the allocator heap.

Time Not Working

• Ensure set_time_source() is called at init.
• Ensure update_monotonic_time() is called regularly from a timer.
• Verify the timer interrupt is enabled.