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.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us>// SPDX-License-Identifier: MIT
//! An MQTT client session with no I/O in it.//!//! Everything this library knows about being a client lives here: the packet//! identifier counter, the QoS 1 and QoS 2 handshakes in both directions, the//! flow control window the broker imposes, topic aliases, keepalive, and what//! has to be sent again when a session is resumed. None of it touches a socket,//! a file, a clock or a thread.//!//! The caller moves the bytes and supplies the time://!//! ```zig//! var conn: Connection = .init(gpa, .{});//! defer conn.deinit();//!//! try conn.connect(.{ .client_id = "sensor-7" }, now());//!//! while (true) {//! // Anything the connection wants to say.//! const out = conn.pending();//! if (out.len != 0) conn.sent(try stream.write(out), now());//!//! // Anything the broker said.//! const n = try stream.read(&buf);//! try conn.receive(buf[0..n], now());//! while (try conn.nextEvent()) |event| switch (event) {//! .message => |m| handle(m),//! else => {},//! }//!//! // Keepalive, retransmission, flow control.//! try conn.tick(now());//! }//! ```//!//! **Events borrow.** A `.message` event's topic and payload point into the//! receive buffer, and stay valid only until the next call to `nextEvent` or//! `receive`. An application that keeps a message must copy it. This is what//! makes delivering a message free of allocation, and it is the one sharp edge//! in the interface.//!//! Time is whatever monotonic millisecond count the caller has. The connection//! never asks what time it is; it is told, at the three points where it could//! matter -- when bytes went out, when bytes came in, and when the caller wants//! timers serviced.
const std = @import("std");const Allocator = std.mem.Allocator;
const protocol = @import("protocol.zig");const packet = protocol.packet;const topic_rules = protocol.topic;const session_format = @import("session.zig");
const Packet = packet.Packet;const PacketId = protocol.types.PacketId;const Properties = protocol.properties.Properties;const Qos = protocol.types.Qos;const ReasonCode = protocol.types.ReasonCode;const Version = protocol.types.Version;
const Connection = @This();
/// How the connection is set up, decided once and never renegotiated.pub const Options = struct { /// Which version to speak. A broker that does not support 5.0 refuses the /// CONNECT with `unsupported_protocol_version`, at which point a client /// that wants to fall back builds a new `Connection` with `.v3_1_1`. version: Version = .v5_0,
/// The largest packet this client will accept. A broker that sends a larger /// one has broken the limit it was told about in the CONNECT, and the /// connection fails rather than growing the receive buffer to fit. /// /// In 3.1.1 there is no way to tell the broker, so this is enforced locally /// and a broker that exceeds it is simply disconnected from. max_incoming_packet_size: u32 = 1 << 20,
/// How many QoS 1 and QoS 2 messages the broker may have in flight towards /// this client at once. Sent as the Receive Maximum property; 3.1.1 has no /// equivalent and brokers speaking it are unbounded. receive_maximum: u16 = 20,
/// How many topic aliases this client will accept from the broker. Zero /// means none, and a broker that sends one anyway is a protocol error. incoming_topic_alias_maximum: u16 = 8,
/// How many topic aliases this client is willing to establish for its own /// publications, capped by what the broker says it will accept. Zero -- the /// default -- means topics are always sent in full, which is what a client /// that publishes to a handful of topics wants; a client publishing to one /// topic thousands of times a second wants this turned on. outgoing_topic_alias_maximum: u16 = 0,
/// How long the broker should keep this session after the connection ends, /// in seconds, sent as the Session Expiry Interval property. /// /// Null sends nothing, and MQTT 5.0 reads a missing Session Expiry /// Interval as zero -- section 3.1.2.11.2: the session ends when the /// network connection does. That is the specification's default and so it /// is this one, but it is almost never what a client asking for /// `clean_start = false` meant: without it the broker throws the session /// away the moment the socket closes, every reconnection reports /// `session_present` false, and the unacknowledged messages this end is /// holding have nowhere to go. Measured against mosquitto, which does /// exactly that. /// /// So anything that persists or resumes a session wants a value here, and /// wants it longer than the outage it expects to survive. /// `0xFFFF_FFFF` means the session does not expire at all. /// /// 3.1.1 has no equivalent: `clean_session = 0` asks the broker to keep the /// session and the broker decides for how long, so this is ignored there. session_expiry_interval: ?u32 = null,
/// How long after a PINGREQ the broker may go without answering before the /// connection is declared dead, as a multiple of the keep alive interval. /// The specification says nothing about this -- it describes only the /// broker's duty to disconnect a silent client -- so the number is a /// judgement rather than a rule. ping_response_factor: f32 = 1.0,};
/// Where the session has got to.pub const State = enum { /// Nothing sent yet. `connect` is the only thing to do. idle, /// CONNECT has been queued or sent, and the CONNACK has not arrived. connecting, /// The broker accepted the connection. connected, /// DISCONNECT has been queued. The caller should flush and close. disconnecting, /// Over, either because the broker refused or closed, or because something /// went wrong. Session state survives for a reconnection. closed,};
/// What went wrong.pub const Error = error{ /// The bytes from the broker do not form a packet. MalformedPacket, /// The broker said something the protocol forbids -- including, for /// example, acknowledging a packet identifier that was never sent. ProtocolError, /// An incoming packet is larger than `Options.max_incoming_packet_size`, or /// an outgoing one larger than the broker said it would accept. PacketTooLarge, /// The packet cannot be expressed in the version being spoken. UnsupportedByVersion, /// A command that requires an established connection was issued before /// CONNACK, or after the connection ended. NotConnected, /// `connect` on a connection that is already connecting or connected. AlreadyConnected, /// The broker's CONNACK gave a Maximum QoS below the one asked for. QosNotSupported, /// The broker's CONNACK said it does not retain messages. RetainNotSupported, /// Every one of the 65535 packet identifiers is in use. Only reachable by /// a client with a very large Receive Maximum and a very slow broker. PacketIdentifiersExhausted, /// The broker did not answer a PINGREQ in time. KeepAliveTimeout, /// The broker used a topic alias above the maximum it was given, or one it /// had not established. TopicAliasInvalid, /// `auth` on a connection whose CONNECT named no authentication method. /// There is no exchange to continue. NotAuthenticating, /// `auth` with a reason code that is not one an endpoint may send: /// section 3.15.2.1 allows `continue_authentication` and `reauthenticate` /// from a client, and `success` only from the broker. BadReasonCode, /// `auth` naming a different authentication method from the CONNECT, /// which section 4.12 fixes for the life of the exchange. AuthenticationMethodChanged, OutOfMemory,} || topic_rules.TopicError;
/// Something the broker did.////// The publish and acknowledgement handshakes do not appear here: by the time a/// `.message` is handed out its PUBACK or PUBREC has already been queued, and a/// `.published` means the whole handshake finished. What is left is what an/// application would want to know about.pub const Event = union(enum) { /// The broker accepted the connection. Carries the CONNACK, whose /// properties say what limits the broker is imposing and, if the client /// sent an empty identifier, what it has been named. connected: packet.ConnAck, /// A message. Its topic and payload borrow from the receive buffer and are /// valid only until the next call to `nextEvent` or `receive`. message: packet.Publish, /// A QoS 1 or QoS 2 publication this client sent has finished its /// handshake, successfully or otherwise. published: Published, /// The answer to a `subscribe`, with one reason code per filter. subscribed: packet.SubAck, /// The answer to an `unsubscribe`. unsubscribed: packet.SubAck, /// The broker answered a PINGREQ. pong, /// A step in an extended authentication exchange. The caller is expected to /// answer with `auth`. MQTT 5.0 only. authenticate: packet.Auth, /// The broker refused the connection or closed it, and said why. The /// connection is `closed` and the transport should be too. disconnected: Disconnected,};
/// A publication of this client's that has finished.pub const Published = struct { packet_id: PacketId, qos: Qos, /// What the broker made of it. `no_matching_subscribers` is a success that /// is worth knowing about; anything at or above 0x80 is a failure, and the /// message was not delivered. reason_code: ReasonCode = .success, properties: Properties = .empty,
pub fn succeeded(p: Published) bool { return !p.reason_code.isError(); }};
/// The end of the connection, as the broker described it.pub const Disconnected = struct { reason_code: ReasonCode, properties: Properties = .empty, /// Whether this came from a DISCONNECT packet or from a CONNACK that /// refused the connection. The distinction matters to a client deciding /// whether to retry: a refused CONNECT will usually be refused again. origin: Origin,
pub const Origin = enum { /// The broker refused the connection outright. connack, /// The broker ended an established connection. disconnect, };};
/// What to say in the CONNECT packet.////// Nothing here is retained: the strings are copied into the outgoing buffer/// before `connect` returns, so they may be stack-allocated. A reconnection/// calls `connect` again and supplies them again.pub const ConnectOptions = struct { client_id: []const u8 = "", clean_start: bool = true, /// Seconds. Zero disables keepalive entirely, which means a connection to a /// broker that has silently gone away is never noticed. keep_alive: u16 = 60, username: ?[]const u8 = null, password: ?[]const u8 = null, will: ?packet.Will = null, /// Extra properties. Receive Maximum, Maximum Packet Size and Topic Alias /// Maximum are filled in from `Options` if they are not set here, so that a /// caller that does not care gets limits that match what this connection /// will actually enforce. properties: Properties = .empty,};
/// A message to publish.pub const Message = struct { topic: []const u8, payload: []const u8 = "", qos: Qos = .at_most_once, retain: bool = false, properties: Properties = .empty,};
gpa: Allocator,options: Options,version: Version,state: State = .idle,
/// Bytes received from the transport and not yet decoded, plus a cursor. The/// cursor is what lets a decoded packet keep borrowing from the buffer after/// it has been handed to the caller: nothing is discarded until the caller asks/// for the next event.rx: std.ArrayList(u8) = .empty,rx_start: usize = 0,/// The length of the packet most recently returned as an event, which is what/// `rx_start` advances by on the next call.rx_held: usize = 0,
/// Bytes to hand to the transport, plus a cursor over what has gone already.tx: std.ArrayList(u8) = .empty,tx_start: usize = 0,
next_packet_id: PacketId = 1,
/// Changes whenever the session state does.////// For a caller persisting the session: keep the value from the last save, and/// there is nothing to write while it still matches. It is generous in one/// direction only -- it can move when a save would produce the same bytes, and/// it never fails to move when the bytes would differ -- which is the direction/// that matters, because the cost of the first is a redundant write and the/// cost of the second is a lost message.////// Wraps rather than saturating, because a `u64` of changes is not reachable/// and comparing for equality does not care.session_serial: u64 = 0,
/// Publications waiting for or undergoing an acknowledgement handshake, oldest/// first. Bounded by the broker's Receive Maximum plus whatever the caller has/// queued beyond it, and searched linearly, which is the right shape for the/// tens of entries this holds in practice.outbound: std.ArrayList(Outbound) = .empty,
/// Packet identifiers of QoS 2 messages received and not yet released, which is/// how a retransmitted PUBLISH is recognised and not delivered twice.inbound_qos2: std.AutoHashMapUnmanaged(PacketId, void) = .empty,
/// Topic aliases the broker has established, and the topics they stand for./// The topics are owned, because the PUBLISH that introduced the alias is long/// out of the receive buffer by the time the alias is used.incoming_aliases: std.AutoHashMapUnmanaged(u16, []u8) = .empty,
/// Topic aliases this client has established. Keys are owned.outgoing_aliases: std.StringHashMapUnmanaged(u16) = .empty,next_outgoing_alias: u16 = 1,
// -- what the broker said in its CONNACK -------------------------------------
/// How many QoS 1 and 2 publications may be in flight towards the broker./// 65535 until told otherwise, which is the specification's default.server_receive_maximum: u16 = 65535,server_maximum_packet_size: ?u32 = null,server_maximum_qos: Qos = .exactly_once,server_retain_available: bool = true,server_topic_alias_maximum: u16 = 0,
/// The keep alive actually in force: what was asked for, unless the broker's/// CONNACK overrode it with a Server Keep Alive.keep_alive: u16 = 0,last_sent_ms: i64 = 0,last_received_ms: i64 = 0,ping_sent_ms: i64 = 0,ping_outstanding: bool = false,
/// The authentication method named in the CONNECT, owned, or null when this/// connection asked for no extended exchange.////// Owned rather than borrowed because it outlives the call that supplied it and/// is compared against every AUTH for the life of the connection -- section/// 4.12 fixes the method for the whole exchange, and a caller that rebuilt its/// mechanism between attempts would otherwise leave this pointing at a name/// that had been freed.auth_method: ?[]u8 = null,
/// One publication in flight.const Outbound = struct { packet_id: PacketId, qos: Qos, stage: Stage, /// The encoded PUBLISH, kept so that it can be sent again with the /// duplicate flag set if the session is resumed. Freed as soon as it can no /// longer be needed -- on PUBACK for QoS 1, and on PUBREC for QoS 2, since /// after that the handshake is carried by PUBREL and PUBCOMP alone. bytes: []u8, /// Set when a resumed session owes this exchange a PUBREL again, which is /// the case for a QoS 2 message whose PUBCOMP never arrived. It is a flag /// rather than an immediate `enqueue` so that the resend goes out in list /// order along with the publications being sent again. resend_pubrel: bool = false,
const Stage = enum { /// Held back by flow control; not yet on the wire. queued, awaiting_puback, awaiting_pubrec, awaiting_pubcomp, };
fn inFlight(o: Outbound) bool { return o.stage != .queued; }};
pub fn init(gpa: Allocator, options: Options) Connection { return .{ .gpa = gpa, .options = options, .version = options.version };}
pub fn deinit(c: *Connection) void { c.clearSession(); if (c.auth_method) |m| c.gpa.free(m); c.rx.deinit(c.gpa); c.tx.deinit(c.gpa); c.outbound.deinit(c.gpa); c.inbound_qos2.deinit(c.gpa); c.incoming_aliases.deinit(c.gpa); c.outgoing_aliases.deinit(c.gpa); c.* = undefined;}
// -- moving bytes ------------------------------------------------------------
/// Hand over bytes read from the transport.////// This invalidates any event still being held from a previous `nextEvent`,/// because it may move the buffer those borrowed slices point into.pub fn receive(c: *Connection, bytes: []const u8, now_ms: i64) Error!void { c.last_received_ms = now_ms; try c.rx.appendSlice(c.gpa, bytes);}
/// The bytes waiting to be written to the transport. Empty when there is/// nothing to say.pub fn pending(c: *const Connection) []const u8 { return c.tx.items[c.tx_start..];}
/// Report that the first `n` bytes of `pending` reached the transport.////// This is where the keepalive clock is reset, because what matters to the/// broker is when bytes arrived, not when they were encoded.pub fn sent(c: *Connection, n: usize, now_ms: i64) void { std.debug.assert(n <= c.pending().len); if (n == 0) return; c.tx_start += n; c.last_sent_ms = now_ms; if (c.tx_start == c.tx.items.len) { c.tx.clearRetainingCapacity(); c.tx_start = 0; } else if (c.tx_start > c.tx.items.len / 2) { // Compaction is amortised: only once the cursor has passed the halfway // point, so a connection dribbling bytes out does not memmove the whole // buffer on every write. c.tx.replaceRangeAssumeCapacity(0, c.tx_start, &.{}); c.tx_start = 0; }}
/// Whether there is anything to write.pub fn wantsWrite(c: *const Connection) bool { return c.pending().len != 0;}
// -- events ------------------------------------------------------------------
/// Decode the next thing the broker said, or null if nothing complete has/// arrived.////// The returned event borrows from the receive buffer and is invalidated by the/// next call to this function or to `receive`.////// Packets that need no decision from the caller -- PUBREL, and the PUBACK this/// client owes for an incoming QoS 1 message -- are handled here and produce no/// event; the loop keeps going until there is something to report or the buffer/// runs out.pub fn nextEvent(c: *Connection) Error!?Event { while (true) { c.releaseHeld(); const bytes = c.rx.items[c.rx_start..];
const header = packet.peekFixedHeader(bytes) catch |err| switch (err) { error.Incomplete => return null, else => |e| return c.fail(e), }; if (header.remaining_length > c.options.max_incoming_packet_size) { return c.fail(error.PacketTooLarge); } if (bytes.len < header.packetLen()) return null;
const body = bytes[header.encodedLen()..header.packetLen()]; const p = Packet.decode(header, body, c.version) catch |err| switch (err) { // `decode` is handed a complete body, so it cannot want more. error.Incomplete => unreachable, else => |e| return c.fail(e), }; c.rx_held = header.packetLen();
if (try c.handle(p)) |event| return event; }}
/// Discard the packet handed out by the previous call, now that the caller has/// had its turn with it.fn releaseHeld(c: *Connection) void { c.rx_start += c.rx_held; c.rx_held = 0; if (c.rx_start == c.rx.items.len) { c.rx.clearRetainingCapacity(); c.rx_start = 0; } else if (c.rx_start > c.rx.items.len / 2) { c.rx.replaceRangeAssumeCapacity(0, c.rx_start, &.{}); c.rx_start = 0; }}
/// End the connection because of `err`, so that a caller that ignores the error/// and calls back in does not carry on as though nothing happened.fn fail(c: *Connection, err: Error) Error { c.state = .closed; return err;}
fn handle(c: *Connection, p: Packet) Error!?Event { // Section 3.2: the first packet from the broker must be a CONNACK // [MQTT-3.2.0-1]. An AUTH may come first instead, and only then, because // section 4.12 makes the extended authentication exchange part of the // handshake rather than something that follows it. // // This is not pedantry. Without it a broker could deliver a message before // the connection was established, into a session whose limits are not yet // known -- and `Client.connect`, which has nothing sensible to do with a // message, would have to handle one. if (c.state == .connecting and p != .connack and p != .auth) { return c.fail(error.ProtocolError); }
return switch (p) { .connack => |ack| try c.handleConnack(ack), .publish => |m| try c.handlePublish(m), .puback => |a| try c.handlePuback(a), .pubrec => |a| try c.handlePubrec(a), .pubrel => |a| try c.handlePubrel(a), .pubcomp => |a| try c.handlePubcomp(a), .suback => |a| .{ .subscribed = a }, .unsuback => |a| .{ .unsubscribed = a }, .pingresp => blk: { c.ping_outstanding = false; break :blk .pong; }, .disconnect => |d| blk: { c.state = .closed; break :blk .{ .disconnected = .{ .reason_code = d.reason_code, .properties = d.properties, .origin = .disconnect, } }; }, .auth => |a| try c.handleAuth(a), // Everything else is a packet only a broker receives. A broker sending // one has misunderstood which end it is. .connect, .subscribe, .unsubscribe, .pingreq, .reserved => c.fail(error.ProtocolError), };}
// -- commands ----------------------------------------------------------------
/// Queue a CONNECT.////// The properties this connection needs the broker to know about -- how big a/// packet it will accept, how many messages it will take at once, how many/// topic aliases it will remember -- are filled in from `Options` unless the/// caller set them, so that what is advertised is what is actually enforced.pub fn connect(c: *Connection, options: ConnectOptions, now_ms: i64) Error!void { if (c.state == .connecting or c.state == .connected) return error.AlreadyConnected;
if (options.will) |will| try topic_rules.validateName(will.topic);
// Section 4.12: the method is fixed when the CONNECT goes out and every // AUTH after it must name the same one. Kept here so that this can be // checked rather than assumed. if (c.auth_method) |m| c.gpa.free(m); c.auth_method = null; if (c.version == .v5_0) { if (options.properties.authentication_method) |m| { c.auth_method = try c.gpa.dupe(u8, m); } } errdefer if (c.auth_method) |m| { c.gpa.free(m); c.auth_method = null; };
c.state = .connecting; c.keep_alive = options.keep_alive; c.last_sent_ms = now_ms; c.last_received_ms = now_ms; c.ping_outstanding = false;
// A clean start throws away whatever the last connection left behind, // before the CONNECT that asks for it goes out. if (options.clean_start) c.clearSession();
var properties = options.properties; if (c.version == .v5_0) { if (properties.receive_maximum == null and c.options.receive_maximum != 65535) { properties.receive_maximum = c.options.receive_maximum; } if (properties.maximum_packet_size == null) { properties.maximum_packet_size = c.options.max_incoming_packet_size; } if (properties.topic_alias_maximum == null and c.options.incoming_topic_alias_maximum != 0) { properties.topic_alias_maximum = c.options.incoming_topic_alias_maximum; } if (properties.session_expiry_interval == null) { if (c.options.session_expiry_interval) |seconds| { properties.session_expiry_interval = seconds; } } // Whatever ends up in the packet is what will be enforced when an alias // arrives. A caller that set the property by hand would otherwise have // advertised one limit while the connection went on rejecting anything // above another, which is the same failure as advertising a Receive // Maximum that is not the one in force. c.options.incoming_topic_alias_maximum = properties.topic_alias_maximum orelse 0; }
try c.enqueue(.{ .connect = .{ .client_id = options.client_id, .clean_start = options.clean_start, .keep_alive = options.keep_alive, .username = options.username, .password = options.password, .will = options.will, .properties = properties, } });}
/// Publish a message.////// Returns the packet identifier for QoS 1 and QoS 2, which is what a later/// `.published` event will be tagged with, and null for QoS 0, which is never/// acknowledged and so has nothing to wait for.////// A QoS 1 or 2 message beyond the broker's Receive Maximum is queued rather/// than refused, and goes out as earlier ones are acknowledged. There is no/// bound on that queue: a caller publishing faster than the broker acknowledges/// is the one holding the information about what should give way.pub fn publish(c: *Connection, message: Message) Error!?PacketId { if (c.state != .connected) return error.NotConnected; try topic_rules.validateName(message.topic); if (@backingInt(message.qos) > @backingInt(c.server_maximum_qos)) return error.QosNotSupported; if (message.retain and !c.server_retain_available) return error.RetainNotSupported;
var p: packet.Publish = .{ .topic = message.topic, .payload = message.payload, .qos = message.qos, .retain = message.retain, .properties = message.properties, }; // Aliasing rewrites the packet, so it has to happen before the length is // measured against the broker's maximum packet size. if (c.version == .v5_0) { if (p.properties.topic_alias) |alias| { // The caller chose the alias. Sending one above what the broker // said it would accept is a protocol error it will disconnect over, // so it is refused here where it is still a returned error. if (alias == 0 or alias > c.server_topic_alias_maximum) return error.TopicAliasInvalid; } else { try c.applyOutgoingAlias(&p); } }
if (message.qos == .at_most_once) { try c.enqueue(.{ .publish = p }); return null; }
const id = try c.takePacketId(); p.packet_id = id;
const bytes = try c.encodeAlloc(.{ .publish = p }); { // Scoped, because once the append succeeds the list owns `bytes`. An // `errdefer` covering the whole function would free them again if // `pumpQueued` then ran out of memory, leaving an entry in the list // pointing at memory that had been handed back. errdefer c.gpa.free(bytes); c.touchSession(); try c.outbound.append(c.gpa, .{ .packet_id = id, .qos = message.qos, .stage = .queued, .bytes = bytes, }); } try c.pumpQueued(); return id;}
/// Subscribe to one or more filters. Returns the packet identifier the SUBACK/// will carry.pub fn subscribe( c: *Connection, filters: []const packet.TopicFilter, properties: Properties,) Error!PacketId { if (c.state != .connected) return error.NotConnected; if (filters.len == 0) return error.UnsupportedByVersion; for (filters) |f| try topic_rules.validateFilter(f.filter);
const id = try c.takePacketId(); try c.enqueue(.{ .subscribe = .{ .packet_id = id, .filters = filters, .properties = properties, } }); return id;}
/// Unsubscribe from one or more filters. Returns the packet identifier the/// UNSUBACK will carry.pub fn unsubscribe( c: *Connection, filters: []const []const u8, properties: Properties,) Error!PacketId { if (c.state != .connected) return error.NotConnected; if (filters.len == 0) return error.UnsupportedByVersion; for (filters) |f| try topic_rules.validateFilter(f);
const id = try c.takePacketId(); try c.enqueue(.{ .unsubscribe = .{ .packet_id = id, .filters = filters, .properties = properties, } }); return id;}
/// Queue a DISCONNECT and stop accepting commands.////// The caller should flush `pending` before closing the transport: a/// DISCONNECT that never arrives leaves the broker to publish the will message,/// which is the opposite of what a clean shutdown means.pub fn disconnect(c: *Connection, reason_code: ReasonCode, properties: Properties) Error!void { if (c.state != .connected and c.state != .connecting) return error.NotConnected; try c.enqueue(.{ .disconnect = .{ .reason_code = reason_code, .properties = properties } }); c.state = .disconnecting;}
/// Queue an AUTH, continuing an extended authentication exchange. MQTT 5.0 only.////// The authentication method is filled in from what the CONNECT named, because/// section 4.12 requires every AUTH to carry it and there is exactly one right/// answer. A caller that sets it to something else has changed the method/// mid-exchange, which is what the rule exists to forbid.////// `reason_code` is `continue_authentication` to answer a challenge, or/// `reauthenticate` to start the exchange again on a connection that is/// already up (section 4.12.1).pub fn auth(c: *Connection, reason_code: ReasonCode, properties: Properties) Error!void { if (c.state != .connected and c.state != .connecting) return error.NotConnected; const method = c.auth_method orelse return error.NotAuthenticating; switch (reason_code) { .continue_authentication => {}, // Section 4.12.1: only meaningful once the connection is up. .reauthenticate => if (c.state != .connected) return error.NotConnected, else => return error.BadReasonCode, } var p = properties; if (p.authentication_method) |named| { if (!std.mem.eql(u8, named, method)) return error.AuthenticationMethodChanged; } else { p.authentication_method = method; } try c.enqueue(.{ .auth = .{ .reason_code = reason_code, .properties = p } });}
/// Queue a PINGREQ regardless of whether one is due. `tick` sends these on its/// own; this is for a caller that wants to probe the connection now.pub fn ping(c: *Connection, now_ms: i64) Error!void { if (c.state != .connected) return error.NotConnected; try c.enqueue(.pingreq); if (!c.ping_outstanding) { c.ping_outstanding = true; c.ping_sent_ms = now_ms; }}
// -- timers ------------------------------------------------------------------
/// Service the clock: send a PINGREQ if one is due, and give up if the broker/// has stopped answering them.////// Calling this more often than necessary is harmless. Calling it not at all/// means the broker will eventually disconnect this client for being silent.pub fn tick(c: *Connection, now_ms: i64) Error!void { if (c.state != .connected) return; if (c.keep_alive == 0) return;
const interval_ms: i64 = @as(i64, c.keep_alive) * 1000;
if (c.ping_outstanding) { const grace: i64 = @intFromFloat(@as(f32, @floatFromInt(interval_ms)) * c.options.ping_response_factor); if (now_ms - c.ping_sent_ms > grace) return c.fail(error.KeepAliveTimeout); return; }
if (now_ms - c.last_sent_ms >= interval_ms) try c.ping(now_ms);}
/// When `tick` next wants to be called, as an absolute time on the caller's/// clock, or null if there is nothing to wait for.////// A caller with an event loop uses this as the timeout on its read; a caller/// happy to block on the transport can ignore it, since the PINGREQ that falls/// due will be answered and so will wake the read anyway.pub fn deadline(c: *const Connection) ?i64 { if (c.state != .connected or c.keep_alive == 0) return null; const interval_ms: i64 = @as(i64, c.keep_alive) * 1000; if (c.ping_outstanding) { const grace: i64 = @intFromFloat(@as(f32, @floatFromInt(interval_ms)) * c.options.ping_response_factor); return c.ping_sent_ms + grace; } return c.last_sent_ms + interval_ms;}
/// The transport went away without a DISCONNECT.////// Throws away the half-read packet and anything queued to write, and keeps the/// session: the unacknowledged publications, the packet identifier counter, and/// the record of which QoS 2 messages have been received. Reconnecting with/// `clean_start = false` then resumes where this left off, and the broker/// confirming the session in its CONNACK is what causes the unacknowledged/// publications to be sent again.pub fn transportClosed(c: *Connection) void { c.state = .closed; c.rx.clearRetainingCapacity(); c.rx_start = 0; c.rx_held = 0; c.tx.clearRetainingCapacity(); c.tx_start = 0; c.ping_outstanding = false;
// Aliases are a property of one network connection, not of the session, and // both ends forget them when it ends. c.clearIncomingAliases(); c.clearOutgoingAliases();
// Anything on the wire is back to being merely queued, so that resuming the // session sends it again in order. That is session state, so a caller // persisting it has something new to write. c.touchSession(); for (c.outbound.items) |*o| { if (o.stage == .awaiting_puback) o.stage = .queued; if (o.stage == .awaiting_pubrec) o.stage = .queued; }}
/// How many publications are waiting for an acknowledgement or for a window to/// go out in. Mostly useful to a caller deciding whether to publish more.pub fn inFlightCount(c: *const Connection) usize { return c.outbound.items.len;}
/// Whether bytes of an unfinished packet are being held.////// The question a transport layer asks when the stream ends: between packets is/// a conversation that finished, and part-way through one is a conversation/// that was cut off.pub fn hasPartialPacket(c: *const Connection) bool { return c.rx.items.len != c.rx_start;}
// -- incoming packets --------------------------------------------------------
fn handleConnack(c: *Connection, ack: packet.ConnAck) Error!?Event { if (c.state != .connecting) return c.fail(error.ProtocolError);
// Section 4.12: a CONNACK that accepts a connection whose CONNECT named an // authentication method must name the same method. Refusing a success that // does not is not pedantry -- it is the whole of what stops the exchange // being skipped. A broker that simply ignored the method and said yes // would otherwise leave a client believing it had authenticated by a // mechanism that never ran. if (!ack.reason_code.isError()) { if (c.auth_method) |method| { const named = ack.properties.authentication_method orelse return c.fail(error.ProtocolError); if (!std.mem.eql(u8, named, method)) return c.fail(error.ProtocolError); } }
if (ack.reason_code.isError()) { c.state = .closed; return .{ .disconnected = .{ .reason_code = ack.reason_code, .properties = ack.properties, .origin = .connack, } }; }
const p = ack.properties; if (p.receive_maximum) |v| c.server_receive_maximum = v; if (p.maximum_packet_size) |v| c.server_maximum_packet_size = v; if (p.maximum_qos) |v| c.server_maximum_qos = v; if (p.retain_available) |v| c.server_retain_available = v; if (p.topic_alias_maximum) |v| c.server_topic_alias_maximum = v; // Section 3.2.2.3.14: a Server Keep Alive replaces what the client asked // for, and the client must use it. if (p.server_keep_alive) |v| c.keep_alive = v;
c.state = .connected;
if (!ack.session_present) { // A fresh session: whatever was left over belongs to a session the // broker has forgotten, and sending it would be sending it into // nothing. c.clearSession(); } else { // Section 4.4: everything unacknowledged goes again, in the order it // was first sent, with the duplicate flag set. Both kinds of debt are // only marked here; `pumpQueued` is what puts them on the wire, so that // a PUBLISH and a PUBREL owed by the same session go out in the order // the messages were first published rather than in two separate passes. c.touchSession(); for (c.outbound.items) |*o| { switch (o.stage) { .awaiting_puback, .awaiting_pubrec, .queued => o.stage = .queued, // The PUBLISH is long gone; what is owed is the PUBREL. .awaiting_pubcomp => o.resend_pubrel = true, } } } try c.pumpQueued();
return .{ .connected = ack };}
/// Section 4.12, which is the whole of what makes an AUTH meaningful.////// An AUTH carries no packet identifier and nothing that ties it to anything,/// so the only thing distinguishing a step of an exchange from a packet/// somebody made up is that the exchange was asked for and the method matches.fn handleAuth(c: *Connection, a: packet.Auth) Error!?Event { // An AUTH belongs to an exchange the CONNECT started. A broker sending one // to a client that named no method has invented a conversation, and a // caller has nothing to answer it with. const method = c.auth_method orelse return c.fail(error.ProtocolError);
// The method is fixed for the life of the exchange, so a broker naming a // different one -- or none -- is either confused or trying to move the // client onto something weaker. const named = a.properties.authentication_method orelse return c.fail(error.ProtocolError); if (!std.mem.eql(u8, named, method)) return c.fail(error.ProtocolError);
switch (a.reason_code) { // The ordinary step: the broker wants another message. .continue_authentication => {}, // Section 4.12.1 ends a *re-authentication* with this. The first // exchange ends with a CONNACK instead, so a success while still // connecting is a broker skipping the CONNACK -- and a client that // took it would consider itself connected without ever having been // told the session limits. .success => if (c.state != .connected) return c.fail(error.ProtocolError), // Only a client sends this, to ask for re-authentication. A broker // sending it has the direction backwards. .reauthenticate => return c.fail(error.ProtocolError), else => return c.fail(error.ProtocolError), } return .{ .authenticate = a };}
fn handlePublish(c: *Connection, incoming: packet.Publish) Error!?Event { var message = incoming;
if (c.version == .v5_0) { c.resolveIncomingAlias(&message) catch |err| switch (err) { // Running out of memory says nothing about the broker, and leaves // the connection perfectly able to carry on if the caller can find // some. Everything else is the broker breaking the protocol. error.OutOfMemory => return err, else => return c.fail(err), }; }
switch (message.qos) { .at_most_once => return .{ .message = message }, .at_least_once => { const id = message.packet_id orelse return c.fail(error.ProtocolError); try c.enqueue(.{ .puback = .{ .packet_id = id } }); return .{ .message = message }; }, .exactly_once => { const id = message.packet_id orelse return c.fail(error.ProtocolError); c.touchSession(); const entry = try c.inbound_qos2.getOrPut(c.gpa, id); try c.enqueue(.{ .pubrec = .{ .packet_id = id } }); // Already held: this is a retransmission of something delivered the // first time, and delivering it again is exactly what QoS 2 exists // to prevent. if (entry.found_existing) return null; return .{ .message = message }; }, }}
fn handlePuback(c: *Connection, ack: packet.Ack) Error!?Event { c.touchSession(); const index = c.findOutbound(ack.packet_id, .awaiting_puback) orelse return c.fail(error.ProtocolError); const o = c.outbound.orderedRemove(index); c.gpa.free(o.bytes); try c.pumpQueued(); return .{ .published = .{ .packet_id = ack.packet_id, .qos = .at_least_once, .reason_code = ack.reason_code, .properties = ack.properties, } };}
fn handlePubrec(c: *Connection, ack: packet.Ack) Error!?Event { c.touchSession(); const index = c.findOutbound(ack.packet_id, .awaiting_pubrec) orelse return c.fail(error.ProtocolError); const o = &c.outbound.items[index];
// The PUBLISH cannot be needed again either way: the broker has it, or has // refused it. c.gpa.free(o.bytes); o.bytes = &.{};
if (ack.reason_code.isError()) { // Section 4.3.3: a PUBREC that reports a failure ends the handshake // there. No PUBREL, no PUBCOMP. const id = ack.packet_id; _ = c.outbound.orderedRemove(index); try c.pumpQueued(); return .{ .published = .{ .packet_id = id, .qos = .exactly_once, .reason_code = ack.reason_code, .properties = ack.properties, } }; }
o.stage = .awaiting_pubcomp; try c.enqueue(.{ .pubrel = .{ .packet_id = ack.packet_id } }); return null;}
fn handlePubrel(c: *Connection, ack: packet.Ack) Error!?Event { c.touchSession(); // The broker is releasing a message this client received at QoS 2. Whether // or not it is still held -- a PUBCOMP that was lost means this arrives // again after the identifier was forgotten -- the answer is the same. _ = c.inbound_qos2.remove(ack.packet_id); try c.enqueue(.{ .pubcomp = .{ .packet_id = ack.packet_id } }); return null;}
fn handlePubcomp(c: *Connection, ack: packet.Ack) Error!?Event { c.touchSession(); const index = c.findOutbound(ack.packet_id, .awaiting_pubcomp) orelse return c.fail(error.ProtocolError); const o = c.outbound.orderedRemove(index); std.debug.assert(o.bytes.len == 0); try c.pumpQueued(); return .{ .published = .{ .packet_id = ack.packet_id, .qos = .exactly_once, .reason_code = ack.reason_code, .properties = ack.properties, } };}
// -- flow control and identifiers --------------------------------------------
fn findOutbound(c: *Connection, id: PacketId, stage: Outbound.Stage) ?usize { for (c.outbound.items, 0..) |o, i| { if (o.packet_id == id and o.stage == stage) return i; } return null;}
/// Send as many queued publications as the broker's Receive Maximum allows.////// Oldest first, which is what makes a resumed session replay in order, and/// what keeps a slow broker from starving the earliest message indefinitely.fn pumpQueued(c: *Connection) Error!void { if (c.state != .connected) return;
var in_flight: usize = 0; for (c.outbound.items) |o| { if (o.inFlight()) in_flight += 1; }
for (c.outbound.items) |*o| { // A PUBREL a resumed session owes goes out whatever the window says: // its exchange is already counted against the window, and finishing it // is what frees the space. if (o.resend_pubrel) { c.touchSession(); try c.enqueue(.{ .pubrel = .{ .packet_id = o.packet_id } }); o.resend_pubrel = false; continue; } if (o.stage != .queued) continue; if (in_flight >= c.server_receive_maximum) continue;
// A resend carries the duplicate flag. The first send does not, and // `bytes` was encoded without it, so setting the bit is a rewrite of // one nibble of the fixed header rather than a re-encode. c.touchSession(); try c.writeRaw(o.bytes); o.bytes[0] |= 0x08;
o.stage = switch (o.qos) { .at_most_once => unreachable, .at_least_once => .awaiting_puback, .exactly_once => .awaiting_pubrec, }; in_flight += 1; }}
/// The next unused packet identifier.////// Zero is not a valid identifier, so the counter skips it when it wraps. The/// scan for a collision is linear over the in-flight list, which is bounded by/// the broker's Receive Maximum.fn takePacketId(c: *Connection) Error!PacketId { var tried: u32 = 0; while (tried <= std.math.maxInt(PacketId)) : (tried += 1) { const id = c.next_packet_id; c.next_packet_id = if (id == std.math.maxInt(PacketId)) 1 else id + 1; c.touchSession();
var in_use = false; for (c.outbound.items) |o| { if (o.packet_id == id) in_use = true; } if (!in_use) return id; } return error.PacketIdentifiersExhausted;}
// -- topic aliases -----------------------------------------------------------
/// Replace a repeated topic with the alias the broker has been told to use for/// it, or establish a new alias if there is room.////// The policy is the simplest one that is correct: hand out aliases until the/// broker's limit is reached, then stop. Nothing is evicted, because an/// eviction would have to be announced with a full-topic PUBLISH anyway, and a/// client publishing to more distinct topics than the broker will alias is/// better served by not aliasing than by thrashing.fn applyOutgoingAlias(c: *Connection, p: *packet.Publish) Error!void { const limit = @min(c.options.outgoing_topic_alias_maximum, c.server_topic_alias_maximum); if (limit == 0) return;
if (c.outgoing_aliases.get(p.topic)) |alias| { // Established: the topic is now redundant, and leaving it out is the // whole point. p.properties.topic_alias = alias; p.topic = ""; return; }
if (c.next_outgoing_alias > limit) return;
const alias = c.next_outgoing_alias; const owned = try c.gpa.dupe(u8, p.topic); errdefer c.gpa.free(owned); try c.outgoing_aliases.put(c.gpa, owned, alias); c.next_outgoing_alias += 1;
// The packet that establishes an alias carries both the topic and the // alias; only later ones may leave the topic out. p.properties.topic_alias = alias;}
/// Resolve an incoming topic alias, rewriting `p.topic` to the topic the alias/// stands for.////// Every failure here is a protocol error on the broker's part, and the caller/// closes the connection over it. That it *is* the caller who closes, rather/// than this function, was a bug the fuzzer found: written to hand the error/// back as a value, it threw it instead, and a broker sending an alias it had/// never established left the connection reporting an error and still calling/// itself connected.fn resolveIncomingAlias(c: *Connection, p: *packet.Publish) Error!void { const alias = p.properties.topic_alias orelse { // Section 3.3.2.1: with no alias, the topic is the only thing that says // where the message goes, and it may not be empty. if (p.topic.len == 0) return error.ProtocolError; return; }; if (alias > c.options.incoming_topic_alias_maximum) return error.TopicAliasInvalid;
if (p.topic.len != 0) { // Establishing, or re-establishing, the alias. const entry = try c.incoming_aliases.getOrPut(c.gpa, alias); if (entry.found_existing) c.gpa.free(entry.value_ptr.*); entry.value_ptr.* = c.gpa.dupe(u8, p.topic) catch |err| { // Leaving a live entry pointing at freed memory would be worse than // losing the alias. _ = c.incoming_aliases.remove(alias); return err; }; return; }
p.topic = c.incoming_aliases.get(alias) orelse return error.TopicAliasInvalid;}
fn clearIncomingAliases(c: *Connection) void { var it = c.incoming_aliases.valueIterator(); while (it.next()) |topic| c.gpa.free(topic.*); c.incoming_aliases.clearRetainingCapacity();}
fn clearOutgoingAliases(c: *Connection) void { var it = c.outgoing_aliases.keyIterator(); while (it.next()) |topic| c.gpa.free(topic.*); c.outgoing_aliases.clearRetainingCapacity(); c.next_outgoing_alias = 1;}
/// Throw away everything that belongs to the session rather than to the/// connection./// Note that the session has changed. See `session_serial`.fn touchSession(c: *Connection) void { c.session_serial +%= 1;}
/// Write the session down, so that it can outlive this process.////// `client_id` is what will be sent in the CONNECT that resumes it. MQTT/// identifies a session by the client identifier and nothing else, so it is/// written into the file and checked when it is read back; an empty one -- a/// client that lets the broker choose its name -- is `SessionNotResumable`,/// because such a client gets a different name every time and can never resume/// anything.////// May be called at any point. What is in the outgoing buffer is not session/// state and does not need to survive: anything unacknowledged in it is sent/// again when the session resumes, which is the same guarantee that covers a/// connection that broke mid-write.////// See `session` for the format and for what is deliberately left out of it.pub fn saveSession( c: *Connection, w: *std.Io.Writer, client_id: []const u8,) session_format.SaveError!void { var stored: std.ArrayList(session_format.StoredOutbound) = .empty; defer stored.deinit(c.gpa); try stored.ensureTotalCapacity(c.gpa, c.outbound.items.len); for (c.outbound.items) |o| stored.appendAssumeCapacity(.{ .packet_id = o.packet_id, .qos = o.qos, .stage = switch (o.stage) { .queued => .queued, .awaiting_puback => .awaiting_puback, .awaiting_pubrec => .awaiting_pubrec, .awaiting_pubcomp => .awaiting_pubcomp, }, .resend_pubrel = o.resend_pubrel, .bytes = o.bytes, });
var ids: std.ArrayList(PacketId) = .empty; defer ids.deinit(c.gpa); try ids.ensureTotalCapacity(c.gpa, c.inbound_qos2.count()); var it = c.inbound_qos2.keyIterator(); while (it.next()) |id| ids.appendAssumeCapacity(id.*); // Sorted, because a hash map's iteration order is not a fact about the // session: without this the same session writes different bytes on // different runs, and a caller comparing what it wrote last would write // again forever. std.mem.sort(PacketId, ids.items, {}, std.sort.asc(PacketId));
return session_format.write(c.gpa, w, .{ .version = c.version, .client_id = client_id, .next_packet_id = c.next_packet_id, .outbound = stored.items, .inbound_qos2 = ids.items, });}
pub const LoadSessionError = session_format.LoadError || error{ /// `loadSession` after the CONNECT has gone out. The session has to be in /// place before the broker is asked to resume it. AlreadyConnected,};
/// Read a session back, replacing whatever this connection was holding.////// Before `connect`, and with `clean_start` false -- a CONNECT that asks for a/// clean start tells the broker to throw away its half of what has just been/// restored, which leaves the two disagreeing about every message in flight.////// What comes back is what was owed: publications go out again, with the/// duplicate flag where they had already been sent once, as soon as a CONNACK/// says the session is still there. A CONNACK that says it is not clears all of/// this, which is correct and worth expecting -- a broker that forgot the/// session will not accept the second half of a handshake it has no record of.pub fn loadSession( c: *Connection, r: *std.Io.Reader, client_id: []const u8,) LoadSessionError!void { if (c.state == .connecting or c.state == .connected) return error.AlreadyConnected;
var buffer: session_format.Buffer = .{}; defer buffer.deinit(c.gpa); const snapshot = try session_format.read(c.gpa, r, &buffer);
if (snapshot.version != c.version) return error.SessionVersionMismatch; if (!std.mem.eql(u8, snapshot.client_id, client_id)) { return error.SessionBelongsToAnotherClient; }
// Room for everything first, so that a failure part way through does not // leave half a session behind. try c.outbound.ensureTotalCapacity(c.gpa, snapshot.outbound.len); try c.inbound_qos2.ensureTotalCapacity(c.gpa, @intCast(snapshot.inbound_qos2.len));
c.clearSession(); for (snapshot.outbound) |o| c.outbound.appendAssumeCapacity(.{ .packet_id = o.packet_id, .qos = o.qos, .stage = switch (o.stage) { .queued => .queued, .awaiting_puback => .awaiting_puback, .awaiting_pubrec => .awaiting_pubrec, .awaiting_pubcomp => .awaiting_pubcomp, }, .bytes = o.bytes, .resend_pubrel = o.resend_pubrel, }); // Taken over, so the buffer must not free them. buffer.releaseOutbound();
for (snapshot.inbound_qos2) |id| c.inbound_qos2.putAssumeCapacity(id, {}); c.next_packet_id = snapshot.next_packet_id; c.touchSession();}
fn clearSession(c: *Connection) void { c.touchSession(); for (c.outbound.items) |o| c.gpa.free(o.bytes); c.outbound.clearRetainingCapacity(); c.inbound_qos2.clearRetainingCapacity(); c.clearIncomingAliases(); c.clearOutgoingAliases(); c.next_packet_id = 1;}
// -- writing -----------------------------------------------------------------
/// Encode a packet into the outgoing buffer.////// The length is measured first and the capacity reserved, so the writer it is/// handed cannot run out of room. That turns what would otherwise be a/// `WriteFailed` -- an allocation failure wearing a disguise -- back into an/// `OutOfMemory` the caller can act on.fn enqueue(c: *Connection, p: Packet) Error!void { const len = try p.encodedLen(c.version); if (c.server_maximum_packet_size) |max| { if (len > max) return error.PacketTooLarge; } try c.tx.ensureUnusedCapacity(c.gpa, len); var w: std.Io.Writer = .fixed(c.tx.unusedCapacitySlice()); p.encode(&w, c.version) catch |err| switch (err) { error.WriteFailed => unreachable, // capacity was reserved above else => |e| return e, }; c.tx.items.len += w.end;}
/// Encode a packet into memory of its own, for a publication that may have to/// be sent more than once.fn encodeAlloc(c: *Connection, p: Packet) Error![]u8 { const len = try p.encodedLen(c.version); if (c.server_maximum_packet_size) |max| { if (len > max) return error.PacketTooLarge; } const bytes = try c.gpa.alloc(u8, len); errdefer c.gpa.free(bytes); var w: std.Io.Writer = .fixed(bytes); p.encode(&w, c.version) catch |err| switch (err) { error.WriteFailed => unreachable, else => |e| return e, }; std.debug.assert(w.end == len); return bytes;}
fn writeRaw(c: *Connection, bytes: []const u8) Error!void { try c.tx.appendSlice(c.gpa, bytes);}
// -- tests -------------------------------------------------------------------
const testing = std.testing;
/// A connection and the scaffolding a test needs around it: somewhere to put/// the bytes going each way, and a clock that only moves when told.const Harness = struct { conn: Connection, /// Bytes taken off the outgoing queue, which is where a decoded outgoing /// packet borrows from. Copying them out means marking them sent -- and so /// letting the connection compact its buffer -- cannot pull the ground out /// from under an assertion. scratch: [4096]u8 = undefined, now: i64 = 0,
fn init(options: Options) Harness { return .{ .conn = .init(testing.allocator, options) }; }
fn deinit(h: *Harness) void { h.conn.deinit(); }
/// Decode the next packet the connection wants to send, marking it sent. fn take(h: *Harness) !?Packet { const out = h.conn.pending(); if (out.len == 0) return null; const header = try packet.peekFixedHeader(out); const len = header.packetLen(); @memcpy(h.scratch[0..len], out[0..len]); h.conn.sent(len, h.now); return try Packet.decode(header, h.scratch[header.encodedLen()..len], h.conn.version); }
/// Assert that exactly nothing is queued to send. fn expectQuiet(h: *Harness) !void { try testing.expectEqual(@as(usize, 0), h.conn.pending().len); }
/// Hand the connection a packet, as the broker would. fn deliver(h: *Harness, p: Packet) !void { var buf: [4096]u8 = undefined; var w: std.Io.Writer = .fixed(&buf); try p.encode(&w, h.conn.version); try h.conn.receive(w.buffered(), h.now); }
/// Drive the handshake to the point where publishing is allowed. fn connectAndAccept(h: *Harness, ack: packet.ConnAck) !void { try h.conn.connect(.{ .client_id = "test", .keep_alive = 60 }, h.now); _ = try h.take(); try h.deliver(.{ .connack = ack }); const event = (try h.conn.nextEvent()).?; try testing.expectEqual(State.connected, h.conn.state); try testing.expect(event == .connected); }
fn advance(h: *Harness, ms: i64) void { h.now += ms; }};
test "a connection starts idle and refuses commands until it is connected" { var h: Harness = .init(.{}); defer h.deinit();
try testing.expectEqual(State.idle, h.conn.state); try testing.expectError(error.NotConnected, h.conn.publish(.{ .topic = "a" })); try testing.expectError(error.NotConnected, h.conn.subscribe(&.{.{ .filter = "a" }}, .empty));
try h.conn.connect(.{ .client_id = "test" }, h.now); try testing.expectEqual(State.connecting, h.conn.state); // Still not connected: the CONNACK has not arrived. try testing.expectError(error.NotConnected, h.conn.publish(.{ .topic = "a" })); try testing.expectError(error.AlreadyConnected, h.conn.connect(.{}, h.now));}
test "the connect packet advertises the limits this connection enforces" { var h: Harness = .init(.{ .receive_maximum = 5, .max_incoming_packet_size = 4096, .incoming_topic_alias_maximum = 3, }); defer h.deinit();
try h.conn.connect(.{ .client_id = "sensor", .keep_alive = 30 }, h.now); const p = (try h.take()).?.connect;
try testing.expectEqualStrings("sensor", p.client_id); try testing.expectEqual(@as(u16, 30), p.keep_alive); // Advertising a limit the connection would not actually enforce is the // failure mode this guards against. try testing.expectEqual(@as(u16, 5), p.properties.receive_maximum.?); try testing.expectEqual(@as(u32, 4096), p.properties.maximum_packet_size.?); try testing.expectEqual(@as(u16, 3), p.properties.topic_alias_maximum.?);}
test "the broker's connack limits are adopted" { var h: Harness = .init(.{}); defer h.deinit();
try h.connectAndAccept(.{ .properties = .{ .receive_maximum = 2, .maximum_qos = .at_least_once, .retain_available = false, .server_keep_alive = 15, .maximum_packet_size = 64, } });
try testing.expectEqual(@as(u16, 2), h.conn.server_receive_maximum); // Section 3.2.2.3.14: the broker's keep alive replaces what was asked for. try testing.expectEqual(@as(u16, 15), h.conn.keep_alive);
// And the limits are enforced rather than merely recorded. try testing.expectError(error.QosNotSupported, h.conn.publish(.{ .topic = "a", .qos = .exactly_once, })); try testing.expectError(error.RetainNotSupported, h.conn.publish(.{ .topic = "a", .retain = true, })); try testing.expectError(error.PacketTooLarge, h.conn.publish(.{ .topic = "a", .payload = &@as([128]u8, @splat(0)), }));}
test "a refused connection is a disconnection with a reason" { var h: Harness = .init(.{}); defer h.deinit();
try h.conn.connect(.{ .client_id = "test" }, h.now); _ = try h.take(); try h.deliver(.{ .connack = .{ .reason_code = .bad_user_name_or_password, .properties = .{ .reason_string = "no" }, } });
const event = (try h.conn.nextEvent()).?; try testing.expectEqual(ReasonCode.bad_user_name_or_password, event.disconnected.reason_code); try testing.expectEqual(Disconnected.Origin.connack, event.disconnected.origin); try testing.expectEqualStrings("no", event.disconnected.properties.reason_string.?); try testing.expectEqual(State.closed, h.conn.state);}
test "a qos 0 publish is sent and forgotten" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
try testing.expectEqual(@as(?PacketId, null), try h.conn.publish(.{ .topic = "a/b", .payload = "hello", })); const p = (try h.take()).?.publish; try testing.expectEqualStrings("a/b", p.topic); try testing.expectEqualStrings("hello", p.payload); try testing.expectEqual(@as(usize, 0), h.conn.inFlightCount());}
test "a qos 1 publish completes on its puback" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
const id = (try h.conn.publish(.{ .topic = "a/b", .payload = "x", .qos = .at_least_once, })).?; const sent_packet = (try h.take()).?.publish; try testing.expectEqual(id, sent_packet.packet_id.?); try testing.expect(!sent_packet.duplicate); try testing.expectEqual(@as(usize, 1), h.conn.inFlightCount());
try h.deliver(.{ .puback = .{ .packet_id = id, .reason_code = .no_matching_subscribers } }); const event = (try h.conn.nextEvent()).?; try testing.expectEqual(id, event.published.packet_id); try testing.expectEqual(ReasonCode.no_matching_subscribers, event.published.reason_code); // Delivered to nobody is still delivered, as far as the handshake goes. try testing.expect(event.published.succeeded()); try testing.expectEqual(@as(usize, 0), h.conn.inFlightCount());}
test "a qos 2 publish takes four packets" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
const id = (try h.conn.publish(.{ .topic = "a/b", .payload = "x", .qos = .exactly_once, })).?; try testing.expectEqual(id, (try h.take()).?.publish.packet_id.?);
// PUBREC gets a PUBREL back, and no event: nothing has finished. try h.deliver(.{ .pubrec = .{ .packet_id = id } }); try testing.expectEqual(@as(?Event, null), try h.conn.nextEvent()); try testing.expectEqual(id, (try h.take()).?.pubrel.packet_id); try testing.expectEqual(@as(usize, 1), h.conn.inFlightCount());
try h.deliver(.{ .pubcomp = .{ .packet_id = id } }); const event = (try h.conn.nextEvent()).?; try testing.expectEqual(id, event.published.packet_id); try testing.expectEqual(Qos.exactly_once, event.published.qos); try testing.expectEqual(@as(usize, 0), h.conn.inFlightCount());}
test "a pubrec that reports a failure ends the handshake there" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
const id = (try h.conn.publish(.{ .topic = "a/b", .qos = .exactly_once })).?; _ = try h.take();
try h.deliver(.{ .pubrec = .{ .packet_id = id, .reason_code = .not_authorized } }); const event = (try h.conn.nextEvent()).?; try testing.expect(!event.published.succeeded()); try testing.expectEqual(ReasonCode.not_authorized, event.published.reason_code); // Section 4.3.3: no PUBREL follows a rejected PUBREC. try h.expectQuiet(); try testing.expectEqual(@as(usize, 0), h.conn.inFlightCount());}
test "an acknowledgement for a packet that was never sent is a protocol error" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
try h.deliver(.{ .puback = .{ .packet_id = 999 } }); try testing.expectError(error.ProtocolError, h.conn.nextEvent()); // And the connection does not carry on as though nothing happened. try testing.expectEqual(State.closed, h.conn.state);}
test "an incoming qos 1 message is acknowledged before it is handed over" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
try h.deliver(.{ .publish = .{ .topic = "a/b", .payload = "hi", .qos = .at_least_once, .packet_id = 5, } }); const event = (try h.conn.nextEvent()).?; try testing.expectEqualStrings("a/b", event.message.topic); try testing.expectEqualStrings("hi", event.message.payload); try testing.expectEqual(@as(PacketId, 5), (try h.take()).?.puback.packet_id);}
test "a retransmitted qos 2 message is acknowledged but not delivered twice" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
const incoming: Packet = .{ .publish = .{ .topic = "a/b", .payload = "once", .qos = .exactly_once, .packet_id = 7, } };
try h.deliver(incoming); try testing.expectEqualStrings("once", (try h.conn.nextEvent()).?.message.payload); try testing.expectEqual(@as(PacketId, 7), (try h.take()).?.pubrec.packet_id);
// The PUBREC was lost, so the broker sends the message again. It must be // acknowledged again -- and must not reach the application again, which is // the entire purpose of QoS 2. try h.deliver(incoming); try testing.expectEqual(@as(?Event, null), try h.conn.nextEvent()); try testing.expectEqual(@as(PacketId, 7), (try h.take()).?.pubrec.packet_id);
// PUBREL releases it, and the identifier may be used again afterwards. try h.deliver(.{ .pubrel = .{ .packet_id = 7 } }); try testing.expectEqual(@as(?Event, null), try h.conn.nextEvent()); try testing.expectEqual(@as(PacketId, 7), (try h.take()).?.pubcomp.packet_id);
try h.deliver(incoming); try testing.expectEqualStrings("once", (try h.conn.nextEvent()).?.message.payload);}
test "a pubrel for a message already forgotten is still answered" { // A lost PUBCOMP means the broker sends PUBREL again after the identifier // has been released. Answering is what stops it retrying forever. var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
try h.deliver(.{ .pubrel = .{ .packet_id = 42 } }); try testing.expectEqual(@as(?Event, null), try h.conn.nextEvent()); try testing.expectEqual(@as(PacketId, 42), (try h.take()).?.pubcomp.packet_id);}
test "the broker's receive maximum holds publications back" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{ .properties = .{ .receive_maximum = 2 } });
const first = (try h.conn.publish(.{ .topic = "a", .qos = .at_least_once })).?; const second = (try h.conn.publish(.{ .topic = "b", .qos = .at_least_once })).?; const third = (try h.conn.publish(.{ .topic = "c", .qos = .at_least_once })).?;
// All three are accepted; only two are on the wire. try testing.expectEqual(@as(usize, 3), h.conn.inFlightCount()); try testing.expectEqualStrings("a", (try h.take()).?.publish.topic); try testing.expectEqualStrings("b", (try h.take()).?.publish.topic); try h.expectQuiet();
// Acknowledging the first opens the window for the third, in order. try h.deliver(.{ .puback = .{ .packet_id = first } }); _ = try h.conn.nextEvent(); try testing.expectEqualStrings("c", (try h.take()).?.publish.topic);
try h.deliver(.{ .puback = .{ .packet_id = second } }); _ = try h.conn.nextEvent(); try h.deliver(.{ .puback = .{ .packet_id = third } }); _ = try h.conn.nextEvent(); try testing.expectEqual(@as(usize, 0), h.conn.inFlightCount());}
test "a resumed session sends the unacknowledged publications again" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
const qos1 = (try h.conn.publish(.{ .topic = "a", .qos = .at_least_once })).?; const qos2 = (try h.conn.publish(.{ .topic = "b", .qos = .exactly_once })).?; _ = try h.take(); _ = try h.take();
// The QoS 2 one gets as far as PUBREC before the connection dies, so what // it is owed on resumption is a PUBREL and not the message again. try h.deliver(.{ .pubrec = .{ .packet_id = qos2 } }); _ = try h.conn.nextEvent(); _ = try h.take();
h.conn.transportClosed(); try testing.expectEqual(@as(usize, 2), h.conn.inFlightCount());
try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now); _ = try h.take(); try h.deliver(.{ .connack = .{ .session_present = true } }); _ = try h.conn.nextEvent();
// Section 4.4: in the order they were first sent, and flagged as repeats. const again = (try h.take()).?.publish; try testing.expectEqualStrings("a", again.topic); try testing.expectEqual(qos1, again.packet_id.?); try testing.expect(again.duplicate);
try testing.expectEqual(qos2, (try h.take()).?.pubrel.packet_id); try h.expectQuiet();}
test "a session the broker has forgotten takes its publications with it" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
_ = try h.conn.publish(.{ .topic = "a", .qos = .at_least_once }); _ = try h.take(); h.conn.transportClosed();
try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now); _ = try h.take(); // The broker says it has no session. Sending the message again would be // sending it into a session that no longer exists. try h.deliver(.{ .connack = .{ .session_present = false } }); _ = try h.conn.nextEvent();
try testing.expectEqual(@as(usize, 0), h.conn.inFlightCount()); try h.expectQuiet();}
test "a clean start throws the session away before the connect goes out" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
_ = try h.conn.publish(.{ .topic = "a", .qos = .at_least_once }); _ = try h.take(); h.conn.transportClosed();
try h.conn.connect(.{ .client_id = "test", .clean_start = true }, h.now); try testing.expectEqual(@as(usize, 0), h.conn.inFlightCount());}
test "keepalive sends a ping when the connection has been quiet" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{ .properties = .{ .server_keep_alive = 10 } });
try h.conn.tick(h.now); try h.expectQuiet(); try testing.expectEqual(h.now + 10_000, h.conn.deadline().?);
h.advance(10_000); try h.conn.tick(h.now); try testing.expectEqual(@as(Packet, .pingreq), (try h.take()).?);
// One ping at a time: until it is answered, more time passing adds nothing. h.advance(1); try h.conn.tick(h.now); try h.expectQuiet();
try h.deliver(.pingresp); try testing.expectEqual(@as(Event, .pong), (try h.conn.nextEvent()).?); try testing.expect(!h.conn.ping_outstanding);}
test "a broker that stops answering pings is given up on" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{ .properties = .{ .server_keep_alive = 10 } });
h.advance(10_000); try h.conn.tick(h.now); _ = try h.take();
h.advance(10_000); try h.conn.tick(h.now); // exactly at the deadline is still in time h.advance(1); try testing.expectError(error.KeepAliveTimeout, h.conn.tick(h.now)); try testing.expectEqual(State.closed, h.conn.state);}
test "a keep alive of zero turns the mechanism off entirely" { var h: Harness = .init(.{}); defer h.deinit(); try h.conn.connect(.{ .client_id = "test", .keep_alive = 0 }, h.now); _ = try h.take(); try h.deliver(.{ .connack = .{} }); _ = try h.conn.nextEvent();
try testing.expectEqual(@as(?i64, null), h.conn.deadline()); h.advance(1_000_000); try h.conn.tick(h.now); try h.expectQuiet();}
test "sending anything at all postpones the next ping" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{ .properties = .{ .server_keep_alive = 10 } });
h.advance(9_000); _ = try h.conn.publish(.{ .topic = "a" }); _ = try h.take(); // marking it sent is what moves the clock
h.advance(2_000); // 11 seconds since the CONNACK, 2 since the publish try h.conn.tick(h.now); try h.expectQuiet();}
test "subscribe and unsubscribe are answered by their own packet identifiers" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
const sub_id = try h.conn.subscribe(&.{ .{ .filter = "a/#", .qos = .at_least_once }, .{ .filter = "b/+" }, }, .{ .subscription_identifiers = &.{9} });
const sub = (try h.take()).?.subscribe; try testing.expectEqual(sub_id, sub.packet_id); try testing.expectEqual(@as(u32, 9), sub.properties.subscriptionIdentifier().?);
try h.deliver(.{ .suback = .{ .packet_id = sub_id, .reason_codes = &.{ .granted_qos_1, .not_authorized }, } }); const suback = (try h.conn.nextEvent()).?.subscribed; try testing.expectEqual(Qos.at_least_once, suback.reason_codes[0].grantedQos().?); try testing.expect(suback.reason_codes[1].isError());
const unsub_id = try h.conn.unsubscribe(&.{"a/#"}, .empty); try testing.expectEqual(unsub_id, (try h.take()).?.unsubscribe.packet_id); try h.deliver(.{ .unsuback = .{ .packet_id = unsub_id, .reason_codes = &.{.success} } }); try testing.expectEqual(unsub_id, (try h.conn.nextEvent()).?.unsubscribed.packet_id);
// Identifiers are not reused while something is outstanding. try testing.expect(sub_id != unsub_id);}
test "a bad topic is caught before anything is sent" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
try testing.expectError(error.WildcardInTopicName, h.conn.publish(.{ .topic = "a/#" })); try testing.expectError(error.MalformedWildcard, h.conn.subscribe(&.{.{ .filter = "a/#/b" }}, .empty)); try testing.expectError(error.Empty, h.conn.unsubscribe(&.{""}, .empty)); try h.expectQuiet();}
test "an incoming topic alias is remembered and then resolved" { var h: Harness = .init(.{ .incoming_topic_alias_maximum = 4 }); defer h.deinit(); try h.connectAndAccept(.{});
// The packet that establishes the alias carries the topic as well. try h.deliver(.{ .publish = .{ .topic = "sensor/7/temperature", .payload = "20", .properties = .{ .topic_alias = 2 }, } }); try testing.expectEqualStrings("sensor/7/temperature", (try h.conn.nextEvent()).?.message.topic);
// Later ones leave it out, and the connection puts it back. try h.deliver(.{ .publish = .{ .payload = "21", .properties = .{ .topic_alias = 2 } } }); const second = (try h.conn.nextEvent()).?.message; try testing.expectEqualStrings("sensor/7/temperature", second.topic); try testing.expectEqualStrings("21", second.payload);}
test "an alias above the advertised maximum, or one never established, is refused" { var h: Harness = .init(.{ .incoming_topic_alias_maximum = 2 }); defer h.deinit(); try h.connectAndAccept(.{});
try h.deliver(.{ .publish = .{ .topic = "a", .properties = .{ .topic_alias = 3 } } }); try testing.expectError(error.TopicAliasInvalid, h.conn.nextEvent());
var h2: Harness = .init(.{ .incoming_topic_alias_maximum = 2 }); defer h2.deinit(); try h2.connectAndAccept(.{}); try h2.deliver(.{ .publish = .{ .properties = .{ .topic_alias = 1 } } }); try testing.expectError(error.TopicAliasInvalid, h2.conn.nextEvent());}
test "a publish with neither a topic nor an alias is a protocol error" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
try h.deliver(.{ .publish = .{ .topic = "" } }); try testing.expectError(error.ProtocolError, h.conn.nextEvent());}
test "outgoing aliases are established once and used thereafter" { var h: Harness = .init(.{ .outgoing_topic_alias_maximum = 2 }); defer h.deinit(); try h.connectAndAccept(.{ .properties = .{ .topic_alias_maximum = 10 } });
// First time: topic and alias together, so the broker learns the mapping. _ = try h.conn.publish(.{ .topic = "sensor/7/temperature", .payload = "20" }); const first = (try h.take()).?.publish; try testing.expectEqualStrings("sensor/7/temperature", first.topic); try testing.expectEqual(@as(u16, 1), first.properties.topic_alias.?);
// Afterwards: the alias alone, which is the saving the feature exists for. _ = try h.conn.publish(.{ .topic = "sensor/7/temperature", .payload = "21" }); const second = (try h.take()).?.publish; try testing.expectEqualStrings("", second.topic); try testing.expectEqual(@as(u16, 1), second.properties.topic_alias.?);
// A second topic gets the next alias; a third exhausts the limit and is // sent in full rather than evicting anything. _ = try h.conn.publish(.{ .topic = "sensor/8/temperature" }); try testing.expectEqual(@as(u16, 2), (try h.take()).?.publish.properties.topic_alias.?); _ = try h.conn.publish(.{ .topic = "sensor/9/temperature" }); const third = (try h.take()).?.publish; try testing.expectEqualStrings("sensor/9/temperature", third.topic); try testing.expectEqual(@as(?u16, null), third.properties.topic_alias);}
test "outgoing aliases are not used when the broker will not take any" { var h: Harness = .init(.{ .outgoing_topic_alias_maximum = 8 }); defer h.deinit(); // No Topic Alias Maximum in the CONNACK means zero, and a client that sent // an alias anyway would be disconnected for it. try h.connectAndAccept(.{});
_ = try h.conn.publish(.{ .topic = "a/b" }); const p = (try h.take()).?.publish; try testing.expectEqualStrings("a/b", p.topic); try testing.expectEqual(@as(?u16, null), p.properties.topic_alias);}
test "aliases are forgotten when the network connection ends" { // Section 3.3.2.3.4: topic alias mappings belong to the network connection, // not the session, so a resumed session must re-establish them. var h: Harness = .init(.{ .outgoing_topic_alias_maximum = 2, .incoming_topic_alias_maximum = 2 }); defer h.deinit(); try h.connectAndAccept(.{ .properties = .{ .topic_alias_maximum = 10 } });
_ = try h.conn.publish(.{ .topic = "a/b" }); _ = try h.take(); h.conn.transportClosed();
try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now); _ = try h.take(); try h.deliver(.{ .connack = .{ .session_present = true, .properties = .{ .topic_alias_maximum = 10 } } }); _ = try h.conn.nextEvent();
_ = try h.conn.publish(.{ .topic = "a/b" }); const p = (try h.take()).?.publish; try testing.expectEqualStrings("a/b", p.topic); try testing.expectEqual(@as(u16, 1), p.properties.topic_alias.?);}
test "a packet arriving one byte at a time is decoded once it is whole" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
var buf: [128]u8 = undefined; var w: std.Io.Writer = .fixed(&buf); try (Packet{ .publish = .{ .topic = "a/b", .payload = "hello" } }).encode(&w, .v5_0); const bytes = w.buffered();
for (bytes[0 .. bytes.len - 1]) |byte| { try h.conn.receive(&.{byte}, h.now); try testing.expectEqual(@as(?Event, null), try h.conn.nextEvent()); } try h.conn.receive(bytes[bytes.len - 1 ..], h.now); try testing.expectEqualStrings("hello", (try h.conn.nextEvent()).?.message.payload);}
test "several packets in one read are decoded one after another" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
var buf: [256]u8 = undefined; var w: std.Io.Writer = .fixed(&buf); for ([_][]const u8{ "one", "two", "three" }) |payload| { try (Packet{ .publish = .{ .topic = "a", .payload = payload } }).encode(&w, .v5_0); } try h.conn.receive(w.buffered(), h.now);
try testing.expectEqualStrings("one", (try h.conn.nextEvent()).?.message.payload); try testing.expectEqualStrings("two", (try h.conn.nextEvent()).?.message.payload); try testing.expectEqualStrings("three", (try h.conn.nextEvent()).?.message.payload); try testing.expectEqual(@as(?Event, null), try h.conn.nextEvent());}
test "a packet larger than this client will accept ends the connection" { var h: Harness = .init(.{ .max_incoming_packet_size = 32 }); defer h.deinit(); try h.connectAndAccept(.{});
// Only the fixed header is needed to know it is too big, so this never // grows the receive buffer to hold the rest. try h.conn.receive(&.{ 0x30, 0xff, 0x01 }, h.now); try testing.expectError(error.PacketTooLarge, h.conn.nextEvent());}
test "rubbish from the broker ends the connection rather than being skipped" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
// Packet type 0 is reserved. try h.conn.receive(&.{ 0x00, 0x00 }, h.now); try testing.expectError(error.ProtocolError, h.conn.nextEvent()); try testing.expectEqual(State.closed, h.conn.state);}
test "a packet only a broker receives is a protocol error coming the other way" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
try h.deliver(.pingreq); try testing.expectError(error.ProtocolError, h.conn.nextEvent());}
test "the broker can end the connection and say why" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
try h.deliver(.{ .disconnect = .{ .reason_code = .server_shutting_down, .properties = .{ .server_reference = "other.example:1883" }, } }); const event = (try h.conn.nextEvent()).?.disconnected; try testing.expectEqual(ReasonCode.server_shutting_down, event.reason_code); try testing.expectEqual(Disconnected.Origin.disconnect, event.origin); try testing.expectEqualStrings("other.example:1883", event.properties.server_reference.?); try testing.expectEqual(State.closed, h.conn.state);}
test "disconnecting stops the connection taking commands" { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
try h.conn.disconnect(.normal_disconnection, .empty); const p = (try h.take()).?.disconnect; try testing.expectEqual(ReasonCode.success, p.reason_code); try testing.expectEqual(State.disconnecting, h.conn.state); try testing.expectError(error.NotConnected, h.conn.publish(.{ .topic = "a" }));}
test "a 3.1.1 connection speaks 3.1.1" { var h: Harness = .init(.{ .version = .v3_1_1 }); defer h.deinit();
try h.conn.connect(.{ .client_id = "old", .keep_alive = 60 }, h.now); const connect_packet = (try h.take()).?.connect; try testing.expectEqualStrings("old", connect_packet.client_id); // The properties this connection would have advertised in 5.0 have nowhere // to go, and are left out rather than making the packet unencodable. try testing.expect(connect_packet.properties.isEmpty());
try h.deliver(.{ .connack = .{} }); _ = try h.conn.nextEvent();
const id = (try h.conn.publish(.{ .topic = "a/b", .qos = .at_least_once })).?; try testing.expectEqual(id, (try h.take()).?.publish.packet_id.?); try h.deliver(.{ .puback = .{ .packet_id = id } }); try testing.expectEqual(id, (try h.conn.nextEvent()).?.published.packet_id);
// A 3.1.1 broker's CONNACK has no properties to carry a keep alive // override, so what was asked for is what is in force. try testing.expectEqual(@as(u16, 60), h.conn.keep_alive);}
test "the buffers do not grow without bound" { // A long conversation in small pieces: without compaction on both sides // this grows until it runs out of memory rather than staying flat. var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
for (0..2000) |i| { _ = try h.conn.publish(.{ .topic = "a/b", .payload = "x" }); _ = try h.take(); try h.deliver(.{ .publish = .{ .topic = "c/d", .payload = "y" } }); _ = try h.conn.nextEvent(); _ = i; } // Drain the last event so the receive buffer can be released. _ = try h.conn.nextEvent();
try testing.expect(h.conn.rx.capacity < 1024); try testing.expect(h.conn.tx.capacity < 1024);}
test "nothing may precede the connack" { // Section 3.2, [MQTT-3.2.0-1]. A broker that delivers a message before the // connection is established has skipped telling the client what its limits // are, and `Client.connect` has nowhere to put the message. var h: Harness = .init(.{}); defer h.deinit();
try h.conn.connect(.{ .client_id = "test" }, h.now); _ = try h.take(); try h.deliver(.{ .publish = .{ .topic = "a/b", .payload = "too soon" } });
try testing.expectError(error.ProtocolError, h.conn.nextEvent()); try testing.expectEqual(State.closed, h.conn.state);}
test "an unasked-for authentication exchange is refused" { // Section 4.12: an AUTH is a step in an exchange the CONNECT started. A // broker sending one to a client that named no method has invented a // conversation, and there is nothing to answer it with. var h: Harness = .init(.{}); defer h.deinit();
try h.conn.connect(.{ .client_id = "test" }, h.now); _ = try h.take(); try h.deliver(.{ .auth = .{ .reason_code = .continue_authentication, .properties = .{ .authentication_method = "SCRAM-SHA-256" }, } }); try testing.expectError(error.ProtocolError, h.conn.nextEvent());}
test "the authentication method cannot change mid-exchange" { // Section 4.12 fixes the method when the CONNECT goes out. A broker naming // a different one is either confused or moving the client onto something // weaker, and the second is the reason this is checked rather than // tolerated. var h: Harness = .init(.{}); defer h.deinit();
try h.conn.connect(.{ .client_id = "test", .properties = .{ .authentication_method = "SCRAM-SHA-256" }, }, h.now); _ = try h.take(); try h.deliver(.{ .auth = .{ .reason_code = .continue_authentication, .properties = .{ .authentication_method = "PLAIN" }, } }); try testing.expectError(error.ProtocolError, h.conn.nextEvent());}
test "an authentication step with no method at all is refused" { var h: Harness = .init(.{}); defer h.deinit();
try h.conn.connect(.{ .client_id = "test", .properties = .{ .authentication_method = "SCRAM-SHA-256" }, }, h.now); _ = try h.take(); try h.deliver(.{ .auth = .{ .reason_code = .continue_authentication } }); try testing.expectError(error.ProtocolError, h.conn.nextEvent());}
test "a broker cannot end the first exchange with an AUTH success" { // Section 4.12.1 ends a *re-authentication* that way. The first exchange // ends with a CONNACK, and a client that took a bare success for one would // consider itself connected without ever having been told the session's // limits. var h: Harness = .init(.{}); defer h.deinit();
try h.conn.connect(.{ .client_id = "test", .properties = .{ .authentication_method = "SCRAM-SHA-256" }, }, h.now); _ = try h.take(); try h.deliver(.{ .auth = .{ .reason_code = .success, .properties = .{ .authentication_method = "SCRAM-SHA-256" }, } }); try testing.expectError(error.ProtocolError, h.conn.nextEvent());}
test "a broker cannot ask the client to re-authenticate" { // `reauthenticate` travels one way only. A broker that wants another // exchange closes the connection and lets the client start one. var h: Harness = .init(.{}); defer h.deinit();
try h.conn.connect(.{ .client_id = "test", .properties = .{ .authentication_method = "SCRAM-SHA-256" }, }, h.now); _ = try h.take(); try h.deliver(.{ .auth = .{ .reason_code = .reauthenticate, .properties = .{ .authentication_method = "SCRAM-SHA-256" }, } }); try testing.expectError(error.ProtocolError, h.conn.nextEvent());}
test "a success that skipped the exchange is refused" { // The downgrade this exists to stop: the CONNECT asked for a mechanism, // and something answered yes without running it. A client that accepted // that would believe it had authenticated by a method that never ran. var h: Harness = .init(.{}); defer h.deinit();
try h.conn.connect(.{ .client_id = "test", .properties = .{ .authentication_method = "SCRAM-SHA-256" }, }, h.now); _ = try h.take(); try h.deliver(.{ .connack = .{} }); try testing.expectError(error.ProtocolError, h.conn.nextEvent());
// And a CONNACK that names the method is accepted, so the refusal above is // about the missing property and not about the exchange having happened. var ok: Harness = .init(.{}); defer ok.deinit(); try ok.conn.connect(.{ .client_id = "test", .properties = .{ .authentication_method = "SCRAM-SHA-256" }, }, ok.now); _ = try ok.take(); try ok.deliver(.{ .connack = .{ .properties = .{ .authentication_method = "SCRAM-SHA-256" }, } }); try testing.expect((try ok.conn.nextEvent()).? == .connected);
// A refusal need not carry it: there is no exchange left to protect. var refused: Harness = .init(.{}); defer refused.deinit(); try refused.conn.connect(.{ .client_id = "test", .properties = .{ .authentication_method = "SCRAM-SHA-256" }, }, refused.now); _ = try refused.take(); try refused.deliver(.{ .connack = .{ .reason_code = .not_authorized } }); try testing.expectEqual( ReasonCode.not_authorized, (try refused.conn.nextEvent()).?.disconnected.reason_code, );}
test "an AUTH the caller sends carries the method it does not have to remember" { var h: Harness = .init(.{}); defer h.deinit();
// Nothing to continue: the CONNECT named no method. try h.conn.connect(.{ .client_id = "test" }, h.now); _ = try h.take(); try testing.expectError(error.NotAuthenticating, h.conn.auth(.continue_authentication, .empty));
var e: Harness = .init(.{}); defer e.deinit(); try e.conn.connect(.{ .client_id = "test", .properties = .{ .authentication_method = "SCRAM-SHA-256" }, }, e.now); _ = try e.take();
// Filled in, because there is exactly one right answer and section 4.12 // requires it on every AUTH. try e.conn.auth(.continue_authentication, .{ .authentication_data = "client-final" }); const outgoing = (try e.take()).?.auth; try testing.expectEqualStrings("SCRAM-SHA-256", outgoing.properties.authentication_method.?); try testing.expectEqualStrings("client-final", outgoing.properties.authentication_data.?);
// And a caller that names a different one has done the thing the rule // forbids, rather than having its value quietly replaced. try testing.expectError(error.AuthenticationMethodChanged, e.conn.auth( .continue_authentication, .{ .authentication_method = "PLAIN" }, ));
// `success` is the broker's to send, not ours. try testing.expectError(error.BadReasonCode, e.conn.auth(.success, .empty));
// And re-authentication needs a connection to re-authenticate on. try testing.expectError(error.NotConnected, e.conn.auth(.reauthenticate, .empty));}
test "re-authentication runs on a connection that is already up" { // Section 4.12.1: the session is untouched and the exchange happens over // it, which is what makes a credential that expires cost nothing. var h: Harness = .init(.{}); defer h.deinit();
try h.conn.connect(.{ .client_id = "test", .properties = .{ .authentication_method = "SCRAM-SHA-256" }, }, h.now); _ = try h.take(); try h.deliver(.{ .connack = .{ .properties = .{ .authentication_method = "SCRAM-SHA-256" }, } }); _ = try h.conn.nextEvent();
try h.conn.auth(.reauthenticate, .{ .authentication_data = "client-first" }); try testing.expectEqual(ReasonCode.reauthenticate, (try h.take()).?.auth.reason_code);
try h.deliver(.{ .auth = .{ .reason_code = .continue_authentication, .properties = .{ .authentication_method = "SCRAM-SHA-256", .authentication_data = "server-first", }, } }); try testing.expectEqualStrings( "server-first", (try h.conn.nextEvent()).?.authenticate.properties.authentication_data.?, );
// And this time a `success` on an AUTH is the right way to end it. try h.deliver(.{ .auth = .{ .reason_code = .success, .properties = .{ .authentication_method = "SCRAM-SHA-256" }, } }); try testing.expectEqual( ReasonCode.success, (try h.conn.nextEvent()).?.authenticate.reason_code, ); // The session carried on throughout: publishing was never interrupted. _ = try h.conn.publish(.{ .topic = "a/b", .payload = "still here" }); try testing.expectEqualStrings("still here", (try h.take()).?.publish.payload);}
test "an authentication exchange is handed to the caller" { var h: Harness = .init(.{}); defer h.deinit();
try h.conn.connect(.{ .client_id = "test", .properties = .{ .authentication_method = "SCRAM-SHA-256", .authentication_data = "client-first" }, }, h.now); _ = try h.take();
try h.deliver(.{ .auth = .{ .reason_code = .continue_authentication, .properties = .{ .authentication_method = "SCRAM-SHA-256", .authentication_data = "server-first" }, } }); const event = (try h.conn.nextEvent()).?.authenticate; try testing.expectEqualStrings("server-first", event.properties.authentication_data.?);
try h.conn.auth(.continue_authentication, .{ .authentication_method = "SCRAM-SHA-256", .authentication_data = "client-final", }); try testing.expectEqualStrings( "client-final", (try h.take()).?.auth.properties.authentication_data.?, );}
test "a connection that reports an error has stopped" { // The invariant the fuzzer checks after every failure, and which one path // -- an unknown incoming topic alias -- used to break: whatever the error, // the connection is closed afterwards and will not take another command. // A caller that ignores an error must not be able to keep publishing into // a session the broker is about to drop. const Case = struct { bytes: []const u8, expected: anyerror }; const cases = [_]Case{ // An alias that was never established. .{ .bytes = &.{ 0x30, 0x08, 0x00, 0x00, 0x03, 0x23, 0x00, 0x01, 'y', 'o' }, .expected = error.TopicAliasInvalid }, // An alias above the advertised maximum. .{ .bytes = &.{ 0x30, 0x08, 0x00, 0x00, 0x03, 0x23, 0x00, 0x63, 'y', 'o' }, .expected = error.TopicAliasInvalid }, // Neither a topic nor an alias. .{ .bytes = &.{ 0x30, 0x03, 0x00, 0x00, 0x00 }, .expected = error.ProtocolError }, // A reserved packet type. .{ .bytes = &.{ 0x00, 0x00 }, .expected = error.ProtocolError }, // An acknowledgement for something never sent. .{ .bytes = &.{ 0x40, 0x02, 0x03, 0xe7 }, .expected = error.ProtocolError }, };
for (cases) |case| { var h: Harness = .init(.{ .incoming_topic_alias_maximum = 4 }); defer h.deinit(); try h.connectAndAccept(.{});
try h.conn.receive(case.bytes, h.now); try testing.expectError(case.expected, h.conn.nextEvent()); try testing.expectEqual(State.closed, h.conn.state); try testing.expectError(error.NotConnected, h.conn.publish(.{ .topic = "a" })); }}
test "what the connect advertises is what the connection enforces" { // A caller that sets the Topic Alias Maximum property by hand must get a // connection that honours that number, not the one in `Options`. var h: Harness = .init(.{ .incoming_topic_alias_maximum = 2 }); defer h.deinit();
try h.conn.connect(.{ .client_id = "test", .properties = .{ .topic_alias_maximum = 5 } }, h.now); try testing.expectEqual(@as(u16, 5), (try h.take()).?.connect.properties.topic_alias_maximum.?); try h.deliver(.{ .connack = .{} }); _ = try h.conn.nextEvent();
// Alias 5 was advertised as acceptable, so it must be accepted. try h.deliver(.{ .publish = .{ .topic = "a/b", .properties = .{ .topic_alias = 5 } } }); try testing.expectEqualStrings("a/b", (try h.conn.nextEvent()).?.message.topic);}
test "an outgoing alias the broker will not take is refused before it is sent" { var h: Harness = .init(.{}); defer h.deinit(); // No Topic Alias Maximum in the CONNACK means zero: the broker takes none. try h.connectAndAccept(.{});
try testing.expectError(error.TopicAliasInvalid, h.conn.publish(.{ .topic = "a/b", .properties = .{ .topic_alias = 1 }, })); try h.expectQuiet();
// With a broker that does take them, the caller's own alias goes through. var h2: Harness = .init(.{}); defer h2.deinit(); try h2.connectAndAccept(.{ .properties = .{ .topic_alias_maximum = 4 } }); _ = try h2.conn.publish(.{ .topic = "a/b", .properties = .{ .topic_alias = 3 } }); try testing.expectEqual(@as(u16, 3), (try h2.take()).?.publish.properties.topic_alias.?);}
test "a publish that runs out of memory part-way leaves nothing dangling" { // `encodeAlloc` succeeds, the append succeeds, and then `pumpQueued` fails. // The in-flight entry owns the encoded packet by then, so an `errdefer` // reaching that far would hand the same memory back twice. // // Rather than arranging that one ordering by hand, every allocation in the // sequence is failed in turn and the connection is torn down each time; the // testing allocator underneath is what reports a double free or a leak. for (0..24) |budget| { var failing: std.testing.FailingAllocator = .init(testing.allocator, .{ .fail_index = budget }); var conn: Connection = .init(failing.allocator(), .{}); defer conn.deinit();
conn.connect(.{ .client_id = "oom", .keep_alive = 0 }, 0) catch continue; conn.sent(conn.pending().len, 0); conn.receive(&.{ 0x20, 0x03, 0x00, 0x00, 0x00 }, 0) catch continue; _ = conn.nextEvent() catch continue;
_ = conn.publish(.{ .topic = "a/b", .payload = "hello", .qos = .at_least_once, }) catch {}; }}
/// What the session looked like last time it was asked, so that a test can say/// "the bytes changed" and "the serial changed" and compare the two.const SessionWitness = struct { serial: u64 = 0, /// Whether `bytes` holds a session or merely has not been filled in yet, /// which are different things and only the first is worth comparing. primed: bool = false, bytes: std.ArrayList(u8) = .empty,
fn deinit(w: *SessionWitness) void { w.bytes.deinit(testing.allocator); }
/// The guarantee `session_serial` makes: it may move when the session did /// not change, and it may never fail to move when the session did. fn check(w: *SessionWitness, c: *Connection) !void { var out: std.Io.Writer.Allocating = .init(testing.allocator); defer out.deinit(); try c.saveSession(&out.writer, "test");
if (w.primed and !std.mem.eql(u8, out.written(), w.bytes.items) and c.session_serial == w.serial) { return error.SessionSerialMissedAChange; } w.primed = true; w.serial = c.session_serial; w.bytes.clearRetainingCapacity(); try w.bytes.appendSlice(testing.allocator, out.written()); }};
test "the session serial moves whenever the session does" { // The check that keeps `session_serial` worth having. A caller that skips // a save because the serial did not move must be right to skip it, so // every step of a conversation is measured both ways: what the session // serialises to, and what the counter says. var h: Harness = .init(.{}); defer h.deinit(); var witness: SessionWitness = .{}; defer witness.deinit();
try witness.check(&h.conn); try h.connectAndAccept(.{}); try witness.check(&h.conn);
// QoS 1, all the way through. const one = (try h.conn.publish(.{ .topic = "a/b", .payload = "x", .qos = .at_least_once })).?; try witness.check(&h.conn); _ = try h.take(); try witness.check(&h.conn); try h.deliver(.{ .puback = .{ .packet_id = one } }); _ = try h.conn.nextEvent(); try witness.check(&h.conn);
// QoS 2, all four packets. const two = (try h.conn.publish(.{ .topic = "c/d", .payload = "y", .qos = .exactly_once })).?; try witness.check(&h.conn); _ = try h.take(); try witness.check(&h.conn); try h.deliver(.{ .pubrec = .{ .packet_id = two } }); _ = try h.conn.nextEvent(); try witness.check(&h.conn); _ = try h.take(); try witness.check(&h.conn); try h.deliver(.{ .pubcomp = .{ .packet_id = two } }); _ = try h.conn.nextEvent(); try witness.check(&h.conn);
// And the other direction: a QoS 2 message arriving is session state too, // because its identifier has to be remembered until the PUBREL. try h.deliver(.{ .publish = .{ .topic = "e/f", .payload = "z", .qos = .exactly_once, .packet_id = 55, } }); _ = try h.conn.nextEvent(); try witness.check(&h.conn); _ = try h.take(); try witness.check(&h.conn); try h.deliver(.{ .pubrel = .{ .packet_id = 55 } }); _ = try h.conn.nextEvent(); try witness.check(&h.conn);
// Something left in flight, and then the transport goes. _ = try h.conn.publish(.{ .topic = "g/h", .qos = .at_least_once }); try witness.check(&h.conn); _ = try h.take(); try witness.check(&h.conn); h.conn.transportClosed(); try witness.check(&h.conn);
// A resumed session marks it for sending again, which is a change, and so // is the duplicate flag that goes on it when it goes. try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now); _ = try h.take(); try witness.check(&h.conn); try h.deliver(.{ .connack = .{ .session_present = true } }); _ = try h.conn.nextEvent(); try witness.check(&h.conn); _ = try h.take(); try witness.check(&h.conn);
// And a session the broker has forgotten is emptied. h.conn.transportClosed(); try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now); _ = try h.take(); try h.deliver(.{ .connack = .{ .session_present = false } }); _ = try h.conn.nextEvent(); try witness.check(&h.conn);}
test "a session outlives the process that made it" { // The whole point. A QoS 2 publication whose PUBCOMP never arrived is a // message the broker has been told to hold and not yet told to deliver; // a client that forgot it on exit has left it there forever. var out: std.Io.Writer.Allocating = .init(testing.allocator); defer out.deinit(); var packet_id: PacketId = undefined;
{ var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{});
packet_id = (try h.conn.publish(.{ .topic = "a/b", .payload = "important", .qos = .exactly_once, })).?; _ = try h.take(); try h.deliver(.{ .pubrec = .{ .packet_id = packet_id } }); _ = try h.conn.nextEvent(); // The PUBREL goes out, and then the process dies before the PUBCOMP. _ = try h.take();
try h.conn.saveSession(&out.writer, "test"); }
// A new process, a new connection, and the same session. var h: Harness = .init(.{}); defer h.deinit();
var reader: std.Io.Reader = .fixed(out.written()); try h.conn.loadSession(&reader, "test");
// Resumed, not cleaned: a clean start would tell the broker to forget the // half of this that it is holding. try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now); _ = try h.take(); try h.deliver(.{ .connack = .{ .session_present = true } }); _ = try h.conn.nextEvent();
// The PUBREL is owed, and goes again. const owed = (try h.take()).?; try testing.expectEqual(packet_id, owed.pubrel.packet_id);
try h.deliver(.{ .pubcomp = .{ .packet_id = packet_id } }); const done = (try h.conn.nextEvent()).?.published; try testing.expectEqual(packet_id, done.packet_id);
// And the identifier counter came back too, so the next message cannot // collide with one the broker is still holding. const next = (try h.conn.publish(.{ .topic = "c/d", .qos = .at_least_once })).?; try testing.expect(next != packet_id);}
test "a publication that never went out survives as well" { // Not section 4.1's session state -- it was never sent -- but losing it // would be losing a message the caller believes it published, which is a // worse failure than storing slightly more than the standard requires. var out: std.Io.Writer.Allocating = .init(testing.allocator); defer out.deinit();
{ var h: Harness = .init(.{}); defer h.deinit(); // A window of one, so the second publication is held back. try h.connectAndAccept(.{ .properties = .{ .receive_maximum = 1 } }); _ = try h.conn.publish(.{ .topic = "a/b", .qos = .at_least_once }); _ = try h.conn.publish(.{ .topic = "c/d", .payload = "waiting", .qos = .at_least_once }); _ = try h.take(); try h.expectQuiet(); try h.conn.saveSession(&out.writer, "test"); }
var h: Harness = .init(.{}); defer h.deinit(); var reader: std.Io.Reader = .fixed(out.written()); try h.conn.loadSession(&reader, "test"); try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now); _ = try h.take(); try h.deliver(.{ .connack = .{ .session_present = true } }); _ = try h.conn.nextEvent();
// Both go out: the one that had been sent carries the duplicate flag, and // the one that never did does not. const first = (try h.take()).?.publish; try testing.expectEqualStrings("a/b", first.topic); try testing.expect(first.duplicate); const second = (try h.take()).?.publish; try testing.expectEqualStrings("c/d", second.topic); try testing.expect(!second.duplicate);}
test "a session belongs to one client and one protocol version" { var out: std.Io.Writer.Allocating = .init(testing.allocator); defer out.deinit(); { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{}); _ = try h.conn.publish(.{ .topic = "a/b", .qos = .at_least_once }); try h.conn.saveSession(&out.writer, "test"); }
// Another client's unacknowledged messages are not this client's to send. { var h: Harness = .init(.{}); defer h.deinit(); var reader: std.Io.Reader = .fixed(out.written()); try testing.expectError( error.SessionBelongsToAnotherClient, h.conn.loadSession(&reader, "somebody-else"), ); }
// And the packets in it are encoded for the version that wrote them. { var h: Harness = .init(.{ .version = .v3_1_1 }); defer h.deinit(); var reader: std.Io.Reader = .fixed(out.written()); try testing.expectError( error.SessionVersionMismatch, h.conn.loadSession(&reader, "test"), ); }
// Loading is something that happens before the CONNECT, because the broker // has to be asked to resume what has just been restored. { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{}); var reader: std.Io.Reader = .fixed(out.written()); try testing.expectError(error.AlreadyConnected, h.conn.loadSession(&reader, "test")); }}
test "a broker that forgot the session throws away what was restored" { // The honest outcome rather than a hidden one: a broker with no record of // the session will not accept the second half of a handshake it never saw, // so sending the restored messages again would be sending them into // nothing. var out: std.Io.Writer.Allocating = .init(testing.allocator); defer out.deinit(); { var h: Harness = .init(.{}); defer h.deinit(); try h.connectAndAccept(.{}); _ = try h.conn.publish(.{ .topic = "a/b", .qos = .exactly_once }); _ = try h.take(); try h.conn.saveSession(&out.writer, "test"); }
var h: Harness = .init(.{}); defer h.deinit(); var reader: std.Io.Reader = .fixed(out.written()); try h.conn.loadSession(&reader, "test"); try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now); _ = try h.take(); try h.deliver(.{ .connack = .{ .session_present = false } }); _ = try h.conn.nextEvent();
try h.expectQuiet();}
test "asking the broker to keep a session is something you have to ask for" { // Section 3.1.2.11.2: a missing Session Expiry Interval means zero, and // zero means the session ends with the network connection. So // `clean_start = false` on its own asks the broker to resume a session it // is about to throw away, which is a trap worth having a knob for -- // measured against mosquitto, which does exactly that. { var h: Harness = .init(.{}); defer h.deinit(); try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now); const out = (try h.take()).?.connect; try testing.expectEqual(@as(?u32, null), out.properties.session_expiry_interval); }
{ var h: Harness = .init(.{ .session_expiry_interval = 3600 }); defer h.deinit(); try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now); const out = (try h.take()).?.connect; try testing.expectEqual(@as(?u32, 3600), out.properties.session_expiry_interval); }
// What the caller put in the CONNECT wins, including a zero that means // "drop it after all". { var h: Harness = .init(.{ .session_expiry_interval = 3600 }); defer h.deinit(); try h.conn.connect(.{ .client_id = "test", .clean_start = false, .properties = .{ .session_expiry_interval = 0 }, }, h.now); const out = (try h.take()).?.connect; try testing.expectEqual(@as(?u32, 0), out.properties.session_expiry_interval); }
// 3.1.1 has no such property: `clean_session = 0` asks the broker to keep // the session and the broker decides for how long. { var h: Harness = .init(.{ .version = .v3_1_1, .session_expiry_interval = 3600 }); defer h.deinit(); try h.conn.connect(.{ .client_id = "test", .clean_start = false }, h.now); const out = (try h.take()).?.connect; try testing.expectEqual(@as(?u32, null), out.properties.session_expiry_interval); }}