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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053// SPDX-FileCopyrightText: © 2026 Jeffrey C. Ollie <jeff@ocjtech.us>// SPDX-License-Identifier: MIT
//! A `Connection` with the byte-shuffling done for you.//!//! This is the thin layer between the sans-I/O core and a stream: it moves//! bytes out of a `std.Io.Writer` and into a `std.Io.Reader`, asks the clock//! what time it is, and services the keepalive timer. It adds no protocol//! knowledge whatsoever -- everything MQTT is in `mqtt.Connection`, in the//! module this one depends on -- which is why it is about two hundred lines//! and why anything it cannot do can be done by driving `Connection` directly.//!//! The reader and writer are the caller's. That is the whole point: a plain TCP//! stream, a TLS session, a WebSocket carrying binary frames, a pair of pipes,//! or a buffer in a test all present the same two interfaces, and this layer//! cannot tell them apart.//!//! ```zig//! var read_buffer: [4096]u8 = undefined;//! var write_buffer: [4096]u8 = undefined;//! var stream_reader = stream.reader(io, &read_buffer);//! var stream_writer = stream.writer(io, &write_buffer);//!//! // The session is the caller's, because it outlives this transport and any//! // number of later ones -- see `Reconnecting`, which relies on exactly that.//! var conn: mqtt.Connection = .init(gpa, .{});//! defer conn.deinit();//!//! var client: Client = .initOver(&conn, io, transport, streams);//!//! _ = try client.connect(.{ .client_id = "sensor-7" });//! _ = try client.subscribe(&.{.{ .filter = "commands/#" }}, .empty);//!//! while (try client.next()) |event| switch (event) {//! .message => |m| std.debug.print("{s}: {s}\n", .{ m.topic, m.payload }),//! else => {},//! }//! ```//!//! **A plain `Client` cannot keep itself alive.** A `std.Io.Reader` has no//! deadline, so `next` blocks in it for as long as the transport does -- and a//! PINGREQ that falls due while it is blocked is never sent, so a broker//! eventually decides the client is gone and closes the connection. On an idle//! subscription that is a disconnection every keep-alive-and-a-half, which is//! not a subtle failure but is a silent one.//!//! `initOver` is the answer: build the client over a `Transport` and every wait//! is bounded by whenever the connection next wants the clock looked at. The//! same constructor picks up the transport's flush, for one that buffers below//! the writer it hands out -- neither transport here does, but a client built//! over one that did and flushed only its own writer would send nothing at all.//!//! `Reconnecting` always uses it. `init` is for a reader and writer that came//! from somewhere else entirely; there, keepalive is the caller's problem and//! `Connection.deadline` says when it falls due.
const std = @import("std");const mqtt = @import("mqtt");const sasl = @import("sasl");
const Transport = @import("Transport.zig");
const Connection = mqtt.Connection;const protocol = mqtt.protocol;const packet = protocol.packet;
const PacketId = protocol.types.PacketId;const Properties = protocol.properties.Properties;const ReasonCode = protocol.types.ReasonCode;
const Client = @This();
pub const Event = Connection.Event;pub const Message = Connection.Message;pub const Options = Connection.Options;pub const ConnectOptions = Connection.ConnectOptions;
pub const Error = Connection.Error || sasl.Client.Error || std.Io.Reader.Error || std.Io.Writer.Error || error{ /// The broker closed the transport in the middle of a packet. UnexpectedEndOfStream, /// The broker refused the connection. The CONNACK's reason code is in /// `refusal`. ConnectionRefused, /// The broker asked for an extended authentication exchange and no /// mechanism was given to answer it with. Set `authentication` before /// calling `connect`, or drive `Connection.auth` and `nextEvent` yourself. AuthenticationRequired, /// The broker accepted the connection, and the mechanism says it has not /// finished proving what it set out to prove. /// /// For a mechanism with mutual authentication -- SCRAM, DIGEST-MD5 -- this /// is the case the mutual half exists to catch: something answered, and it /// did not know the secret it would have needed to answer properly. The /// connection is not usable and has been disconnected. ServerNotAuthenticated, /// The response a mechanism wanted to send is longer than /// `Authentication.buffer`. AuthenticationDataTooLong, /// `reauthenticate` on a connection that never authenticated by a /// mechanism in the first place. Section 4.12.1 has nothing to say about /// that connection. NotAuthenticating,};
/// How to answer a broker that wants more than a user name and a password.////// MQTT 5.0 section 4.12 carries a SASL exchange in its own packets: the method/// name goes in the CONNECT, each challenge arrives as an AUTH, and the last/// message from the broker rides on the CONNACK. What it does *not* do is say/// which mechanism -- so this takes `sasl.Client`, the interface, and the/// mechanism is whatever the caller built.////// ```zig/// var plain: sasl.Plain = .init("user", "hunter2");/// var buffer: [512]u8 = undefined;/// client.authentication = .{ .mechanism = plain.client(), .buffer = &buffer };/// ```////// SCRAM is the same three lines through zig-scram's `scram-sasl` adapter,/// and nothing here knows the difference.pub const Authentication = struct { /// The mechanism, which runs **once**. A `Client` is one connection, so /// that is the right lifetime; `Reconnecting.Authenticator` is how to get /// a fresh one per attempt. mechanism: sasl.Client, /// Where each outgoing message is built. It only has to hold one message /// of whatever mechanism this is -- a few hundred bytes for SCRAM. buffer: []u8,};
/// The session, which this borrows rather than owns.////// Borrowing is what makes `Reconnecting` possible: a `Client` is one/// transport, a `Connection` is a session, and a session outlives any number of/// transports. It also keeps the layering honest -- everything MQTT is over/// there, and a caller can reach past this at any point without unpicking/// anything.conn: *Connection,io: std.Io,reader: *std.Io.Reader,writer: *std.Io.Writer,
/// The parts of the transport this needs beyond a reader and a writer: how to/// wait with a deadline, and how to push bytes the writer has handed down but/// not sent. See `next` and `flush`.hooks: Transport.Hooks = .none,
/// Why the broker refused, kept from the CONNACK so that `connect` can return/// an error and still say what happened.refusal: ?Connection.Disconnected = null,
/// The mechanism to authenticate with, when the broker wants an exchange.////// Set before `connect`. Left null, an AUTH from the broker comes back as/// `error.AuthenticationRequired` and the exchange is the caller's to drive.authentication: ?Authentication = null,
pub fn init( conn: *Connection, io: std.Io, reader: *std.Io.Reader, writer: *std.Io.Writer,) Client { return .{ .conn = conn, .io = io, .reader = reader, .writer = writer };}
/// A client over a transport, which is the form to prefer.////// `streams` is what `transport.open` returned. The transport is asked for its/// hooks rather than being kept, because the two things this needs from it --/// waiting with a deadline, and pushing what is buffered below the writer --/// are the only parts a client has any business with.////// Both matter. Without the first, a `std.Io.Reader` has no deadline, so an/// idle connection is eventually closed by the broker for never having pinged./// Without the second, a TLS session encrypts every packet into a buffer and/// sends none of it.pub fn initOver( conn: *Connection, io: std.Io, transport: Transport, streams: Transport.Streams,) Client { return .{ .conn = conn, .io = io, .reader = streams.reader, .writer = streams.writer, .hooks = transport.hooks(), };}
/// The clock, in milliseconds, which is the only thing this layer asks the/// `Io` for.////// `.boot` rather than `.awake`, because it counts time the machine spent/// suspended and the broker's keepalive timer certainly did: a laptop that/// slept for an hour has been disconnected for an hour, and a client using a/// clock that stopped with it would wake up believing it had just pinged.fn now(c: *const Client) i64 { return std.Io.Timestamp.now(c.io, .boot).toMilliseconds();}
/// Write everything the connection has queued, and flush it.////// Called by everything else here, so a caller only needs it when driving the/// connection directly between calls to this layer.pub fn flush(c: *Client) Error!void { while (true) { const out = c.conn.pending(); if (out.len == 0) break; try c.writer.writeAll(out); c.conn.sent(out.len, c.now()); } try c.writer.flush(); // And then whatever is below it. For a socket that is nothing; for a TLS // session it is the difference between having encrypted the packet and // having sent it. if (c.hooks.flush) |push| push(c.hooks.context, c.io) catch return error.WriteFailed;}
/// What `nextBefore` found: one of three outcomes, where `next` has only two.pub const Waited = union(enum) { event: Event, /// The broker closed the transport cleanly between packets, which is the /// normal end of a conversation that finished with a DISCONNECT. end_of_stream, /// Nothing arrived before the deadline. The connection is untouched and /// asking again carries on where this left off. timeout,};
/// The next event, reading from the transport for as long as it takes.////// Returns null when the broker closes the transport cleanly between packets,/// which is the normal end of a conversation that finished with a DISCONNECT./// A transport that ends in the middle of a packet is `UnexpectedEndOfStream`.pub fn next(c: *Client) Error!?Event { return switch (try c.nextBefore(.none)) { .event => |event| event, .end_of_stream => null, // `.none` is the one timeout that cannot expire. .timeout => unreachable, };}
/// The next event, giving up at `timeout` and answering `.timeout` if nothing/// has arrived by then.////// For a caller that has something else to do on the same task. MQTT is one/// connection carrying both directions, and this type is not safe to use from/// two tasks at once -- `publish` and this both drive the same `Connection`,/// which owns the transmit buffer, the packet identifier counter and the/// in-flight window. A subscriber that also publishes therefore has to/// interleave the two on one task, and an unbounded wait gives it no moment in/// which to do so.////// Keepalive is unaffected: the wait is the earlier of this deadline and/// whenever the connection next wants the clock looked at, and a PINGREQ that/// falls due while waiting goes out before this returns.////// A deadline needs a client built with `initOver`, because the bound is/// applied to the transport's `readable` hook. A plain `init` over a bare/// reader has nothing to bound -- a `std.Io.Reader` cannot be asked to wait/// only so long -- and the read blocks as it always did.pub fn nextBefore(c: *Client, timeout: std.Io.Timeout) Error!Waited { // Resolved to an absolute instant once, here, rather than being applied on // each turn of the loop below: a duration restarted after every PINGREQ // would be a deadline that never arrives. // Measured as a remaining duration and added to this clock's own reading, // rather than with `Clock.Timestamp.toClock`: that is how a deadline on // any clock lands in the milliseconds-since-boot this file works in. const wait_until: ?i64 = switch (timeout.toDeadline(c.io)) { .none => null, .deadline => |at| c.now() + at.durationFromNow(c.io).raw.toMilliseconds(), // `toDeadline` returns one of the two above and never a duration. .duration => unreachable, };
while (true) { if (try c.conn.nextEvent()) |event| { // Decoding may have queued a reply: the PUBACK owed for an incoming // QoS 1 message, a PUBREL, or the publications a just-resumed // session has to send again. Those go out now rather than waiting // for the next read to block -- a caller draining a burst of // buffered events would otherwise sit on its acknowledgements, // and a broker waiting for them would stop sending. // // Flushing does not touch the receive buffer, so the event being // returned is still good. if (c.conn.wantsWrite()) try c.flush(); return .{ .event = event }; }
// Anything the connection wants to say, including a PINGREQ that has // just fallen due, goes out before the wait. try c.conn.tick(c.now()); try c.flush();
// With a waiter, the wait is bounded by whenever the connection next // wants the clock looked at, so a PINGREQ falling due in the middle of // an idle hour is actually sent. Without one the read below blocks // until the transport says otherwise, and an idle connection is // eventually closed by the broker for not having pinged -- which is // why `initOver` exists and why `Reconnecting` always uses it. if (c.hooks.readable) |readable| { while (c.reader.bufferedLen() == 0) { // Tested here, where the alternative is to wait, rather than // earlier: a caller polling with a deadline already past still // gets everything that has arrived, and still pings. Reaching // it a second time means the deadline that just expired was // the caller's and not the connection's, whose expiry the tick // below has already answered. if (c.expired(wait_until)) return .timeout; readable(c.hooks.context, c.io, c.waitTimeout(wait_until)) catch |err| switch (err) { error.Timeout => { try c.conn.tick(c.now()); try c.flush(); continue; }, else => return error.ReadFailed, }; break; } } else if (c.reader.bufferedLen() == 0 and c.expired(wait_until)) { // Nothing in hand and nothing to bound the read below, which would // block past the deadline and make a liar of it. return .timeout; }
const available = c.reader.peekGreedy(1) catch |err| switch (err) { error.EndOfStream => { // Clean only if nothing was part-way through. if (c.conn.hasPartialPacket()) return error.UnexpectedEndOfStream; return .end_of_stream; }, error.ReadFailed => return error.ReadFailed, }; try c.conn.receive(available, c.now()); c.reader.toss(available.len); }}
/// Whether a caller's deadline, if there is one, has arrived.fn expired(c: *const Client, wait_until: ?i64) bool { const until = wait_until orelse return false; return c.now() >= until;}
/// How long to wait: until the connection wants the clock looked at again, or/// until the caller's deadline, whichever comes first.fn waitTimeout(c: *const Client, wait_until: ?i64) std.Io.Timeout { const deadline_ms = if (c.conn.deadline()) |wants| if (wait_until) |until| @min(wants, until) else wants else wait_until orelse return .none; const remaining = deadline_ms - c.now(); return .{ .duration = .{ .raw = .fromMilliseconds(@max(0, remaining)), // The same clock the deadline was computed on, so that a machine which // suspends for an hour wakes up knowing the hour passed. .clock = .boot, }, };}
/// The next event that can be produced from bytes already in hand, without/// waiting for the transport.////// For a caller with an event loop of its own: poll the transport, `feed` what/// it gave you, then drain this until it returns null.pub fn nextBuffered(c: *Client) Error!?Event { return c.conn.nextEvent();}
/// Hand over bytes read from the transport by somebody else.pub fn feed(c: *Client, bytes: []const u8) Error!void { try c.conn.receive(bytes, c.now());}
/// Send a CONNECT and wait for the CONNACK, carrying on an authentication/// exchange if the broker asks for one.////// Nothing can arrive before a CONNACK except an AUTH, so unlike the other/// round trips here this one cannot swallow an event the caller wanted.////// With `authentication` set, the whole of section 4.12 happens inside this/// call: the method and the mechanism's first message go out in the CONNECT,/// every AUTH is answered, and the data on the CONNACK -- which is where a/// mechanism's *last* message arrives, the one where the broker proves itself/// -- is fed back to the mechanism before this returns. Without it an AUTH is/// reported as `error.AuthenticationRequired` and the conversation is the/// caller's to hold with `Connection.auth` and `nextEvent`.pub fn connect(c: *Client, options: ConnectOptions) Error!packet.ConnAck { try c.conn.connect(try c.prepareConnect(options), c.now()); try c.flush();
while (true) { const event = (try c.next()) orelse return error.UnexpectedEndOfStream; switch (event) { .connected => |ack| { c.finishAuthentication(ack) catch |err| { // Nothing about this connection is worth keeping, and the // broker is owed a DISCONNECT rather than a silence. c.disconnect(.normal_disconnection, .empty) catch {}; return err; }; return ack; }, .disconnected => |d| { c.refusal = d; return error.ConnectionRefused; }, .authenticate => |challenge| try c.answerChallenge(challenge), // Nothing else can precede a CONNACK, and `Connection` has already // refused anything that tried. else => unreachable, } }}
/// Fill in what section 4.12 requires the CONNECT to carry, when there is a/// mechanism to authenticate with.////// The method names the mechanism and the data, if it has anything to say/// first, is its opening message. `Initial.none` is a mechanism that will not/// speak until spoken to, which here means sending the method and no data.////// Public because `Reconnecting` sends its own CONNECT and needs the same/// preparation; a caller driving `Connection` directly can use it too.pub fn prepareConnect(c: *Client, options: ConnectOptions) Error!ConnectOptions { const a = if (c.authentication) |*auth_| auth_ else return options; if (options.properties.authentication_method != null) return error.AuthenticationMethodChanged;
var opts = options; opts.properties.authentication_method = a.mechanism.name(); var w: std.Io.Writer = .fixed(a.buffer); switch (a.mechanism.initial(&w) catch |err| return mapSasl(err)) { .written => opts.properties.authentication_data = w.buffered(), .none => {}, } return opts;}
/// Answer one AUTH by running the mechanism over the data it carried.////// `error.AuthenticationRequired` when there is no mechanism, which is a broker/// asking for a conversation nobody here can hold.pub fn answerChallenge(c: *Client, challenge: packet.Auth) Error!void { const a = if (c.authentication) |*auth_| auth_ else return error.AuthenticationRequired; var w: std.Io.Writer = .fixed(a.buffer); // An AUTH with no data is a challenge of zero bytes, which several // mechanisms expect at least once and none can tell from a missing one. const data = challenge.properties.authentication_data orelse ""; a.mechanism.respond(data, &w) catch |err| return mapSasl(err); try c.auth(.continue_authentication, .{ .authentication_data = w.buffered() });}
/// Finish an exchange at the CONNACK, which is where MQTT puts the last/// message of a mechanism that has one.////// Most carriers have nowhere to put it -- SMTP says so outright -- and have to/// deliver it as one more challenge answered with nothing. MQTT does not, which/// is the one way this profile is easier than the others.////// The `satisfied` check is what the mutual half of a mechanism exists for./// Something accepted the connection; that says it is willing to talk to us and/// nothing at all about whether it is who it claims to be.pub fn finishAuthentication(c: *Client, ack: packet.ConnAck) Error!void { const a = if (c.authentication) |*auth_| auth_ else return; if (ack.properties.authentication_data) |data| { var w: std.Io.Writer = .fixed(a.buffer); a.mechanism.respond(data, &w) catch |err| return mapSasl(err); // Whatever it wanted to say back is dropped: section 4.12 ends the // exchange at the CONNACK and there is no packet left to carry a // reply. Every mechanism that can finish here answers an empty // response anyway. } if (!a.mechanism.satisfied()) return error.ServerNotAuthenticated;}
/// `sasl.Client.Error` is mostly `Io.Writer.Error`, which here means the/// mechanism ran out of the buffer it was given rather than that a socket/// broke -- and a caller told `WriteFailed` would go looking at the transport.fn mapSasl(err: sasl.Client.Error) Error { return switch (err) { error.WriteFailed => error.AuthenticationDataTooLong, else => |e| e, };}
/// Queue an AUTH and send it, continuing an authentication exchange.////// The method is filled in by `Connection`, which knows what the CONNECT/// named. For a caller driving an exchange by hand; `connect` uses it too.pub fn auth(c: *Client, reason_code: ReasonCode, properties: Properties) Error!void { try c.conn.auth(reason_code, properties); try c.flush();}
/// Authenticate again on a connection that is already up: section 4.12.1.////// The mechanism in `authentication` runs from the beginning, so it must be a/// fresh one -- a SASL exchange is a state machine that runs once, and the one/// that established this connection has finished. Replace it first:////// ```zig/// var again: sasl.Plain = .init("user", "hunter2");/// client.authentication = .{ .mechanism = again.client(), .buffer = &buffer };/// try client.reauthenticate();/// ```////// The session is untouched and messages keep flowing throughout, which is/// what section 4.12.1 is for: a credential that expires need not cost a/// reconnection. Events that arrive while this is waiting are handled as they/// would be by `next`, which means a message can be **dropped** -- like the/// other round trips here, this is for a client that is not also a subscriber.pub fn reauthenticate(c: *Client) Error!void { try c.beginReauthentication();
while (true) { const event = (try c.next()) orelse return error.UnexpectedEndOfStream; switch (event) { .authenticate => |challenge| switch (challenge.reason_code) { .success => return c.finishReauthentication(challenge), else => try c.answerChallenge(challenge), }, // Section 4.12.1: a broker that will not re-authenticate closes // the connection rather than refusing in place. .disconnected => |d| { c.refusal = d; return error.ConnectionRefused; }, else => {}, } }}
/// Send the AUTH that starts a re-authentication and return without waiting.////// For a caller that drives its own event loop and would rather not have this/// one swallow what arrives while the exchange runs. `answerChallenge` handles/// each `continue_authentication` and `finishReauthentication` the `success`/// that ends it; `Reconnecting` does exactly that.pub fn beginReauthentication(c: *Client) Error!void { const a: *Authentication = if (c.authentication) |*auth_| auth_ else return error.NotAuthenticating; var w: std.Io.Writer = .fixed(a.buffer); const initial = a.mechanism.initial(&w) catch |err| return mapSasl(err); try c.auth(.reauthenticate, switch (initial) { .written => .{ .authentication_data = w.buffered() }, .none => .empty, });}
/// Finish a re-authentication at the AUTH the broker ended it with.////// The broker is content, which is what its `success` says. Whether we are is/// a separate question and the one that matters -- the same check `connect`/// makes at the CONNACK, for the same reason.pub fn finishReauthentication(c: *Client, challenge: packet.Auth) Error!void { const a: *Authentication = if (c.authentication) |*auth_| auth_ else return error.NotAuthenticating; if (challenge.properties.authentication_data) |data| { var w: std.Io.Writer = .fixed(a.buffer); a.mechanism.respond(data, &w) catch |err| return mapSasl(err); } if (!a.mechanism.satisfied()) return error.ServerNotAuthenticated;}
/// Publish a message, returning the packet identifier a later `.published`/// event will carry, or null for QoS 0.////// Returns as soon as the message is on the transport. Waiting for the/// acknowledgement is the caller's business, because a client that waits cannot/// have more than one message in flight and the flow control window exists/// precisely so that it can.pub fn publish(c: *Client, message: Message) Error!?PacketId { const id = try c.conn.publish(message); try c.flush(); return id;}
/// Subscribe, returning the packet identifier the SUBACK will carry.////// The SUBACK arrives as a `.subscribed` event from `next`, along with whatever/// else the broker has to say -- which will include messages on the new/// subscription, possibly before the SUBACK itself.pub fn subscribe( c: *Client, filters: []const packet.TopicFilter, properties: Properties,) Error!PacketId { const id = try c.conn.subscribe(filters, properties); try c.flush(); return id;}
/// Unsubscribe, returning the packet identifier the UNSUBACK will carry.pub fn unsubscribe( c: *Client, filters: []const []const u8, properties: Properties,) Error!PacketId { const id = try c.conn.unsubscribe(filters, properties); try c.flush(); return id;}
/// Send a DISCONNECT and flush it.////// The transport is the caller's to close, and should be closed after this: a/// DISCONNECT that is still in a buffer when the socket goes is a DISCONNECT/// that never happened, and the broker will publish the will message.pub fn disconnect(c: *Client, reason_code: ReasonCode, properties: Properties) Error!void { try c.conn.disconnect(reason_code, properties); try c.flush();}
/// Send a PINGREQ now. `next` sends these when they fall due; this is for a/// caller that wants to know the broker is there.pub fn ping(c: *Client) Error!void { try c.conn.ping(c.now()); try c.flush();}
/// The transport has gone. Keeps the session so that a new `Client` over a new/// transport can resume it -- see `Connection.transportClosed`.pub fn transportClosed(c: *Client) void { c.conn.transportClosed();}
const testing = std.testing;
/// A pair of buffers pretending to be a transport, so that the layer can be/// tested without a socket. Which is, after all, the claim it makes.const Loopback = struct { to_client: std.Io.Reader, from_client: std.Io.Writer.Allocating,
fn init(broker_says: []const u8) Loopback { return .{ .to_client = .fixed(broker_says), .from_client = .init(testing.allocator), }; }
fn deinit(l: *Loopback) void { l.from_client.deinit(); }};
/// The bytes a broker would send to accept a connection: a CONNACK with nothing/// to say.const connack_bytes = [_]u8{ 0x20, 0x03, 0x00, 0x00, 0x00 };
test "a client connects over a pair of buffers" { var loop: Loopback = .init(&connack_bytes); defer loop.deinit();
var conn: Connection = .init(testing.allocator, .{}); defer conn.deinit(); var client: Client = .init(&conn, testing.io, &loop.to_client, &loop.from_client.writer);
const ack = try client.connect(.{ .client_id = "test", .keep_alive = 0 }); try testing.expect(!ack.session_present); try testing.expectEqual(ReasonCode.success, ack.reason_code);
// And the CONNECT really did reach the writer. const written = loop.from_client.written(); const header = try packet.peekFixedHeader(written); const sent_connect = try packet.Packet.decode( header, written[header.encodedLen()..header.packetLen()], .v5_0, ); try testing.expectEqualStrings("test", sent_connect.connect.client_id);}
test "a refused connection is an error that still says why" { // CONNACK with reason code 0x87, Not authorized. var loop: Loopback = .init(&.{ 0x20, 0x03, 0x00, 0x87, 0x00 }); defer loop.deinit();
var conn: Connection = .init(testing.allocator, .{}); defer conn.deinit(); var client: Client = .init(&conn, testing.io, &loop.to_client, &loop.from_client.writer);
try testing.expectError(error.ConnectionRefused, client.connect(.{ .client_id = "test" })); try testing.expectEqual(ReasonCode.not_authorized, client.refusal.?.reason_code); try testing.expectEqual(Connection.Disconnected.Origin.connack, client.refusal.?.origin);}
test "messages are read out of the transport one event at a time" { // A CONNACK, then two QoS 0 publishes, then a DISCONNECT. const broker = connack_bytes ++ [_]u8{ 0x30, 0x09, 0x00, 0x03, 'a', '/', 'b', 0x00, 'o', 'n', 'e', // publish "one" 0x30, 0x09, 0x00, 0x03, 'a', '/', 'b', 0x00, 't', 'w', 'o', // publish "two" 0xe0, 0x00, // disconnect }; var loop: Loopback = .init(&broker); defer loop.deinit();
var conn: Connection = .init(testing.allocator, .{}); defer conn.deinit(); var client: Client = .init(&conn, testing.io, &loop.to_client, &loop.from_client.writer);
_ = try client.connect(.{ .client_id = "test", .keep_alive = 0 }); try testing.expectEqualStrings("one", (try client.next()).?.message.payload); try testing.expectEqualStrings("two", (try client.next()).?.message.payload); try testing.expectEqual( ReasonCode.success, (try client.next()).?.disconnected.reason_code, ); // The stream ends cleanly between packets, which is not an error. try testing.expectEqual(@as(?Event, null), try client.next());}
test "a transport that ends in the middle of a packet is an error" { // The CONNACK, then a PUBLISH whose header promises ten bytes and delivers // three. const broker = connack_bytes ++ [_]u8{ 0x30, 0x0a, 0x00, 0x03 }; var loop: Loopback = .init(&broker); defer loop.deinit();
var conn: Connection = .init(testing.allocator, .{}); defer conn.deinit(); var client: Client = .init(&conn, testing.io, &loop.to_client, &loop.from_client.writer);
_ = try client.connect(.{ .client_id = "test", .keep_alive = 0 }); try testing.expectError(error.UnexpectedEndOfStream, client.next());}
test "an acknowledgement is sent before the message it acknowledges is handed over" { // A QoS 1 message and a QoS 0 one in the same read. Draining both without // flushing would leave the first one's PUBACK in the buffer while the // caller is told the message arrived -- and a broker with a small Receive // Maximum would then stop sending. const broker = connack_bytes ++ [_]u8{ 0x32, 0x0b, 0x00, 0x03, 'a', '/', 'b', 0x00, 0x2a, 0x00, 'o', 'n', 'e', 0x30, 0x09, 0x00, 0x03, 'a', '/', 'b', 0x00, 't', 'w', 'o', }; var loop: Loopback = .init(&broker); defer loop.deinit();
var conn: Connection = .init(testing.allocator, .{}); defer conn.deinit(); var client: Client = .init(&conn, testing.io, &loop.to_client, &loop.from_client.writer);
_ = try client.connect(.{ .client_id = "test", .keep_alive = 0 }); const before = loop.from_client.written().len;
try testing.expectEqualStrings("one", (try client.next()).?.message.payload); const after = loop.from_client.written(); try testing.expect(after.len > before);
// And what went out is the PUBACK for packet identifier 42. const header = try packet.peekFixedHeader(after[before..]); const ack = try packet.Packet.decode( header, after[before..][header.encodedLen()..header.packetLen()], .v5_0, ); try testing.expectEqual(@as(mqtt.protocol.types.PacketId, 42), ack.puback.packet_id);
try testing.expectEqualStrings("two", (try client.next()).?.message.payload);}
test "publishing and subscribing reach the transport immediately" { var loop: Loopback = .init(&connack_bytes); defer loop.deinit();
var conn: Connection = .init(testing.allocator, .{}); defer conn.deinit(); var client: Client = .init(&conn, testing.io, &loop.to_client, &loop.from_client.writer);
_ = try client.connect(.{ .client_id = "test", .keep_alive = 0 }); const before = loop.from_client.written().len;
_ = try client.publish(.{ .topic = "a/b", .payload = "x" }); const after_publish = loop.from_client.written().len; try testing.expect(after_publish > before);
const sub_id = try client.subscribe(&.{.{ .filter = "c/#" }}, .empty); try testing.expect(loop.from_client.written().len > after_publish); try testing.expect(sub_id != 0);}
/// A reader that hands over one chunk per call and then says the stream ended.////// `Reader.fixed` will not do for the keepalive test: it holds every byte from/// the start, so the client never has to wait for anything and the waiting path/// is never taken. What is needed is a stream where the second packet has not/// arrived yet, which is what every real one looks like.const ChunkedReader = struct { chunks: []const []const u8, index: usize = 0, interface: std.Io.Reader, buffer: [512]u8 = undefined,
fn init(chunks: []const []const u8) ChunkedReader { return .{ .chunks = chunks, .interface = .{ .vtable = &.{ .stream = streamImpl }, .buffer = &.{}, .seek = 0, .end = 0, }, }; }
/// Point the interface at this struct's own buffer. Done after the struct /// has stopped moving, because the interface holds a slice of it. fn attach(r: *ChunkedReader) *std.Io.Reader { r.interface.buffer = &r.buffer; return &r.interface; }
fn streamImpl( io_r: *std.Io.Reader, w: *std.Io.Writer, limit: std.Io.Limit, ) std.Io.Reader.StreamError!usize { const r: *ChunkedReader = @alignCast(@fieldParentPtr("interface", io_r)); if (r.index >= r.chunks.len) return error.EndOfStream; const chunk = r.chunks[r.index]; r.index += 1; const dest = limit.slice(try w.writableSliceGreedy(1)); const n = @min(dest.len, chunk.len); @memcpy(dest[0..n], chunk[0..n]); w.advance(n); return n; }};
/// A waiter that reports nothing to read until it has done so once, sleeping/// for as long as it was asked to wait.////// That is what a real transport does when the broker is silent, and it is the/// shape that used to break keepalive: the client blocked in the read, the/// PINGREQ that fell due never went out, and the broker closed the connection/// at a keep-alive and a half.const IdleWaiter = struct { timeouts: usize = 0,
fn readable(context: *anyopaque, io: std.Io, timeout: std.Io.Timeout) anyerror!void { const w: *IdleWaiter = @ptrCast(@alignCast(context)); // `.none` means the connection has no timer running, which during the // handshake it does not: it is waiting for the CONNACK and nothing // else. A real waiter blocks; this one lets the reply through. if (timeout == .none) return; if (w.timeouts >= 1) return; w.timeouts += 1; timeout.sleep(io) catch {}; return error.Timeout; }
fn hooks(w: *IdleWaiter) Transport.Hooks { return .{ .context = w, .readable = readable }; }};
test "an idle connection pings rather than waiting to be disconnected" { // The broker says nothing after its CONNACK until it answers the ping, so // a client that never pinged would sit in the read until the transport gave // up -- which is how a real broker disconnected this client at a keep alive // and a half, and what the waiter is for. A keep alive of one second, so // the test costs about that. var reader: ChunkedReader = .init(&.{ &connack_bytes, &.{ 0xd0, 0x00 } }); var out: std.Io.Writer.Allocating = .init(testing.allocator); defer out.deinit();
var conn: Connection = .init(testing.allocator, .{}); defer conn.deinit();
var idle: IdleWaiter = .{}; var client: Client = .{ .conn = &conn, .io = testing.io, .reader = reader.attach(), .writer = &out.writer, .hooks = idle.hooks(), };
_ = try client.connect(.{ .client_id = "idle", .keep_alive = 1 }); const before = out.written().len;
// The waiter reports nothing to read, which is where the ping has to // happen, and then lets the PINGRESP through. try testing.expectEqual(@as(Event, .pong), (try client.next()).?); try testing.expectEqual(@as(usize, 1), idle.timeouts);
// And a PINGREQ really did go out in between. const after = out.written()[before..]; const header = try packet.peekFixedHeader(after); try testing.expectEqual(mqtt.protocol.types.PacketType.pingreq, header.packet_type);}
/// A waiter that never has anything to read: a broker that is connected and/// silent, which is the case a deadline exists for.const SilentWaiter = struct { timeouts: usize = 0, /// How many times to report nothing before letting the transport through. /// Null is a broker that never speaks again; zero is one with something to /// say already, so that a test can tell "did not wait" apart from "waited /// and heard nothing". limit: ?usize = null,
fn readable(context: *anyopaque, io: std.Io, timeout: std.Io.Timeout) anyerror!void { const w: *SilentWaiter = @ptrCast(@alignCast(context)); // The handshake waits on nothing in particular; let the CONNACK // through rather than deadlocking the connection. if (timeout == .none) return; if (w.limit) |limit| if (w.timeouts >= limit) return; w.timeouts += 1; timeout.sleep(io) catch {}; return error.Timeout; }
fn hooks(w: *SilentWaiter) Transport.Hooks { return .{ .context = w, .readable = readable }; }};
test "a deadline hands the task back rather than waiting for a silent broker" { var reader: ChunkedReader = .init(&.{&connack_bytes}); var out: std.Io.Writer.Allocating = .init(testing.allocator); defer out.deinit();
var conn: Connection = .init(testing.allocator, .{}); defer conn.deinit();
var silent: SilentWaiter = .{}; var client: Client = .{ .conn = &conn, .io = testing.io, .reader = reader.attach(), .writer = &out.writer, .hooks = silent.hooks(), };
// A keep alive far longer than the deadline, so that what ends the wait // can only be the deadline. _ = try client.connect(.{ .client_id = "deadline", .keep_alive = 600 });
const waited = try client.nextBefore(.{ .duration = .{ .raw = .fromMilliseconds(30), .clock = .boot, } }); try testing.expect(waited == .timeout); try testing.expect(silent.timeouts >= 1);
// And the connection is untouched: asking again is not an error, and the // client still believes it is connected. const again = try client.nextBefore(.{ .duration = .{ .raw = .fromMilliseconds(30), .clock = .boot, } }); try testing.expect(again == .timeout);}
test "a deadline already past is a poll rather than a wait" { var reader: ChunkedReader = .init(&.{ &connack_bytes, &.{ 0xd0, 0x00 } }); var out: std.Io.Writer.Allocating = .init(testing.allocator); defer out.deinit();
var conn: Connection = .init(testing.allocator, .{}); defer conn.deinit();
// Nothing to wait for: whatever the transport has is there to be read, so // a poll that consults the waiter at all would show up as a timeout here. var silent: SilentWaiter = .{ .limit = 0 }; var client: Client = .{ .conn = &conn, .io = testing.io, .reader = reader.attach(), .writer = &out.writer, .hooks = silent.hooks(), };
_ = try client.connect(.{ .client_id = "poll", .keep_alive = 600 }); try client.ping();
const zero: std.Io.Timeout = .{ .duration = .{ .raw = .fromMilliseconds(0), .clock = .boot } };
// Zero milliseconds is "do not wait", and the PINGRESP is one read away // rather than in hand -- so this answers without ever asking the waiter to // sleep, which is what makes such a poll cheap enough to do in a loop. try testing.expect((try client.nextBefore(zero)) == .timeout); try testing.expectEqual(@as(usize, 0), silent.timeouts);
// The reply is still there, and a deadline with room in it finds it. The // poll consumed nothing and lost nothing. try testing.expectEqual(@as(Event, .pong), (try client.nextBefore(.{ .duration = .{ .raw = .fromMilliseconds(500), .clock = .boot, } })).event);}
test "a deadline does not suppress the keepalive it outlives" { // A keep alive of one second inside a deadline of about one and a half, so // the ping falls due while the caller is still waiting. The test costs // roughly that. var reader: ChunkedReader = .init(&.{&connack_bytes}); var out: std.Io.Writer.Allocating = .init(testing.allocator); defer out.deinit();
var conn: Connection = .init(testing.allocator, .{}); defer conn.deinit();
var silent: SilentWaiter = .{}; var client: Client = .{ .conn = &conn, .io = testing.io, .reader = reader.attach(), .writer = &out.writer, .hooks = silent.hooks(), };
_ = try client.connect(.{ .client_id = "pinger", .keep_alive = 1 }); const before = out.written().len;
try testing.expect((try client.nextBefore(.{ .duration = .{ .raw = .fromMilliseconds(1500), .clock = .boot, } })) == .timeout);
// The wait was the earlier of the two deadlines each time round, so the // PINGREQ went out at a second and the caller was answered at a second and // a half. const after = out.written()[before..]; const header = try packet.peekFixedHeader(after); try testing.expectEqual(mqtt.protocol.types.PacketType.pingreq, header.packet_type);}
test "without the transport hooks the client still works, it just cannot ping" { // The plain constructor is not broken, only limited -- everything that does // not depend on the clock behaves identically. var loop: Loopback = .init(&(connack_bytes ++ [_]u8{ 0xd0, 0x00 })); defer loop.deinit();
var conn: Connection = .init(testing.allocator, .{}); defer conn.deinit(); var client: Client = .init(&conn, testing.io, &loop.to_client, &loop.from_client.writer);
_ = try client.connect(.{ .client_id = "plain", .keep_alive = 0 }); try client.ping(); try testing.expectEqual(@as(Event, .pong), (try client.next()).?); try testing.expect(client.hooks.readable == null); try testing.expect(client.hooks.flush == null);}