Skip to content
Plugins (190)

mqtt

stable

Encode and decode MQTT 3.1.1 protocol packets for publish/subscribe messaging, including authentication, will messages, and QoS acknowledgement flows.

use plugin mqtt
21 functions Networking
Runs with the native Zolo host

Plugin snippets can be edited in the Playground, but native capabilities are executed locally with the Zolo CLI.

/ filter jk navigate Esc clear
Functions (21)
  1. encode_connect Encode a CONNECT packet
  2. encode_connect_with_auth Encode CONNECT with username and password
  3. encode_connect_with_will Encode CONNECT with a last-will message
  4. encode_publish Encode a PUBLISH packet
  5. encode_publish_with_options Encode PUBLISH with retain, dup, and packet ID
  6. encode_subscribe Encode a SUBSCRIBE packet for one topic
  7. encode_subscribe_multiple Encode SUBSCRIBE for multiple topics
  8. encode_unsubscribe Encode an UNSUBSCRIBE packet
  9. encode_disconnect Encode a DISCONNECT packet
  10. encode_pingreq Encode a PINGREQ packet
  11. encode_puback Encode a PUBACK acknowledgement
  12. encode_pubrec Encode a PUBREC (QoS 2 received) packet
  13. encode_pubrel Encode a PUBREL (QoS 2 release) packet
  14. encode_pubcomp Encode a PUBCOMP (QoS 2 complete) packet
  15. decode_connack Decode a CONNACK response packet
  16. decode_publish Decode a PUBLISH packet
  17. decode_suback Decode a SUBACK response packet
  18. decode_packet_type Decode the first byte of any MQTT packet
  19. packet_type_name Convert packet type integer to name string
  20. remaining_length_encode Encode an integer as MQTT variable-length field
  21. remaining_length_decode Decode an MQTT variable-length field from bytes

From import to result

  1. 01Import only what you use
  2. 02Copy the smallest example
  3. 03Run it with the Zolo CLI

Overview

mqtt is a dependency-free codec for the MQTT 3.1.1 wire protocol. It does not open sockets or manage a connection — instead, each encode_* function builds the raw bytes of one control packet, and each decode_* function parses incoming bytes back into a table. You drive the actual transport yourself (typically over the net plugin or a raw TCP stream), feeding encoded packets out and handing received bytes to the decoders. This keeps the plugin small and lets you implement exactly the publish/subscribe flow your broker needs.

The mental model is a request/response handshake expressed as byte buffers: send a CONNECT, read a CONNACK; send a SUBSCRIBE, read a SUBACK; send or receive PUBLISH frames, and for QoS 1/2 exchange the matching PUBACK/PUBREC/PUBREL/PUBCOMP acknowledgements. Use decode_packet_type to peek at any inbound frame's first byte and dispatch it to the right decoder. Use it whenever you need an embedded, allocation-light MQTT client without pulling in a full broker library.

Common patterns

Open a session: encode a CONNECT, send it, then decode the broker's CONNACK reply.

use plugin mqtt::{encode_connect, decode_connack}

let connect = encode_connect("device-01", true)
// ... send `connect` over your transport, then read the reply bytes ...
let reply = [0x20, 0x02, 0x00, 0x00]
let ack = decode_connack(reply)
print("accepted={ack["return_code"] == 0}")

Round-trip a message: encode a PUBLISH and decode it back into its fields.

use plugin mqtt::{encode_publish, decode_publish}

let frame = encode_publish("sensors/temp", "23.5", 1)
let msg = decode_publish(frame)
print("{msg["topic"]} = {msg["payload"]} (QoS {msg["qos"]})")

Dispatch any inbound frame by peeking at its type byte before decoding.

use plugin mqtt::{decode_packet_type, decode_publish, decode_suback}

fn handle(bytes) {
  let info = decode_packet_type(bytes)
  if info["type_name"] == "PUBLISH" {
    let m = decode_publish(bytes)
    print("message on {m["topic"]}")
  } else if info["type_name"] == "SUBACK" {
    let s = decode_suback(bytes)
    print("subscribed, packet {s["packet_id"]}")
  }
}

encode_connect(client_id, clean_session) → any

Encode a CONNECT packet

Encodes a minimal MQTT 3.1.1 CONNECT packet with the given client ID and clean-session flag. Uses a 60-second keep-alive with no username, password, or will.

use plugin mqtt::{encode_connect, decode_connack}

