GreyCat MQTT Library

@library("mqtt", "0.0.0");

Publish to and subscribe from an MQTT broker. Speaks MQTT 3.1, 3.1.1 and 5.0, on top of the Eclipse Paho async C client, with TLS through the bundled OpenSSL.

Quick Start

Publish

var client = MqttClient { uri: "tcp://localhost:1883" };

// Blocks until the broker acknowledges.
client.publish("sensors/room1", "21.5", MqttQos::at_least_once, null, null);

// Returns immediately with a delivery token.
var token = client.publish_async("sensors/room1", "21.6", MqttQos::at_least_once, null, null);
client.wait(token, 10s);

fn on_reading(msg: MqttMessage) {
    info("${msg.topic} -> ${msg.payload}");
}

var client = MqttClient { uri: "tcp://localhost:1883" };
client.subscribe("sensors/#", MqttQos::at_least_once, on_reading);

Each incoming message spawns a GreyCat task that calls the handler, so it runs with its own transaction and appears in Task::history() under its own name.

There is no connect(). The client connects on its first publish or subscribe, and disconnects when the object is collected – unless a subscription is still registered, which holds the connection open so messages keep arriving. unsubscribe cancels the subscription and releases it; disconnect releases the connection and leaves the subscription with the broker.

A client belongs to one task

MqttClient is @volatile: it owns a socket, a TLS session and a subscription registry, none of which can be written to the graph. Every @exposed endpoint and every subscription callback is its own task, and there is no process-lifetime slot to put a client in, so a client is reachable only from the task that built it.

That shapes an application in three ways:

stats(), is_connected() and unsubscribe() answer for the connection the calling task owns. A monitoring endpoint is a different task, so it cannot see the ingest client’s counters. Derive health from the ingested data instead: the newest timestamp, the silence since it, and the gaps in whatever sequence the publisher carries.
Subscriptions are established once at boot and held for the life of the process. Retiring one means unsubscribe from the task that created it.
A publish-only client is built per task. Leave client_id null so concurrent tasks do not present the same identifier to the broker, which would make it drop all but the newest.

Within one task, hold the client rather than rebuilding it per call. Against a broker on loopback, the handshake costs about 0.5ms and a QoS 1 publish about 0.12ms, so a fresh client per publish is roughly five times the work.

What the QoS levels mean here

Between the publisher and the broker they mean what the spec says: publish blocks until the broker has acknowledged, and throws if it does not.

Between the broker and a subscription they mean less than they appear to. The broker’s PUBACK goes out when the message is taken off the wire, before the handler task has run, and the library never learns whether that task committed. A handler that throws, or whose transaction fails to merge with a concurrent one, loses its message with no redelivery and no counter movement.

So on the receiving side, treat at_least_once as “the broker will retry until this process has the bytes”, not “the handler will see it”. An application that cannot lose messages needs its own sequence number or checksum in the payload, and a gap count on ingest. Nothing the library exposes can tell you instead.

Payloads are bytes

MQTT carries arbitrary bytes, so MqttMessage.payload is a byte buffer rather than text. It may hold NUL or invalid UTF-8, and interpolating it into a string is only meaningful when the publisher sends text.

fn on_reading(msg: MqttMessage) {
    var reading = Json<Reading> {}.parse(msg.payload);
    info("${reading.celsius}");
}

payload.get(i) returns a char, which is signed: a payload byte of 0xff reads back as -1, not 255. Add 256 to a negative value to get the unsigned byte, which matters for any binary protocol.

Never return a raw payload from an @exposed function. GreyCat’s JSON writer passes control bytes through unescaped, so a payload holding a NUL produces a response body that standard JSON parsers reject. Encode it first, with Crypto::base64_encode or Crypto::hex_encode.

On the way out, a String payload is sent as-is and anything else is serialized to JSON first.

Security

