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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us>// SPDX-License-Identifier: MIT
//! MQTT 5.0 properties: the extensible key-value block that every 5.0 packet//! carries between its variable header and its payload.//!//! A property is an identifier, written as a variable byte integer, followed by//! a value whose type the identifier determines. The block as a whole is//! preceded by its own length, so a receiver that meets an identifier it does//! not know can still find the end -- except that it cannot, because it does//! not know how long that unknown property's value is. The specification//! resolves this by making an unrecognised identifier a protocol error, which//! is what `Properties.decode` does.//!//! Which properties may appear depends on the packet, and the specification//! spells that out in a table per packet type. `Context` is that table, and//! decoding checks it: a Will Delay Interval on a PUBLISH, or a Topic Alias on//! a CONNECT, is a protocol error rather than a curiosity to be ignored.//!//! Nothing here allocates. A decoded `Properties` borrows every string from the//! buffer it was decoded out of, and the two properties that may appear more//! than once -- user properties and subscription identifiers -- are read back//! through iterators that re-walk the original block rather than through a//! slice that would have to be built somewhere.
const std = @import("std");const codec = @import("codec.zig");const types = @import("types.zig");
const Decoder = codec.Decoder;const Encoder = codec.Encoder;const StringPair = codec.StringPair;const DecodeError = types.DecodeError;const Qos = types.Qos;
/// A property identifier. The values are those of MQTT 5.0 section 2.2.2.2,/// Table 2-4, and the gaps in the numbering are the specification's own.////// Exhaustive on purpose, unlike `ReasonCode`: an identifier this library does/// not know is not a value to be carried around, because its value cannot be/// skipped over, and the packet it is in cannot be parsed at all.pub const Id = enum(u8) { payload_format_indicator = 0x01, message_expiry_interval = 0x02, content_type = 0x03, response_topic = 0x08, correlation_data = 0x09, subscription_identifier = 0x0b, session_expiry_interval = 0x11, assigned_client_identifier = 0x12, server_keep_alive = 0x13, authentication_method = 0x15, authentication_data = 0x16, request_problem_information = 0x17, will_delay_interval = 0x18, request_response_information = 0x19, response_information = 0x1a, server_reference = 0x1c, reason_string = 0x1f, receive_maximum = 0x21, topic_alias_maximum = 0x22, topic_alias = 0x23, maximum_qos = 0x24, retain_available = 0x25, user_property = 0x26, maximum_packet_size = 0x27, wildcard_subscription_available = 0x28, subscription_identifier_available = 0x29, shared_subscription_available = 0x2a,
/// The shape of this property's value, which is what tells a decoder how /// many bytes to read. pub fn valueKind(id: Id) ValueKind { return switch (id) { .payload_format_indicator, .request_problem_information, .request_response_information, .maximum_qos, .retain_available, .wildcard_subscription_available, .subscription_identifier_available, .shared_subscription_available, => .byte,
.server_keep_alive, .receive_maximum, .topic_alias_maximum, .topic_alias, => .int16,
.message_expiry_interval, .session_expiry_interval, .will_delay_interval, .maximum_packet_size, => .int32,
.subscription_identifier => .varint,
.content_type, .response_topic, .assigned_client_identifier, .authentication_method, .response_information, .server_reference, .reason_string, => .string,
.correlation_data, .authentication_data => .binary,
.user_property => .string_pair, }; }
/// Whether this property may appear more than once in one block. Only two /// may: a user property, which is the whole point of it, and a subscription /// identifier on a PUBLISH, where one appears per subscription the message /// matched. pub fn isRepeatable(id: Id) bool { return switch (id) { .user_property, .subscription_identifier => true, else => false, }; }};
/// The shape of a property's value.pub const ValueKind = enum { byte, int16, int32, varint, string, binary, string_pair };
/// Where a property block appears, which decides which properties are allowed/// in it.////// `will` is not a packet type. The Will properties live inside a CONNECT/// packet's payload and have a table of their own, sharing only user properties/// with the CONNECT properties that precede them.pub const Context = enum { connect, will, connack, publish, puback, pubrec, pubrel, pubcomp, subscribe, suback, unsubscribe, unsuback, disconnect, auth,
/// Whether `id` may appear in a block in this position. Straight out of the /// per-packet tables in MQTT 5.0 section 3. pub fn allows(context: Context, id: Id) bool { return switch (context) { .connect => switch (id) { .session_expiry_interval, .receive_maximum, .maximum_packet_size, .topic_alias_maximum, .request_response_information, .request_problem_information, .user_property, .authentication_method, .authentication_data, => true, else => false, }, .will => switch (id) { .will_delay_interval, .payload_format_indicator, .message_expiry_interval, .content_type, .response_topic, .correlation_data, .user_property, => true, else => false, }, .connack => switch (id) { .session_expiry_interval, .receive_maximum, .maximum_qos, .retain_available, .maximum_packet_size, .assigned_client_identifier, .topic_alias_maximum, .reason_string, .user_property, .wildcard_subscription_available, .subscription_identifier_available, .shared_subscription_available, .server_keep_alive, .response_information, .server_reference, .authentication_method, .authentication_data, => true, else => false, }, .publish => switch (id) { .payload_format_indicator, .message_expiry_interval, .topic_alias, .response_topic, .correlation_data, .user_property, .subscription_identifier, .content_type, => true, else => false, }, // The four PUBLISH acknowledgements share one table. .puback, .pubrec, .pubrel, .pubcomp, .suback, .unsuback => switch (id) { .reason_string, .user_property => true, else => false, }, .subscribe => switch (id) { .subscription_identifier, .user_property => true, else => false, }, .unsubscribe => id == .user_property, .disconnect => switch (id) { .session_expiry_interval, .reason_string, .user_property, .server_reference, => true, else => false, }, .auth => switch (id) { .authentication_method, .authentication_data, .reason_string, .user_property, => true, else => false, }, }; }};
/// Whether a PUBLISH payload is declared to be text or unspecified bytes. It is/// a declaration and not a guarantee: the specification lets a broker reject a/// payload that says UTF-8 and is not, but does not require it to look.pub const PayloadFormat = enum(u8) { unspecified = 0, utf8 = 1,};
/// One property block.////// The same type is used to encode and to decode, which means the two/// many-valued properties have two sources: a caller building a packet fills in/// `user_properties` and `subscription_identifiers` with slices it owns, while/// a decoded block leaves those empty and sets `raw` to the bytes it was parsed/// from. `userProperties()` and `subscriptionIdentifiers()` read from whichever/// is populated, so a packet that is decoded and then re-encoded keeps its user/// properties without anything having had to copy them.pub const Properties = struct { payload_format_indicator: ?PayloadFormat = null, message_expiry_interval: ?u32 = null, content_type: ?[]const u8 = null, response_topic: ?[]const u8 = null, correlation_data: ?[]const u8 = null, session_expiry_interval: ?u32 = null, assigned_client_identifier: ?[]const u8 = null, server_keep_alive: ?u16 = null, authentication_method: ?[]const u8 = null, authentication_data: ?[]const u8 = null, request_problem_information: ?bool = null, will_delay_interval: ?u32 = null, request_response_information: ?bool = null, response_information: ?[]const u8 = null, server_reference: ?[]const u8 = null, reason_string: ?[]const u8 = null, receive_maximum: ?u16 = null, topic_alias_maximum: ?u16 = null, topic_alias: ?u16 = null, maximum_qos: ?Qos = null, retain_available: ?bool = null, maximum_packet_size: ?u32 = null, wildcard_subscription_available: ?bool = null, subscription_identifier_available: ?bool = null, shared_subscription_available: ?bool = null,
/// Set when encoding. Read back through `userProperties()`. user_properties: []const StringPair = &.{}, /// Set when encoding. Read back through `subscriptionIdentifiers()`. subscription_identifiers: []const u32 = &.{},
/// The bytes this block was decoded from, excluding the length prefix. /// Set by `decode` and left empty by a caller building a packet. raw: []const u8 = &.{},
/// An empty block, which is what a 5.0 packet carries when it has nothing /// to say: a single zero length byte rather than nothing at all. pub const empty: Properties = .{};
/// Read the user properties, in the order they appeared. /// /// Order is preserved because the specification requires it: user /// properties are the one place where the same name may appear twice and /// mean two different things, so a receiver that sorted or deduplicated /// them would be changing the message. pub fn userProperties(p: *const Properties) UserPropertyIterator { return if (p.raw.len != 0) .{ .source = .{ .raw = .init(p.raw) } } else .{ .source = .{ .slice = p.user_properties } }; }
/// Read the subscription identifiers, in the order they appeared. A PUBLISH /// delivered because it matched three of this client's subscriptions /// carries three of them. pub fn subscriptionIdentifiers(p: *const Properties) SubscriptionIdIterator { return if (p.raw.len != 0) .{ .source = .{ .raw = .init(p.raw) } } else .{ .source = .{ .slice = p.subscription_identifiers } }; }
/// The first subscription identifier, which is all a client that never /// overlaps its subscriptions needs to ask for. pub fn subscriptionIdentifier(p: *const Properties) ?u32 { var it = p.subscriptionIdentifiers(); return it.next(); }
/// Whether this block would encode to nothing but its zero length byte. pub fn isEmpty(p: *const Properties) bool { return p.encodedBodyLen() == 0; }
/// Read a property block, including its leading length, out of `d`. /// /// `context` says which packet the block belongs to and therefore which /// properties are allowed. A property that is not allowed there, one that /// repeats when it may not, an unrecognised identifier, or a value outside /// the range the specification gives it are all protocol errors; a block /// whose length overruns the packet is malformed. pub fn decode(d: *Decoder, context: Context) DecodeError!Properties { const len = try d.varint(); const block = try d.take(len);
var p: Properties = .{ .raw = block }; var seen: std.EnumSet(Id) = .initEmpty(); var inner: Decoder = .init(block);
while (!inner.isEmpty()) { const raw_id = try inner.varint(); if (raw_id > std.math.maxInt(u8)) return error.ProtocolError; const id = std.enums.fromInt(Id, @as(u8, @intCast(raw_id))) orelse return error.ProtocolError;
if (!context.allows(id)) return error.ProtocolError; if (!id.isRepeatable() and seen.contains(id)) return error.ProtocolError; seen.insert(id);
switch (id) { // -- byte-valued ------------------------------------------- .payload_format_indicator => p.payload_format_indicator = std.enums.fromInt(PayloadFormat, try inner.byte()) orelse return error.ProtocolError, .request_problem_information => p.request_problem_information = try decodeBool(&inner), .request_response_information => p.request_response_information = try decodeBool(&inner), .retain_available => p.retain_available = try decodeBool(&inner), .wildcard_subscription_available => p.wildcard_subscription_available = try decodeBool(&inner), .subscription_identifier_available => p.subscription_identifier_available = try decodeBool(&inner), .shared_subscription_available => p.shared_subscription_available = try decodeBool(&inner), .maximum_qos => p.maximum_qos = switch (try inner.byte()) { 0 => .at_most_once, 1 => .at_least_once, // Section 3.2.2.3.4: the value is 0 or 1. A broker that // supports QoS 2 says so by leaving the property out, so // a literal 2 here is a protocol error and not a broker // being helpfully explicit. else => return error.ProtocolError, },
// -- two-byte ---------------------------------------------- .server_keep_alive => p.server_keep_alive = try inner.int16(), .topic_alias_maximum => p.topic_alias_maximum = try inner.int16(), .receive_maximum => { const v = try inner.int16(); // Section 3.1.2.11.3: zero is a protocol error rather than // "no messages in flight, ever". if (v == 0) return error.ProtocolError; p.receive_maximum = v; }, .topic_alias => { const v = try inner.int16(); // Section 3.3.2.3.4: alias 0 is reserved. if (v == 0) return error.ProtocolError; p.topic_alias = v; },
// -- four-byte --------------------------------------------- .message_expiry_interval => p.message_expiry_interval = try inner.int32(), .session_expiry_interval => p.session_expiry_interval = try inner.int32(), .will_delay_interval => p.will_delay_interval = try inner.int32(), .maximum_packet_size => { const v = try inner.int32(); // Section 3.1.2.11.4: zero is a protocol error. if (v == 0) return error.ProtocolError; p.maximum_packet_size = v; },
// -- variable byte ----------------------------------------- .subscription_identifier => { const v = try inner.varint(); // Section 3.3.2.3.8: zero is a protocol error. if (v == 0) return error.ProtocolError; },
// -- strings and bytes ------------------------------------- .content_type => p.content_type = try inner.string(), .response_topic => p.response_topic = try inner.string(), .assigned_client_identifier => p.assigned_client_identifier = try inner.string(), .authentication_method => p.authentication_method = try inner.string(), .response_information => p.response_information = try inner.string(), .server_reference => p.server_reference = try inner.string(), .reason_string => p.reason_string = try inner.string(), .correlation_data => p.correlation_data = try inner.binary(), .authentication_data => p.authentication_data = try inner.binary(),
// Read and discarded: it is reached through the iterator, out // of `raw`, rather than being collected into a slice that would // need somewhere to live. Reading it here is still necessary -- // it is how the cursor gets past it, and how it is validated. .user_property => _ = try inner.stringPair(), } }
return p; }
/// How many bytes the block's contents take, not counting the length prefix /// in front of them. pub fn encodedBodyLen(p: *const Properties) usize { var len: usize = 0;
if (p.payload_format_indicator != null) len += 2; if (p.request_problem_information != null) len += 2; if (p.request_response_information != null) len += 2; if (p.retain_available != null) len += 2; if (p.wildcard_subscription_available != null) len += 2; if (p.subscription_identifier_available != null) len += 2; if (p.shared_subscription_available != null) len += 2; if (p.maximum_qos != null) len += 2;
if (p.server_keep_alive != null) len += 3; if (p.receive_maximum != null) len += 3; if (p.topic_alias_maximum != null) len += 3; if (p.topic_alias != null) len += 3;
if (p.message_expiry_interval != null) len += 5; if (p.session_expiry_interval != null) len += 5; if (p.will_delay_interval != null) len += 5; if (p.maximum_packet_size != null) len += 5;
if (p.content_type) |v| len += 1 + codec.binaryLen(v); if (p.response_topic) |v| len += 1 + codec.binaryLen(v); if (p.assigned_client_identifier) |v| len += 1 + codec.binaryLen(v); if (p.authentication_method) |v| len += 1 + codec.binaryLen(v); if (p.response_information) |v| len += 1 + codec.binaryLen(v); if (p.server_reference) |v| len += 1 + codec.binaryLen(v); if (p.reason_string) |v| len += 1 + codec.binaryLen(v); if (p.correlation_data) |v| len += 1 + codec.binaryLen(v); if (p.authentication_data) |v| len += 1 + codec.binaryLen(v);
var subs = p.subscriptionIdentifiers(); while (subs.next()) |id| len += 1 + codec.varintLen(id);
var users = p.userProperties(); while (users.next()) |pair| len += 1 + codec.stringPairLen(pair);
return len; }
/// How many bytes the block takes on the wire, length prefix included. pub fn encodedLen(p: *const Properties) usize { const body = p.encodedBodyLen(); return codec.varintLen(@intCast(body)) + body; }
/// Write the block, length prefix first. /// /// Properties are written in identifier order, which the specification /// explicitly does not require -- section 2.2.2.1 says the order is not /// significant -- but which makes the output of this library reproducible, /// and so makes a byte-for-byte test of an encoder possible at all. pub fn encode(p: *const Properties, e: Encoder) Encoder.Error!void { const body = p.encodedBodyLen(); if (body > codec.max_variable_byte_integer) return error.PacketTooLarge; try e.varint(@intCast(body));
if (p.payload_format_indicator) |v| { try e.byte(@intFromEnum(Id.payload_format_indicator)); try e.byte(@intFromEnum(v)); } if (p.message_expiry_interval) |v| { try e.byte(@intFromEnum(Id.message_expiry_interval)); try e.int32(v); } if (p.content_type) |v| { try e.byte(@intFromEnum(Id.content_type)); try e.string(v); } if (p.response_topic) |v| { try e.byte(@intFromEnum(Id.response_topic)); try e.string(v); } if (p.correlation_data) |v| { try e.byte(@intFromEnum(Id.correlation_data)); try e.binary(v); }
var subs = p.subscriptionIdentifiers(); while (subs.next()) |v| { try e.byte(@intFromEnum(Id.subscription_identifier)); try e.varint(v); }
if (p.session_expiry_interval) |v| { try e.byte(@intFromEnum(Id.session_expiry_interval)); try e.int32(v); } if (p.assigned_client_identifier) |v| { try e.byte(@intFromEnum(Id.assigned_client_identifier)); try e.string(v); } if (p.server_keep_alive) |v| { try e.byte(@intFromEnum(Id.server_keep_alive)); try e.int16(v); } if (p.authentication_method) |v| { try e.byte(@intFromEnum(Id.authentication_method)); try e.string(v); } if (p.authentication_data) |v| { try e.byte(@intFromEnum(Id.authentication_data)); try e.binary(v); } if (p.request_problem_information) |v| { try e.byte(@intFromEnum(Id.request_problem_information)); try e.byte(@intFromBool(v)); } if (p.will_delay_interval) |v| { try e.byte(@intFromEnum(Id.will_delay_interval)); try e.int32(v); } if (p.request_response_information) |v| { try e.byte(@intFromEnum(Id.request_response_information)); try e.byte(@intFromBool(v)); } if (p.response_information) |v| { try e.byte(@intFromEnum(Id.response_information)); try e.string(v); } if (p.server_reference) |v| { try e.byte(@intFromEnum(Id.server_reference)); try e.string(v); } if (p.reason_string) |v| { try e.byte(@intFromEnum(Id.reason_string)); try e.string(v); } if (p.receive_maximum) |v| { try e.byte(@intFromEnum(Id.receive_maximum)); try e.int16(v); } if (p.topic_alias_maximum) |v| { try e.byte(@intFromEnum(Id.topic_alias_maximum)); try e.int16(v); } if (p.topic_alias) |v| { try e.byte(@intFromEnum(Id.topic_alias)); try e.int16(v); } if (p.maximum_qos) |v| { try e.byte(@intFromEnum(Id.maximum_qos)); try e.byte(@intFromEnum(v)); } if (p.retain_available) |v| { try e.byte(@intFromEnum(Id.retain_available)); try e.byte(@intFromBool(v)); }
var users = p.userProperties(); while (users.next()) |pair| { try e.byte(@intFromEnum(Id.user_property)); try e.stringPair(pair); }
if (p.maximum_packet_size) |v| { try e.byte(@intFromEnum(Id.maximum_packet_size)); try e.int32(v); } if (p.wildcard_subscription_available) |v| { try e.byte(@intFromEnum(Id.wildcard_subscription_available)); try e.byte(@intFromBool(v)); } if (p.subscription_identifier_available) |v| { try e.byte(@intFromEnum(Id.subscription_identifier_available)); try e.byte(@intFromBool(v)); } if (p.shared_subscription_available) |v| { try e.byte(@intFromEnum(Id.shared_subscription_available)); try e.byte(@intFromBool(v)); } }};
/// A byte-valued property whose only legal values are 0 and 1. Section 3.1.2.10/// and its neighbours make anything else a protocol error rather than treating/// non-zero as true.fn decodeBool(d: *Decoder) DecodeError!bool { return switch (try d.byte()) { 0 => false, 1 => true, else => error.ProtocolError, };}
/// Step the cursor past one property's value, given its identifier.////// Used only by the iterators, which walk a block that `Properties.decode` has/// already validated. That is why it can be private and why the iterators can/// treat a failure as the end of the block: by the time either runs, there is/// nothing left to fail on.fn skipValue(d: *Decoder, id: Id) DecodeError!void { switch (id.valueKind()) { .byte => _ = try d.byte(), .int16 => _ = try d.int16(), .int32 => _ = try d.int32(), .varint => _ = try d.varint(), .string, .binary => _ = try d.binary(), .string_pair => _ = try d.stringPair(), }}
/// Walks the user properties of a block, from either of the two places they/// can live.pub const UserPropertyIterator = struct { source: union(enum) { raw: Decoder, slice: []const StringPair, }, index: usize = 0,
pub fn next(it: *UserPropertyIterator) ?StringPair { switch (it.source) { .slice => |s| { if (it.index >= s.len) return null; defer it.index += 1; return s[it.index]; }, .raw => |*d| { while (!d.isEmpty()) { const raw_id = d.varint() catch return null; const id = std.enums.fromInt(Id, std.math.cast(u8, raw_id) orelse return null) orelse return null; if (id == .user_property) return d.stringPair() catch null; skipValue(d, id) catch return null; } return null; }, } }};
/// Walks the subscription identifiers of a block, from either of the two places/// they can live.pub const SubscriptionIdIterator = struct { source: union(enum) { raw: Decoder, slice: []const u32, }, index: usize = 0,
pub fn next(it: *SubscriptionIdIterator) ?u32 { switch (it.source) { .slice => |s| { if (it.index >= s.len) return null; defer it.index += 1; return s[it.index]; }, .raw => |*d| { while (!d.isEmpty()) { const raw_id = d.varint() catch return null; const id = std.enums.fromInt(Id, std.math.cast(u8, raw_id) orelse return null) orelse return null; if (id == .subscription_identifier) return d.varint() catch null; skipValue(d, id) catch return null; } return null; }, } }};
const testing = std.testing;
/// Encode a block into `buf` and hand back the bytes, so that a test can say/// what it means in one line.fn encodeToBuf(p: Properties, buf: []u8) ![]const u8 { var w: std.Io.Writer = .fixed(buf); try p.encode(.init(&w)); return w.buffered();}
test "an empty block is one zero byte" { var buf: [8]u8 = undefined; const bytes = try encodeToBuf(.empty, &buf); try testing.expectEqualSlices(u8, &.{0x00}, bytes); try testing.expectEqual(@as(usize, 1), Properties.empty.encodedLen());
var d: Decoder = .init(bytes); const decoded = try Properties.decode(&d, .publish); try testing.expect(decoded.isEmpty()); try testing.expect(d.isEmpty());}
test "measuring a block agrees with writing it" { // Every property kind at once, so that a wrong size for any one of them // shows up as a disagreement rather than as a packet that happens to work. const p: Properties = .{ .payload_format_indicator = .utf8, .message_expiry_interval = 60, .content_type = "application/json", .response_topic = "reply/to/me", .correlation_data = &.{ 0xde, 0xad, 0xbe, 0xef }, .topic_alias = 7, .subscription_identifiers = &.{ 1, 300 }, .user_properties = &.{ .{ .name = "a", .value = "1" }, .{ .name = "b", .value = "2" }, }, }; var buf: [256]u8 = undefined; const bytes = try encodeToBuf(p, &buf); try testing.expectEqual(p.encodedLen(), bytes.len);}
test "a full publish block round trips" { const p: Properties = .{ .payload_format_indicator = .utf8, .message_expiry_interval = 60, .content_type = "application/json", .response_topic = "reply/to/me", .correlation_data = &.{ 0xde, 0xad, 0xbe, 0xef }, .topic_alias = 7, .subscription_identifiers = &.{ 1, 300 }, .user_properties = &.{ .{ .name = "a", .value = "1" }, .{ .name = "b", .value = "2" }, }, }; var buf: [256]u8 = undefined; const bytes = try encodeToBuf(p, &buf);
var d: Decoder = .init(bytes); const got = try Properties.decode(&d, .publish); try testing.expect(d.isEmpty());
try testing.expectEqual(PayloadFormat.utf8, got.payload_format_indicator.?); try testing.expectEqual(@as(u32, 60), got.message_expiry_interval.?); try testing.expectEqualStrings("application/json", got.content_type.?); try testing.expectEqualStrings("reply/to/me", got.response_topic.?); try testing.expectEqualSlices(u8, &.{ 0xde, 0xad, 0xbe, 0xef }, got.correlation_data.?); try testing.expectEqual(@as(u16, 7), got.topic_alias.?);
var subs = got.subscriptionIdentifiers(); try testing.expectEqual(@as(u32, 1), subs.next().?); try testing.expectEqual(@as(u32, 300), subs.next().?); try testing.expectEqual(@as(?u32, null), subs.next()); try testing.expectEqual(@as(u32, 1), got.subscriptionIdentifier().?);
var users = got.userProperties(); try testing.expect(users.next().?.eql(.{ .name = "a", .value = "1" })); try testing.expect(users.next().?.eql(.{ .name = "b", .value = "2" })); try testing.expectEqual(@as(?StringPair, null), users.next());}
test "a decoded block re-encodes with its user properties intact" { // This is what the two-source iterator buys: nothing had to allocate a // slice of pairs in order to forward a message on. const original: Properties = .{ .content_type = "text/plain", .user_properties = &.{ .{ .name = "trace", .value = "abc" }, .{ .name = "trace", .value = "def" }, }, }; var buf: [128]u8 = undefined; const bytes = try encodeToBuf(original, &buf);
var d: Decoder = .init(bytes); const decoded = try Properties.decode(&d, .publish);
var buf2: [128]u8 = undefined; const again = try encodeToBuf(decoded, &buf2); try testing.expectEqualSlices(u8, bytes, again);}
test "duplicate user properties keep their order and are not merged" { const p: Properties = .{ .user_properties = &.{ .{ .name = "k", .value = "first" }, .{ .name = "k", .value = "second" }, .{ .name = "k", .value = "first" }, } }; var buf: [128]u8 = undefined; var d: Decoder = .init(try encodeToBuf(p, &buf)); const got = try Properties.decode(&d, .publish);
var it = got.userProperties(); try testing.expectEqualStrings("first", it.next().?.value); try testing.expectEqualStrings("second", it.next().?.value); try testing.expectEqualStrings("first", it.next().?.value); try testing.expectEqual(@as(?StringPair, null), it.next());}
test "a property that repeats where it may not is a protocol error" { // Two content types. var d: Decoder = .init(&.{ 0x08, 0x03, 0x00, 0x01, 'a', 0x03, 0x00, 0x01, 'b' }); try testing.expectError(error.ProtocolError, Properties.decode(&d, .publish));}
test "a property in the wrong packet is a protocol error" { // Will Delay Interval on a PUBLISH. It is a perfectly well-formed // property; it just has no meaning there. const block = [_]u8{ 0x05, @intFromEnum(Id.will_delay_interval), 0, 0, 0, 30 }; var as_publish: Decoder = .init(&block); try testing.expectError(error.ProtocolError, Properties.decode(&as_publish, .publish));
var as_will: Decoder = .init(&block); const ok = try Properties.decode(&as_will, .will); try testing.expectEqual(@as(u32, 30), ok.will_delay_interval.?);}
test "an unrecognised property identifier is a protocol error" { // There is no 0x07, and no way to know how many bytes its value would // have taken, so the packet cannot be parsed at all. var d: Decoder = .init(&.{ 0x02, 0x07, 0x00 }); try testing.expectError(error.ProtocolError, Properties.decode(&d, .publish));}
test "a block longer than the bytes it is in is malformed" { var d: Decoder = .init(&.{ 0x20, 0x01, 0x01 }); try testing.expectError(error.MalformedPacket, Properties.decode(&d, .publish));}
test "the values the specification restricts are checked" { const cases = [_]struct { context: Context, block: []const u8 }{ // Receive Maximum of zero. .{ .context = .connack, .block = &.{ 0x03, @intFromEnum(Id.receive_maximum), 0, 0 } }, // Maximum Packet Size of zero. .{ .context = .connack, .block = &.{ 0x05, @intFromEnum(Id.maximum_packet_size), 0, 0, 0, 0 } }, // Topic Alias of zero. .{ .context = .publish, .block = &.{ 0x03, @intFromEnum(Id.topic_alias), 0, 0 } }, // Subscription Identifier of zero. .{ .context = .publish, .block = &.{ 0x02, @intFromEnum(Id.subscription_identifier), 0 } }, // Maximum QoS of 2, which is spelled by omitting the property. .{ .context = .connack, .block = &.{ 0x02, @intFromEnum(Id.maximum_qos), 2 } }, // Retain Available of 2, where only 0 and 1 are defined. .{ .context = .connack, .block = &.{ 0x02, @intFromEnum(Id.retain_available), 2 } }, // Payload Format Indicator of 2. .{ .context = .publish, .block = &.{ 0x02, @intFromEnum(Id.payload_format_indicator), 2 } }, }; for (cases, 0..) |case, i| { var d: Decoder = .init(case.block); testing.expectError(error.ProtocolError, Properties.decode(&d, case.context)) catch |err| { std.debug.print("case {d} did not fail as expected\n", .{i}); return err; }; }}
test "the connack properties a broker actually sends" { // What mosquitto answers a 5.0 CONNECT with, in the order it sends them, // which is not identifier order -- so this also checks that decoding does // not care about order even though encoding fixes one. const block = [_]u8{ 35, // the block's length, which the packet decoder will have read @intFromEnum(Id.topic_alias_maximum), 0x00, 0x0a, @intFromEnum(Id.receive_maximum), 0x00, 0x14, @intFromEnum(Id.maximum_qos), 0x01, @intFromEnum(Id.retain_available), 0x01, @intFromEnum(Id.maximum_packet_size), 0x00, 0x01, 0x00, 0x00, @intFromEnum(Id.wildcard_subscription_available), 0x01, @intFromEnum(Id.subscription_identifier_available), 0x01, @intFromEnum(Id.shared_subscription_available), 0x01, @intFromEnum(Id.server_keep_alive), 0x00, 0x3c, @intFromEnum(Id.assigned_client_identifier), 0x00, 0x08, 'a', 'u', 't', 'o', '-', '1', '2', '3', }; var d: Decoder = .init(&block); const p = try Properties.decode(&d, .connack); try testing.expect(d.isEmpty());
try testing.expectEqual(@as(u16, 10), p.topic_alias_maximum.?); try testing.expectEqual(@as(u16, 20), p.receive_maximum.?); try testing.expectEqual(Qos.at_least_once, p.maximum_qos.?); try testing.expect(p.retain_available.?); try testing.expectEqual(@as(u32, 65536), p.maximum_packet_size.?); try testing.expect(p.wildcard_subscription_available.?); try testing.expect(p.subscription_identifier_available.?); try testing.expect(p.shared_subscription_available.?); try testing.expectEqual(@as(u16, 60), p.server_keep_alive.?); try testing.expectEqualStrings("auto-123", p.assigned_client_identifier.?);}
test "every property identifier has a value kind and a table entry" { // A new property added to `Id` without being added to a `Context.allows` // arm would be decodable nowhere, which is a silent failure rather than a // loud one -- so the table is checked for coverage rather than trusted. for (std.enums.values(Id)) |id| { _ = id.valueKind(); var allowed_somewhere = false; for (std.enums.values(Context)) |context| { if (context.allows(id)) allowed_somewhere = true; } try testing.expect(allowed_somewhere); }}