let pkt = encode_connect("my-client-1", true)
print("CONNECT bytes: {pkt}")

encode_connect_with_auth(client_id, clean_session, keepalive, username, password) → any

Encode CONNECT with username and password

Encodes a CONNECT packet with username and password credentials. Sets the username and password flags in the connect flags byte.

use plugin mqtt::{encode_connect_with_auth}

let pkt = encode_connect_with_auth("device-01", true, 30, "admin", "secret")
print(pkt)

encode_connect_with_will(client_id, clean_session, keepalive, will_topic, will_message, will_qos, will_retain) → any

Encode CONNECT with a last-will message

Encodes a CONNECT packet with a last-will-and-testament message, which the broker publishes if the client disconnects unexpectedly.

use plugin mqtt::{encode_connect_with_will}

let pkt = encode_connect_with_will(
  "sensor-01", true, 60,
  "sensors/sensor-01/status", "offline", 1, true
)

encode_publish(topic, payload, qos) → any

Encode a PUBLISH packet

Encodes a PUBLISH packet with a topic, string payload, and QoS level (0, 1, or 2). For QoS > 0, uses packet ID 1.

use plugin mqtt::{encode_publish, decode_publish}

let pkt = encode_publish("sensors/temperature", "23.5", 0)
let decoded = decode_publish(pkt)
print(decoded["topic"])
print(decoded["payload"])

encode_publish_with_options(topic, payload, qos, retain, dup, packet_id) → any

Encode PUBLISH with retain, dup, and packet ID

Encodes a PUBLISH packet with full control over all MQTT header flags: retain bit, duplicate delivery flag, and explicit packet ID for QoS > 0.

use plugin mqtt::{encode_publish_with_options}

let pkt = encode_publish_with_options("home/lights", "on", 1, true, false, 42)

Set dup to true when re-sending an unacknowledged QoS 1 message with the same packet ID, so the broker knows it may be a duplicate:

use plugin mqtt::{encode_publish_with_options, decode_publish}

let resend = encode_publish_with_options("home/lights", "on", 1, false, true, 42)
let msg = decode_publish(resend)
print("dup={msg["dup"]} packet_id={msg["packet_id"]}")

encode_subscribe(topic, packet_id, qos) → any

Encode a SUBSCRIBE packet for one topic

Encodes a SUBSCRIBE packet for a single topic filter with the requested QoS level.

use plugin mqtt::{encode_subscribe, decode_suback}

let pkt = encode_subscribe("sensors/#", 1, 1)

encode_subscribe_multiple(topics, packet_id) → any

Encode SUBSCRIBE for multiple topics

Encodes a SUBSCRIBE packet for multiple topic filters in a single packet. Each entry in topics is a table with topic and qos fields.

use plugin mqtt::{encode_subscribe_multiple}

let topics = [
  #{"topic": "sensors/temp", "qos": 1},
  #{"topic": "sensors/humidity", "qos": 0},
  #{"topic": "alerts/#", "qos": 2}
]
let pkt = encode_subscribe_multiple(topics, 5)

encode_unsubscribe(topics, packet_id) → any

Encode an UNSUBSCRIBE packet

Encodes an UNSUBSCRIBE packet for one or more topic filters.

use plugin mqtt::{encode_unsubscribe}

let pkt = encode_unsubscribe(["sensors/temp", "sensors/humidity"], 6)

encode_disconnect() → any

Encode a DISCONNECT packet

Returns the 2-byte DISCONNECT packet (0xE0 0x00). Send this before closing the TCP connection for a clean disconnect.

use plugin mqtt::{encode_disconnect}

let pkt = encode_disconnect()

encode_pingreq() → any

Encode a PINGREQ packet

Returns the 2-byte PINGREQ packet used to keep the connection alive. The broker responds with PINGRESP.

use plugin mqtt::{encode_pingreq}

let pkt = encode_pingreq()

encode_puback(packet_id) → any

Encode a PUBACK acknowledgement

Encodes a PUBACK acknowledgement for a QoS 1 PUBLISH. Send this after receiving and processing a QoS 1 message.

use plugin mqtt::{encode_puback}

let ack = encode_puback(42)

encode_pubrec(packet_id) → any

Encode a PUBREC (QoS 2 received) packet

Encodes a PUBREC packet, the first step of the QoS 2 acknowledgement flow (received).

use plugin mqtt::{encode_pubrec, encode_pubrel, encode_pubcomp}

