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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us>// SPDX-License-Identifier: MIT
//! Topic names, topic filters, and whether one matches the other.//!//! A topic name is what a message is published to. A topic filter is what a//! subscription asks for, and may use two wildcards: `+` stands for exactly one//! level and `#` for the rest of them. Both wildcards must occupy a whole//! level -- `sport+` is not a filter, it is an invalid one -- and `#` must be//! the last thing in the filter.//!//! The matching rules have three corners that a naive implementation gets//! wrong, and each has a test below://!//! * `#` matches the parent level as well as the children, so `sport/#`//! matches `sport` and not only `sport/tennis`.//! * A leading `$` is special. A wildcard at the first level does not match a//! topic that begins with one, so `#` does not deliver the broker's own//! `$SYS` statistics to a client that asked for everything.//! * An empty level is a level. `/finance` has two of them, matches `+/finance`//! and does not match `finance`.//!//! Nothing here allocates: matching walks both strings in place.
const std = @import("std");const codec = @import("codec.zig");
/// The prefix that makes a filter a shared subscription: `$share/{name}/{filter}`.pub const shared_prefix = "$share/";
/// Why a topic name or filter is not one.pub const TopicError = error{ /// Zero length. Both a name and a filter must have at least one character, /// which means the empty string is not the same as `#`. Empty, /// Longer than 65535 bytes, which is as long as the length prefix in front /// of it can describe. TooLong, /// A wildcard in a topic name, where only a filter may have one. WildcardInTopicName, /// A `+` or `#` sharing a level with anything else, or a `#` that is not /// the last character. MalformedWildcard, /// Not well-formed UTF-8, or containing U+0000. InvalidEncoding, /// A `$share/` prefix with no share name, or one containing `/`, `+` or `#`. InvalidShareName, /// A topic name beginning with `$share/`. That prefix makes a filter a /// shared subscription, so no subscription could ever match such a name: /// a message published there would be delivered to nobody at all. SharedSubscriptionAsTopicName,};
/// Check that `name` may be published to.////// A topic name is a filter's opposite: it must be concrete, so a wildcard in/// it is an error rather than a pattern. A broker will reject one either way,/// so checking before sending turns a dropped connection into a returned error.////// `$share/...` is rejected as well, and that is a judgement rather than a/// quoted rule. Section 4.8.2 defines `$share/{name}/{filter}` as the syntax of/// a shared subscription *filter* and says nothing about topic names; what/// makes it worth refusing is that a message published there could never reach/// anybody. Every filter that could match it is itself a shared subscription,/// whose prefix is stripped before matching, so the message goes nowhere --/// and section 4.7.2 already says a server SHOULD stop clients exchanging/// messages on dollar-prefixed topics at all.////// `$SYS` and other dollar-prefixed names are left alone, because a broker/// publishes to those and nothing should stop a tool alongside one doing the/// same. Only the prefix that is certain to be undeliverable is refused.pub fn validateName(name: []const u8) TopicError!void { if (name.len == 0) return error.Empty; if (name.len > codec.max_binary_len) return error.TooLong; try validateEncoding(name); if (std.mem.findScalar(u8, name, '+') != null) return error.WildcardInTopicName; if (std.mem.findScalar(u8, name, '#') != null) return error.WildcardInTopicName; if (std.mem.startsWith(u8, name, shared_prefix)) return error.SharedSubscriptionAsTopicName;}
/// Check that `filter` may be subscribed to.////// A `$share/{name}/` prefix is accepted and its share name checked, because a/// shared subscription is a filter with a prefix rather than a different kind/// of thing.pub fn validateFilter(filter: []const u8) TopicError!void { if (filter.len == 0) return error.Empty; if (filter.len > codec.max_binary_len) return error.TooLong; try validateEncoding(filter);
const rest = if (try parseShared(filter)) |shared| shared.filter else filter; if (rest.len == 0) return error.Empty;
var levels = std.mem.splitScalar(u8, rest, '/'); var seen_multi = false; while (levels.next()) |level| { // `#` is the rest of the topic, so anything after it could never match. if (seen_multi) return error.MalformedWildcard; if (std.mem.eql(u8, level, "#")) { seen_multi = true; } else if (std.mem.eql(u8, level, "+")) { // Fine anywhere, any number of times. } else if (std.mem.findScalar(u8, level, '+') != null or std.mem.findScalar(u8, level, '#') != null) { // A wildcard sharing a level with other characters. `sport+` and // `sport#` are both errors rather than literal topic levels. return error.MalformedWildcard; } }}
/// A shared subscription, split into the group that shares it and the filter/// the group is subscribed to.pub const Shared = struct { /// The group name. Every subscriber using the same name and filter shares /// one subscription, and each message goes to one of them rather than all. share_name: []const u8, /// The filter itself, with the prefix removed. filter: []const u8,};
/// Split a `$share/{name}/{filter}` filter, or return null if it is not one.////// Returns an error rather than null when the filter starts with `$share/` and/// is then wrong, because that is a filter a broker will refuse and not a plain/// topic that happens to begin with a dollar sign.pub fn parseShared(filter: []const u8) TopicError!?Shared { if (!std.mem.startsWith(u8, filter, shared_prefix)) return null; const rest = filter[shared_prefix.len..]; const slash = std.mem.findScalar(u8, rest, '/') orelse return error.InvalidShareName; const share_name = rest[0..slash]; if (share_name.len == 0) return error.InvalidShareName; // Section 4.8.2: the share name may not contain a wildcard or a separator. for (share_name) |c| { if (c == '+' or c == '#') return error.InvalidShareName; } return .{ .share_name = share_name, .filter = rest[slash + 1 ..] };}
/// Whether `filter` matches `topic`.////// Both are assumed to be valid; this answers the matching question and not the/// validity one, so a malformed filter gets whatever answer falling through the/// rules produces rather than an error. Validate first if the filter came from/// somewhere that could have got it wrong.////// A shared subscription matches whatever the filter inside it matches, so the/// `$share/{name}/` prefix is stripped before the comparison. That is what a/// client dispatching a received message wants: the message arrives on the/// concrete topic, and the subscription that asked for it is the shared one.pub fn matches(filter: []const u8, topic: []const u8) bool { const pattern = if (parseShared(filter) catch null) |shared| shared.filter else filter;
// Section 4.7.2: a filter beginning with a wildcard does not match a topic // beginning with `$`, so the broker's own `$SYS` tree is not delivered to // somebody who subscribed to `#`. if (topic.len != 0 and topic[0] == '$') { if (pattern.len != 0 and (pattern[0] == '#' or pattern[0] == '+')) return false; }
var filter_levels = std.mem.splitScalar(u8, pattern, '/'); var topic_levels = std.mem.splitScalar(u8, topic, '/');
while (filter_levels.next()) |filter_level| { if (std.mem.eql(u8, filter_level, "#")) { // `#` matches the rest, including nothing at all: `sport/#` matches // `sport` as well as `sport/tennis/player1`. return true; } const topic_level = topic_levels.next() orelse return false; if (std.mem.eql(u8, filter_level, "+")) continue; if (!std.mem.eql(u8, filter_level, topic_level)) return false; }
// The filter is exhausted, so the topic must be too. An empty last level is // still a level, which is why this asks the splitter rather than comparing // lengths. return topic_levels.next() == null;}
fn validateEncoding(text: []const u8) TopicError!void { if (!std.unicode.utf8ValidateSlice(text)) return error.InvalidEncoding; if (std.mem.findScalar(u8, text, 0) != null) return error.InvalidEncoding;}
const testing = std.testing;
test "the specification's own matching examples" { // MQTT 5.0 section 4.7.1, which is where each of these comes from. const yes = [_]struct { filter: []const u8, topic: []const u8 }{ .{ .filter = "sport/tennis/player1/#", .topic = "sport/tennis/player1" }, .{ .filter = "sport/tennis/player1/#", .topic = "sport/tennis/player1/ranking" }, .{ .filter = "sport/tennis/player1/#", .topic = "sport/tennis/player1/score/wimbledon" }, .{ .filter = "sport/#", .topic = "sport" }, .{ .filter = "#", .topic = "anything/at/all" }, .{ .filter = "sport/tennis/+", .topic = "sport/tennis/player1" }, .{ .filter = "sport/tennis/+", .topic = "sport/tennis/player2" }, .{ .filter = "sport/+", .topic = "sport/" }, .{ .filter = "+/+", .topic = "/finance" }, .{ .filter = "/+", .topic = "/finance" }, .{ .filter = "+", .topic = "sport" }, }; for (yes) |case| { testing.expect(matches(case.filter, case.topic)) catch { std.debug.print("expected '{s}' to match '{s}'\n", .{ case.filter, case.topic }); return error.TestExpectedMatch; }; }
const no = [_]struct { filter: []const u8, topic: []const u8 }{ .{ .filter = "sport/tennis/+", .topic = "sport/tennis/player1/ranking" }, .{ .filter = "sport/+", .topic = "sport" }, .{ .filter = "+/finance", .topic = "finance" }, .{ .filter = "sport/tennis", .topic = "sport/tennis/player1" }, .{ .filter = "sport/tennis/player1", .topic = "sport/tennis" }, }; for (no) |case| { testing.expect(!matches(case.filter, case.topic)) catch { std.debug.print("expected '{s}' not to match '{s}'\n", .{ case.filter, case.topic }); return error.TestExpectedNoMatch; }; }}
test "a multi-level wildcard matches the parent level too" { // The corner that a split-and-compare implementation gets wrong: the filter // has more levels than the topic and still matches. try testing.expect(matches("sport/#", "sport")); try testing.expect(matches("sport/tennis/#", "sport/tennis")); try testing.expect(!matches("sport/tennis/#", "sport"));}
test "an empty level is a level" { try testing.expect(matches("/finance", "/finance")); try testing.expect(matches("+/finance", "/finance")); try testing.expect(!matches("finance", "/finance")); try testing.expect(matches("a//b", "a//b")); try testing.expect(matches("a/+/b", "a//b")); try testing.expect(!matches("a/b", "a/b/")); try testing.expect(matches("a/b/+", "a/b/"));}
test "a leading dollar sign is not matched by a leading wildcard" { // Section 4.7.2, and the reason a client that subscribes to `#` does not // find itself reading the broker's own statistics. try testing.expect(!matches("#", "$SYS/broker/uptime")); try testing.expect(!matches("+/broker/uptime", "$SYS/broker/uptime")); try testing.expect(matches("$SYS/#", "$SYS/broker/uptime")); try testing.expect(matches("$SYS/broker/+", "$SYS/broker/uptime")); // Only the *first* level is protected. try testing.expect(matches("a/#", "a/$b")); try testing.expect(matches("a/+", "a/$b"));}
test "a shared subscription matches what the filter inside it matches" { try testing.expect(matches("$share/workers/jobs/+", "jobs/17")); try testing.expect(!matches("$share/workers/jobs/+", "other/17"));
const shared = (try parseShared("$share/workers/jobs/+")).?; try testing.expectEqualStrings("workers", shared.share_name); try testing.expectEqualStrings("jobs/+", shared.filter);
try testing.expectEqual(@as(?Shared, null), try parseShared("jobs/+")); // A topic that merely begins with a dollar sign is not a shared one. try testing.expectEqual(@as(?Shared, null), try parseShared("$SYS/#"));}
test "a share name must be there and must be plain" { try testing.expectError(error.InvalidShareName, parseShared("$share/jobs")); try testing.expectError(error.InvalidShareName, parseShared("$share//jobs/+")); try testing.expectError(error.InvalidShareName, parseShared("$share/work+ers/jobs")); try testing.expectError(error.InvalidShareName, parseShared("$share/work#/jobs"));}
test "a topic name may not contain a wildcard" { try validateName("sport/tennis/player1"); try validateName("/"); try validateName("$SYS/broker/uptime"); try testing.expectError(error.WildcardInTopicName, validateName("sport/#")); try testing.expectError(error.WildcardInTopicName, validateName("sport/+/player1")); try testing.expectError(error.Empty, validateName("")); try testing.expectError(error.InvalidEncoding, validateName("a\x00b"));}
test "a topic name may not be a shared subscription filter" { // Found by the fuzzer, which noticed that a name accepted by `validateName` // did not match itself when used as a filter. It could not: `matches` // strips the `$share/{name}/` prefix from the filter side, so no filter can // ever match a topic name that carries one, and a message published there // would go nowhere at all. Refusing it is this library's choice rather than // a rule out of the specification -- see the note on `validateName`. try testing.expectError( error.SharedSubscriptionAsTopicName, validateName("$share/group/a"), ); try testing.expect(!matches("$share/group/a", "$share/group/a"));
// As a filter it is perfectly good, and it matches what is inside it. try validateFilter("$share/group/a"); try testing.expect(matches("$share/group/a", "a"));}
test "a wildcard must occupy a whole level" { try validateFilter("sport/#"); try validateFilter("#"); try validateFilter("+"); try validateFilter("+/tennis/#"); try validateFilter("sport/+/player1"); try validateFilter("/");
try testing.expectError(error.MalformedWildcard, validateFilter("sport+")); try testing.expectError(error.MalformedWildcard, validateFilter("sport/tennis#")); try testing.expectError(error.MalformedWildcard, validateFilter("sport/#/ranking")); try testing.expectError(error.MalformedWildcard, validateFilter("#/sport")); try testing.expectError(error.Empty, validateFilter(""));}
test "a filter longer than its own length prefix can describe is refused" { const too_long = try testing.allocator.alloc(u8, codec.max_binary_len + 1); defer testing.allocator.free(too_long); @memset(too_long, 'a'); try testing.expectError(error.TooLong, validateFilter(too_long)); try testing.expectError(error.TooLong, validateName(too_long));}