#mqtt API reference

An MQTT 3.1.1 publish/subscribe client over the net system library - the same "protocol clients are modules, net is the transport" line the other network clients follow. MQTT packets are a 1-byte fixed header, a variable remaining-length integer, then a length-prefixed payload, all built and parsed here with Jennifer's bitwise operators and bytes. QoS 0 and QoS 1 publish / subscribe (a synchronous PUBACK handshake), retained messages, a Last-Will set in the CONNECT, and reconnect for session resumption, on top of a single-threaded poll with timeout (via net.setDeadline) so one flow can wait for a packet and send keepalives without a spawned reader, plus blocking receive, ping, and disconnect. QoS 2 and MQTT 5 properties are out of scope. Needs the default jennifer binary (uses net).

Import with import "mqtt.j" as mqtt;. See the mqtt guide for prose and examples.

#Functions

#mqtt.connect(opts as Options)

Open a connection, send CONNECT, and check the CONNACK return code. Uses a clean session and no Last-Will; for a will or session resumption use connectWith.

Parameters

• opts {Options} - the connection settings

Returns {Client} - the open client

Throws

• {Error} - kind "mqtt" when the broker refuses the connection

#mqtt.connectWith(opts as Options, will as Will, cleanSession as bool)

Open a connection with an explicit Last-Will and clean-session flag. A will whose topic is "" registers no will; cleanSession false asks the broker to resume a persistent session keyed by opts.clientId (retained subscriptions and queued QoS-1 messages).

Parameters

• opts {Options} - the connection settings
• will {Will} - the Last-Will to register ("" topic disables it)
• cleanSession {bool} - false to resume a persistent session

Returns {Client} - the open client

Throws

• {Error} - kind "mqtt" when the broker refuses the connection

#mqtt.disconnect(client as Client)

Send DISCONNECT and close the connection.

Parameters

• client {Client} - the open client

#mqtt.ping(client as Client)

Send a PINGREQ keepalive (fire and forget). The matching PINGRESP is consumed by the next poll / receive.

Parameters

• client {Client} - the open client

#mqtt.poll(client as Client, timeoutMs as int)

Poll for a message, waiting up to timeoutMs milliseconds. Returns a list of zero or one Message: empty when nothing arrived in the window (the caller can then ping and loop), one Message when a PUBLISH was received. A QoS-1 PUBLISH is acknowledged with a PUBACK before it is returned. Non-PUBLISH control packets are consumed and reported as an empty poll.

Parameters

• client {Client} - the open client
• timeoutMs {int} - how long to wait for the next packet, in milliseconds

Returns {list of Message} - zero or one received message

#mqtt.publish(client as Client, topic as string, message as string)

Publish a text message to a topic at QoS 0 (UTF-8 encoded).

Parameters

• client {Client} - the open client
• topic {string} - the topic to publish to
• message {string} - the message text

#mqtt.publishBytes(client as Client, topic as string, payload as bytes)

Publish a raw byte payload to a topic at QoS 0 (fire and forget).

Parameters

• client {Client} - the open client
• topic {string} - the topic to publish to
• payload {bytes} - the message bytes

#mqtt.publishBytesRetain(client as Client, topic as string, payload as bytes, retain as bool)

Publish a raw byte payload at QoS 0 with an explicit retain flag. A retained message is stored by the broker and delivered to every future subscriber of the topic; publishing an empty payload with retain true clears it.

Parameters

• client {Client} - the open client
• topic {string} - the topic to publish to
• payload {bytes} - the message bytes
• retain {bool} - whether the broker retains the message

#mqtt.publishQos1(client as Client, topic as string, payload as bytes, retain as bool)

Publish a raw byte payload at QoS 1 (at-least-once) and block until the matching PUBACK. The PUBLISH carries a packet identifier; an unacknowledged send is retried as a DUP (the duplicate-delivery flag set) up to a fixed number of attempts, so at-least-once delivery is honored across a slow or flapping link. Because the call blocks for the PUBACK, only one message is ever in flight, so a fixed non-zero packet identifier is safe. A duplicate the broker already delivered is de-duplicated by the packet id on its side.

Parameters

• client {Client} - the open client
• topic {string} - the topic to publish to
• payload {bytes} - the message bytes
• retain {bool} - whether the broker retains the message

Throws

• {Error} - kind "mqtt" when no PUBACK arrives within the retry budget

#mqtt.publishRetain(client as Client, topic as string, message as string, retain as bool)

Publish a text message at QoS 0 with an explicit retain flag (UTF-8 encoded).

Parameters

• client {Client} - the open client
• topic {string} - the topic to publish to
• message {string} - the message text
• retain {bool} - whether the broker retains the message

#mqtt.receive(client as Client)

Block until the next application message arrives, returning it. A QoS-1 PUBLISH is acknowledged with a PUBACK before it is returned. Non-PUBLISH control packets (e.g. a PINGRESP) are consumed and skipped.

Parameters

• client {Client} - the open client

Returns {Message} - the next received message

#mqtt.reconnect(client as Client)

Re-dial the broker and re-CONNECT with this client's stored settings, then re-subscribe every tracked subscription. Returns a fresh Client the caller reassigns (value semantics - the old handle is best-effort closed).

Session resumption: when the client connected with cleanSession false, the broker keeps the session (subscriptions and queued QoS-1 messages) across the drop and reports it in the CONNACK; the re-subscribe here is idempotent and also covers the clean-session case, where the broker discarded the session.

Parameters

• client {Client} - the (typically disconnected) client to re-establish

Returns {Client} - a freshly connected client with the tracked routes restored

Throws

• {Error} - kind "mqtt" when the re-connect or a re-subscribe fails

#mqtt.subscribe(client as Client, topic as string)

Subscribe to a topic filter at QoS 0, wait for the SUBACK, and track the subscription on the returned Client (so reconnect can restore it). Reassign the result: $c = mqtt.subscribe($c, "topic");.

Parameters

• client {Client} - the open client
• topic {string} - the topic filter (may contain + / # wildcards)

Returns {Client} - the client with the subscription tracked

Throws

• {Error} - kind "mqtt" when the broker rejects the subscription

#mqtt.subscribeQos1(client as Client, topic as string)

Subscribe to a topic filter requesting QoS 1, wait for the SUBACK, and track the subscription (with the QoS the broker granted) on the returned Client. With QoS 1 the broker delivers each matching message as a QoS-1 PUBLISH, which receive / poll acknowledge with a PUBACK. Reassign the result: $c = mqtt.subscribeQos1($c, "topic");.

Parameters

• client {Client} - the open client
• topic {string} - the topic filter (may contain + / # wildcards)

Returns {Client} - the client with the subscription tracked

Throws

• {Error} - kind "mqtt" when the broker rejects the subscription

#Structs

#mqtt.Client

An open MQTT connection. Beyond the socket it carries the settings needed to re-establish the session (opts, will, cleanSession) and the list of subscriptions subscribe / subscribeQos1 have tracked, so reconnect can re-dial and re-subscribe. Value-semantic: copies share the socket handle but each carries its own settings and subscription list.

#mqtt.Message

One received application message.

#mqtt.Options

Connection settings for mqtt.connect.

#mqtt.Subscription

One tracked subscription, remembered on the Client so reconnect can restore the session's routes.

#mqtt.Will

A Last-Will-and-Testament message the broker publishes on this client's behalf if the connection drops without a clean DISCONNECT. An empty topic means "no will" (the CONNECT sets no will flags).