let rec = encode_pubrec(10)
let rel = encode_pubrel(10)
let comp = encode_pubcomp(10)

encode_pubrel(packet_id) → any

Encode a PUBREL (QoS 2 release) packet

Encodes a PUBREL packet, the second step of the QoS 2 flow (release).

use plugin mqtt::{encode_pubrel}

let pkt = encode_pubrel(10)

encode_pubcomp(packet_id) → any

Encode a PUBCOMP (QoS 2 complete) packet

Encodes a PUBCOMP packet, the final step of the QoS 2 flow (complete).

use plugin mqtt::{encode_pubcomp}

let pkt = encode_pubcomp(10)

decode_connack(bytes) → table

Decode a CONNACK response packet

Decodes a CONNACK response into {session_present, return_code}. A return_code of 0 means the connection was accepted.

use plugin mqtt::{decode_connack}

let connack_bytes = [0x20, 0x02, 0x00, 0x00]
let result = decode_connack(connack_bytes)
print("accepted={result["return_code"] == 0}")
print("session_present={result["session_present"]}")

decode_publish(bytes) → table

Decode a PUBLISH packet

Decodes a PUBLISH packet into {topic, payload, qos, retain, dup, packet_id}.

use plugin mqtt::{encode_publish, decode_publish}

let pkt = encode_publish("home/temp", "22.1", 1)
let msg = decode_publish(pkt)
print("{msg["topic"]}: {msg["payload"]} (QoS {msg["qos"]})")

decode_suback(bytes) → table

Decode a SUBACK response packet

Decodes a SUBACK response into {packet_id, return_codes} where each return code is the granted QoS level (0, 1, 2) or 0x80 for failure.

use plugin mqtt::{decode_suback}

let suback_bytes = [0x90, 0x03, 0x00, 0x01, 0x01]
let result = decode_suback(suback_bytes)
print("packet_id={result["packet_id"]}")

Pair it with encode_subscribe_multiple to confirm the broker granted a QoS for every topic you asked for:

use plugin mqtt::{encode_subscribe_multiple, decode_suback}

let topics = [
  #{"topic": "a/#", "qos": 1},
  #{"topic": "b/#", "qos": 2}
]
let sub = encode_subscribe_multiple(topics, 11)
// ... send `sub`, then read the SUBACK reply ...
let granted = decode_suback([0x90, 0x04, 0x00, 0x0B, 0x01, 0x02])
print("granted codes for packet {granted["packet_id"]}")

decode_packet_type(bytes) → table

Decode the first byte of any MQTT packet

Peeks at the first byte of any MQTT packet and returns {type_id, type_name, dup, qos, retain}. Use to dispatch incoming bytes to the correct decoder.

use plugin mqtt::{encode_publish, decode_packet_type}

let pkt = encode_publish("test", "hello", 0)
let info = decode_packet_type(pkt)
print(info["type_name"])

Because it also surfaces the header flags, you can read QoS and retain straight from the type byte without fully decoding the packet:

use plugin mqtt::{encode_publish_with_options, decode_packet_type}

let pkt = encode_publish_with_options("status", "up", 2, true, false, 7)
let info = decode_packet_type(pkt)
print("{info["type_name"]} qos={info["qos"]} retain={info["retain"]}")

packet_type_name(type_int) → string

Convert packet type integer to name string

Converts a packet type integer (1–14) to its name string such as "CONNECT", "PUBLISH", "SUBACK", etc.

use plugin mqtt::{packet_type_name}

print(packet_type_name(3))
print(packet_type_name(9))

remaining_length_encode(length) → any

Encode an integer as MQTT variable-length field

Encodes an integer as an MQTT variable-length field (1–4 bytes). Useful when manually constructing raw MQTT frames.

use plugin mqtt::{remaining_length_encode, remaining_length_decode}

let encoded = remaining_length_encode(300)
let decoded = remaining_length_decode(encoded)
print(decoded["length"])
print(decoded["bytes_used"])

remaining_length_decode(bytes) → table

Decode an MQTT variable-length field from bytes

Decodes an MQTT variable-length field from a byte sequence, returning {length, bytes_used}.

use plugin mqtt::{remaining_length_decode}

let result = remaining_length_decode([0xC1, 0x02])
print(result["length"])
View source code

Global index

Find your way through Zolo

Try an idea

Start here

9 results

9 results
en