var client = MqttClient {
    uri: "ssl://broker.example.com:8883",
    options: MqttOptions {
        username: "sensor",
        password: "...",
        tls: MqttTls { ca_path: "/etc/ssl/certs/ca.pem" },
    },
};

MQTT 5

The default is MQTT 3.1.1, which every broker accepts. Opt in to 5.0 for message properties, richer subscribe options and reason codes:

var client = MqttClient {
    uri: "tcp://localhost:1883",
    options: MqttOptions { version: MqttVersion::v5 },
};

client.publish("cmd/req", "{}", MqttQos::at_least_once, null,
    MqttProperties { content_type: "application/json", response_topic: "cmd/reply" });

MqttProperties is ignored under 3.1 and 3.1.1, and MqttMessage.properties is always null there.

Client identifiers

client_id defaults to greycat plus 16 hex digits: 23 bytes of [0-9a-zA-Z], the length and alphabet every broker is required to accept. A generated id is new on every connect, so it cannot resume a persistent session – set one explicitly for that.

Two live connections presenting the same client_id to one broker is a protocol-level conflict: the broker disconnects the older session, silently taking its subscriptions with it. The library logs a warning when it sees this.

Persistent sessions

clean_session: false with an explicit client_id asks the broker to keep the subscriptions and queue QoS 1 and 2 messages while the client is away. Park the session with disconnect, not unsubscribe: unsubscribe cancels the subscription on the broker, which is the thing the session exists to preserve.

var options = MqttOptions { client_id: "ingest-1", clean_session: false };
var c = MqttClient { uri: "tcp://localhost:1883", options: options };
c.subscribe("sensors/#", MqttQos::at_least_once, on_reading);
c.disconnect();                                  // the broker keeps queueing

var resumed = MqttClient { uri: "tcp://localhost:1883", options: options };
resumed.subscribe("sensors/#", MqttQos::at_least_once, on_reading);

Under MQTT 5, session_expiry bounds how long the broker holds it.

Wildcards and overlapping filters

+ matches one level, # matches the rest of the tree and is only legal as a whole last level. sport/# also matches sport itself. A filter the spec forbids is rejected at subscribe rather than sent, because a broker answers a malformed filter by closing the connection.

Keep the filters on one client disjoint. A broker sends one PUBLISH per matching subscription, and the library then matches each copy against every registered filter, so N overlapping filters cost N*N handler tasks for one published message. “One broad filter for logging plus narrow ones for handling” is the shape that bites. The library logs a warning when it notices an overlap.

Backpressure

A subscription that outruns the task pool drops messages rather than blocking the network thread, which would stall the connection and eventually trip the keepalive. stats().dropped counts them and a throttled warning is logged. A non-zero dropped means the pool is behind, not that the broker misbehaved.

publish_async queues into a send buffer bounded by max_buffered, 100 by default, and throws once it is full. A loop that publishes faster than the broker acknowledges will reach it, so wrap the loop and retry:

var token = 0;
for (_, reading in readings) {
    try {
        token = client.publish_async("sensors/room1", reading, MqttQos::at_least_once, null, null);
    } catch {
        // Let the in-flight window drain against the last accepted message,
        // then retry this one.
        client.wait(token, 5s);
        token = client.publish_async("sensors/room1", reading, MqttQos::at_least_once, null, null);
    }
}

Monitoring

var stats = client.stats();
info("received=${stats.received} dispatched=${stats.dispatched} dropped=${stats.dropped}");

The counters run from the moment the connection was established and are never reset, including by an automatic reconnect. They are readable only from the task that owns the client.

This library’s warnings (dropped messages, a lost connection, a duplicate client_id, an overlapping filter) go to greycat’s log at files/root/log.csv. They are not written to stdout when stdout is redirected, so a containerised server shows nothing in docker logs.

Testing

script/test.sh runs the unit tests against a mosquitto container.

test/integration.sh covers what those cannot: greycat queues the tasks a subscription spawns but only drains them under serve, so a callback body never runs under greycat test. That script starts a real server and asserts on what the callbacks wrote back.