Something went wrong. Try again.
An MQTT 5.0 and 3.1.1 client for Zig: a sans-I/O core, TCP, TLS, WebSocket and wss transports, SASL authentication, and reconnection
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us>// SPDX-License-Identifier: MIT
//! The small vocabulary every MQTT packet is written in: which version of the//! protocol is being spoken, what a packet is, how hard the broker has to try//! to deliver a message, and what went wrong.
const std = @import("std");
/// Which revision of the protocol a connection speaks. The number is the one/// that goes on the wire in the CONNECT packet's protocol level byte, which is/// why 3.1.1 is `4`: the level was bumped for 3.1.1 without the version number/// following it.////// A `Connection` is told which version to use once, in the CONNECT packet, and/// every packet encoded or decoded afterwards is interpreted in that light./// The two differ in more than properties -- 3.1.1 has no reason codes, no/// DISCONNECT payload, and no AUTH packet at all -- so encoding a v5 packet/// against a v3.1.1 connection is a bug the encoder catches rather than a/// message the broker will silently misread.pub const Version = enum(u8) { /// MQTT 3.1.1, OASIS Standard, 29 October 2014. Protocol level 4. v3_1_1 = 4, /// MQTT 5.0, OASIS Standard, 07 March 2019. Protocol level 5. v5_0 = 5,
/// The protocol name that goes with this level. Both versions use `MQTT`; /// only the long-dead 3.1 used `MQIsdp`, which this library does not speak. pub fn name(version: Version) []const u8 { return switch (version) { .v3_1_1, .v5_0 => "MQTT", }; }
/// Whether this version carries property blocks. Almost every difference /// between the two versions reduces to this question, so asking it directly /// reads better than comparing against `.v5_0` in twenty places. pub fn hasProperties(version: Version) bool { return version == .v5_0; }
/// Whether acknowledgements carry a reason code. In 3.1.1 a PUBACK is a /// packet identifier and nothing else, and a CONNACK's one byte is a return /// code from a different, much shorter table. pub fn hasReasonCodes(version: Version) bool { return version == .v5_0; }};
/// How hard the broker must try to deliver a message, and how much bookkeeping/// that costs both ends.pub const Qos = enum(u2) { /// Fire and forget. One packet, no acknowledgement, no packet identifier. at_most_once = 0, /// Acknowledged with PUBACK, and retransmitted until it is. The receiver /// may see a message twice, because an acknowledgement that is lost is /// indistinguishable from a message that was. at_least_once = 1, /// A four-packet handshake -- PUBLISH, PUBREC, PUBREL, PUBCOMP -- in which /// the receiver remembers the packet identifier for long enough to discard /// a retransmission. Delivered once, at the cost of two round trips and /// state on both ends. exactly_once = 2,
/// Whether a PUBLISH at this level needs a packet identifier. Only QoS 0 /// does not, and a QoS 0 PUBLISH that carries one is a malformed packet /// rather than a harmless extra. pub fn hasPacketId(qos: Qos) bool { return qos != .at_most_once; }};
/// The identifier that ties an acknowledgement to the packet it acknowledges.////// Zero is not a valid identifier, which is why `Connection` starts its/// counter at one and skips zero when it wraps.pub const PacketId = u16;
/// The four-bit packet type in the first byte of every fixed header.pub const PacketType = enum(u4) { /// Not a packet. The value is reserved by the specification and its /// appearance on the wire is a protocol error. reserved = 0, connect = 1, connack = 2, publish = 3, puback = 4, pubrec = 5, pubrel = 6, pubcomp = 7, subscribe = 8, suback = 9, unsubscribe = 10, unsuback = 11, pingreq = 12, pingresp = 13, disconnect = 14, /// Extended authentication. MQTT 5.0 only. auth = 15,
/// The low four bits of the first header byte that this packet type must /// carry, or null for PUBLISH, whose flags carry the duplicate bit, the /// QoS level and the retain bit and so vary from packet to packet. /// /// Three packet types -- PUBREL, SUBSCRIBE and UNSUBSCRIBE -- are required /// to set flags of `0b0010` rather than zero, for no reason the /// specification gives beyond the value being what the first implementation /// used. Getting it wrong is a protocol error a broker will disconnect for. pub fn requiredFlags(packet_type: PacketType) ?u4 { return switch (packet_type) { .publish => null, .pubrel, .subscribe, .unsubscribe => 0b0010, else => 0b0000, }; }
/// Whether this packet type exists in the given version. AUTH was added in /// 5.0 and is the only packet type that is not in both. pub fn existsIn(packet_type: PacketType, version: Version) bool { return switch (packet_type) { .reserved => false, .auth => version == .v5_0, else => true, }; }};
/// The flags in the low nibble of a PUBLISH packet's first header byte.pub const PublishFlags = packed struct(u4) { /// Set by the broker on a message it is delivering because the /// subscription matched an existing retained message rather than a live /// publication. retain: bool = false, qos: Qos = .at_most_once, /// Set on a retransmission of a QoS 1 or QoS 2 PUBLISH. It says only that /// this is not the first attempt to send *this packet identifier*; it says /// nothing about whether the receiver has seen the message before, and the /// receiver must not use it to decide. duplicate: bool = false,};
/// Why something happened, or did not.////// MQTT 5.0 gives every acknowledgement a reason code out of one table, with/// the convention that a value below 0x80 is a success and a value at or above/// it is a failure. The name here is the name in the specification's table,/// except where one value means different things in different packets: 0x00 is/// "Success", "Normal disconnection" and "Granted QoS 0" depending on where it/// appears, and is spelled `success` with the other two as aliases below.////// The enum is deliberately non-exhaustive. A broker may send a code from a/// later revision of the specification, and a client that panics rather than/// reporting an unrecognised failure is worse than one that says "0xA3".pub const ReasonCode = enum(u8) { success = 0x00, granted_qos_1 = 0x01, granted_qos_2 = 0x02, disconnect_with_will_message = 0x04, no_matching_subscribers = 0x10, no_subscription_existed = 0x11, continue_authentication = 0x18, reauthenticate = 0x19, unspecified_error = 0x80, malformed_packet = 0x81, protocol_error = 0x82, implementation_specific_error = 0x83, unsupported_protocol_version = 0x84, client_identifier_not_valid = 0x85, bad_user_name_or_password = 0x86, not_authorized = 0x87, server_unavailable = 0x88, server_busy = 0x89, banned = 0x8a, server_shutting_down = 0x8b, bad_authentication_method = 0x8c, keep_alive_timeout = 0x8d, session_taken_over = 0x8e, topic_filter_invalid = 0x8f, topic_name_invalid = 0x90, packet_identifier_in_use = 0x91, packet_identifier_not_found = 0x92, receive_maximum_exceeded = 0x93, topic_alias_invalid = 0x94, packet_too_large = 0x95, message_rate_too_high = 0x96, quota_exceeded = 0x97, administrative_action = 0x98, payload_format_invalid = 0x99, retain_not_supported = 0x9a, qos_not_supported = 0x9b, use_another_server = 0x9c, server_moved = 0x9d, shared_subscriptions_not_supported = 0x9e, connection_rate_exceeded = 0x9f, maximum_connect_time = 0xa0, subscription_identifiers_not_supported = 0xa1, wildcard_subscriptions_not_supported = 0xa2, _,
/// 0x00 in a DISCONNECT, where it means the sender is closing the /// connection on purpose and the will message must not be published. pub const normal_disconnection: ReasonCode = .success; /// 0x00 in a SUBACK, where it means the subscription was granted at QoS 0. pub const granted_qos_0: ReasonCode = .success;
/// Whether this code reports a failure. The specification draws the line at /// 0x80 precisely so that a client meeting a code it does not recognise can /// still tell which way it went. pub fn isError(code: ReasonCode) bool { return @backingInt(code) >= 0x80; }
/// The QoS a SUBACK granted, or null if the subscription was refused. pub fn grantedQos(code: ReasonCode) ?Qos { return switch (@backingInt(code)) { 0x00 => .at_most_once, 0x01 => .at_least_once, 0x02 => .exactly_once, else => null, }; }
/// The MQTT 5.0 code that best describes a 3.1.1 CONNACK return code. /// /// The older table has six values and no room for anything else, so this is /// a widening: every 3.1.1 refusal has a 5.0 code that means the same /// thing, and the mapping is the one the 5.0 specification gives when it /// explains what the new codes replaced. An unrecognised byte becomes /// `unspecified_error` rather than being passed through, because the two /// numbering schemes overlap and passing it through would report, say, /// "granted QoS 1" for a refused connection. pub fn fromV3ConnackReturnCode(byte: u8) ReasonCode { return switch (byte) { 0 => .success, 1 => .unsupported_protocol_version, 2 => .client_identifier_not_valid, 3 => .server_unavailable, 4 => .bad_user_name_or_password, 5 => .not_authorized, else => .unspecified_error, }; }
/// The 3.1.1 CONNACK return code for this reason code, or null if there is /// no such thing. /// /// The inverse of `fromV3ConnackReturnCode`, and it has to exist for the /// same reason that one does: decoding widens a 3.1.1 return code into the /// 5.0 table, so encoding has to narrow it again. Writing the 5.0 byte /// instead sends a number from the wrong table -- `server_unavailable` /// would go out as 0x88, which is not a 3.1.1 return code at all. /// /// Null for every 5.0 code with no older equivalent, which the caller must /// treat as unencodable rather than approximate: there is no honest way to /// tell a 3.1.1 client that it exceeded a quota. pub fn toV3ConnackReturnCode(code: ReasonCode) ?u8 { return switch (code) { .success => 0, .unsupported_protocol_version => 1, .client_identifier_not_valid => 2, .server_unavailable => 3, .bad_user_name_or_password => 4, .not_authorized => 5, else => null, }; }
/// Whether this code may appear in a 3.1.1 SUBACK, whose table is the three /// granted QoS levels and 0x80 for a refusal and nothing else. pub fn isV3SubackReturnCode(code: ReasonCode) bool { return switch (@backingInt(code)) { 0x00, 0x01, 0x02, 0x80 => true, else => false, }; }
/// The MQTT 5.0 code for a 3.1.1 SUBACK return code, whose table is the /// three granted QoS levels and 0x80 for a refusal. pub fn fromV3SubackReturnCode(byte: u8) ReasonCode { return switch (byte) { 0x00 => .success, 0x01 => .granted_qos_1, 0x02 => .granted_qos_2, else => .unspecified_error, }; }};
/// What a packet was found to be wrong about.////// The distinction the specification draws, and that this follows, is between a/// packet that cannot be parsed at all -- `MalformedPacket` -- and one that/// parses but says something the protocol forbids, such as a QoS 3 or a/// duplicate property. The two get different reason codes in the DISCONNECT/// that follows, which is the only reason they are separate errors here.pub const DecodeError = error{ /// The bytes do not form a packet: a length that overruns the buffer, a /// string whose length field lies, a variable byte integer with five /// continuation bytes. MalformedPacket, /// The packet parses, but breaks a rule: a reserved packet type, wrong /// fixed header flags, QoS 3, a property that may not appear in this packet /// or appears twice, a packet identifier of zero. ProtocolError, /// More bytes are needed before this packet can be decoded. Only the /// incremental path returns this; `Packet.decode` is given a complete /// packet body and treats a short one as malformed. Incomplete,};
/// What went wrong while encoding.pub const EncodeError = error{ /// A string longer than 65535 bytes, a payload that would push the packet /// past the protocol's 268435455-byte ceiling, or a property block too /// large to describe. PacketTooLarge, /// The packet cannot be expressed in the version being spoken: an AUTH /// packet on a 3.1.1 connection, a subscription identifier on one, a /// DISCONNECT with a reason code. UnsupportedByVersion,};
test "packet types know which flags they must carry" { try std.testing.expectEqual(@as(?u4, 0b0010), PacketType.pubrel.requiredFlags()); try std.testing.expectEqual(@as(?u4, 0b0010), PacketType.subscribe.requiredFlags()); try std.testing.expectEqual(@as(?u4, 0b0010), PacketType.unsubscribe.requiredFlags()); try std.testing.expectEqual(@as(?u4, 0b0000), PacketType.puback.requiredFlags()); try std.testing.expectEqual(@as(?u4, null), PacketType.publish.requiredFlags());}
test "publish flags pack into the low nibble in wire order" { const flags: PublishFlags = .{ .retain = true, .qos = .exactly_once, .duplicate = true }; try std.testing.expectEqual(@as(u4, 0b1101), @as(u4, @bitCast(flags)));
const round_tripped: PublishFlags = @bitCast(@as(u4, 0b0010)); try std.testing.expectEqual(Qos.at_least_once, round_tripped.qos); try std.testing.expect(!round_tripped.retain); try std.testing.expect(!round_tripped.duplicate);}
test "reason codes divide at 0x80" { try std.testing.expect(!ReasonCode.success.isError()); try std.testing.expect(!ReasonCode.granted_qos_2.isError()); try std.testing.expect(ReasonCode.unspecified_error.isError()); try std.testing.expect(ReasonCode.not_authorized.isError()); // A code from a revision this library predates is still legible as a // failure, which is the whole point of the 0x80 convention. try std.testing.expect(@as(ReasonCode, @fromBackingInt(0xf0)).isError());}
test "a suback reason code says what it granted" { try std.testing.expectEqual(Qos.at_most_once, ReasonCode.success.grantedQos().?); try std.testing.expectEqual(Qos.exactly_once, ReasonCode.granted_qos_2.grantedQos().?); try std.testing.expectEqual(@as(?Qos, null), ReasonCode.not_authorized.grantedQos());}
test "the connack return code narrows back to 3.1.1" { // Found by the fuzzer, once the packet target was actually fuzzing 3.1.1: // decoding widened 3 into `server_unavailable` and encoding wrote 0x88 // straight back out, which is not a 3.1.1 return code at all. Every code // the widening can produce must narrow again, or the round trip is a lie. const table = [_]struct { byte: u8, code: ReasonCode }{ .{ .byte = 0, .code = .success }, .{ .byte = 1, .code = .unsupported_protocol_version }, .{ .byte = 2, .code = .client_identifier_not_valid }, .{ .byte = 3, .code = .server_unavailable }, .{ .byte = 4, .code = .bad_user_name_or_password }, .{ .byte = 5, .code = .not_authorized }, }; for (table) |row| { try std.testing.expectEqual(row.code, ReasonCode.fromV3ConnackReturnCode(row.byte)); try std.testing.expectEqual(row.byte, row.code.toV3ConnackReturnCode().?); }
// And a 5.0 code with no older equivalent has none, rather than an // approximation that would mean something else. try std.testing.expectEqual(@as(?u8, null), ReasonCode.quota_exceeded.toV3ConnackReturnCode()); try std.testing.expectEqual(@as(?u8, null), ReasonCode.banned.toV3ConnackReturnCode());}
test "the 3.1.1 suback table has four entries and no more" { try std.testing.expect(ReasonCode.success.isV3SubackReturnCode()); try std.testing.expect(ReasonCode.granted_qos_1.isV3SubackReturnCode()); try std.testing.expect(ReasonCode.granted_qos_2.isV3SubackReturnCode()); try std.testing.expect(ReasonCode.unspecified_error.isV3SubackReturnCode()); // 0x80 try std.testing.expect(!ReasonCode.not_authorized.isV3SubackReturnCode()); try std.testing.expect(!ReasonCode.quota_exceeded.isV3SubackReturnCode());}
test "3.1.1 return codes widen into the 5.0 table" { try std.testing.expectEqual(ReasonCode.success, ReasonCode.fromV3ConnackReturnCode(0)); try std.testing.expectEqual(ReasonCode.bad_user_name_or_password, ReasonCode.fromV3ConnackReturnCode(4)); // 1 is "granted QoS 1" in the SUBACK table and "unacceptable protocol // version" in the CONNACK one. Passing the byte through unchanged would // conflate them, so each table gets its own mapping. try std.testing.expectEqual(ReasonCode.unsupported_protocol_version, ReasonCode.fromV3ConnackReturnCode(1)); try std.testing.expectEqual(ReasonCode.granted_qos_1, ReasonCode.fromV3SubackReturnCode(1)); try std.testing.expectEqual(ReasonCode.unspecified_error, ReasonCode.fromV3SubackReturnCode(0x80));}
test "auth exists only in 5.0" { try std.testing.expect(PacketType.auth.existsIn(.v5_0)); try std.testing.expect(!PacketType.auth.existsIn(.v3_1_1)); try std.testing.expect(PacketType.publish.existsIn(.v3_1_1)); try std.testing.expect(!PacketType.reserved.existsIn(.v5_0));}