MQTT module

The MQTT module publishes and receives messages over MQTT, the messaging protocol of many connected devices, through a server that hands them out (the broker). A program publishes messages on a topic, such as home/livingroom/temperature, and receives those of the topics it subscribes to with a callback.

The connection to the broker isn't opened from the program: you set it up outside it, in the board profile or, on Windows, in the project settings (see Platforms).

use MQTT

Platforms

	Windows	ESP32 / ESP32-S3	ESP32-P4	ESP32-S31
The connection is set up in	Config › Platform › Windows, in the project (Settings)	The board profile, in ESP32 Tools (Display, sound, SD and network)	Same	Same
When it connects	At startup: it waits for the broker 3 seconds at most and, if it doesn't connect, it starts the program and keeps trying	On its own, while the program is already running	Same	Same
If the connection is lost	It reconnects by itself	It reconnects by itself, also if the WiFi drops	It reconnects by itself	It reconnects by itself
Broker user and password	The ones in the settings	Same	Same	Same
QoS of publish and subscribe	0	0	The one in the settings	The one in the settings
Retain on publish	No	The one in the settings	Same	Same
intStats() and resetIntStats()	Yes	Don't exist: they give ERR_ATTRIBUTE_NOT_FOUND	Same	Same

Functions

Function	What it does	Returns
MQTT.publish(topic, message)	Publishes a message on a topic	true
MQTT.subscribe(topic)	Subscribes to a topic	true
MQTT.unsubscribe(topic)	Stops being subscribed to a topic	true
MQTT.callback(function)	Says which function receives the messages	true
MQTT.isConnect()	Checks whether there's a connection to the broker	true or false
MQTT.pubinfo()	Publishes the VM's version	true, or false without a connection
MQTT.intStats()	Windows only: how the VM's event queue is doing	A map
MQTT.resetIntStats()	Windows only: sets those counts to zero	true

publish, subscribe and unsubscribe do nothing if there's no connection to the broker, and return true all the same: their true confirms nothing. To know whether there's a connection, use isConnect().

If you pass the wrong number of arguments, they return an error value, which try doesn't catch (see Errors and exceptions). The exception is intStats, which throws ERR_INVALID_ARG_NUMBER.

If the topic, or the message of publish, isn't a string, they throw ERR_TYPE_MISMATCH, which try does catch.

publish(topic, message)

Publishes message on the topic topic. Both have to be strings.

use MQTT

var temperature = 25.5
MQTT.publish("sensors/temperature", temperature.toString())

// A message in JSON
MQTT.publish("device/status", '{"active": true, "battery": 85}')

If the message is a number, it throws ERR_TYPE_MISMATCH: convert it to a string first with toString(), as with temperature in the example.

subscribe(topic)

Subscribes to the topic topic, a string: from then on, the messages published on it reach the callback. It accepts MQTT wildcards: + stands for one level of the topic and #, at the end, for all the ones that follow.

use MQTT

MQTT.subscribe("home/livingroom/light")
MQTT.subscribe("commands/#")       // every topic that starts with commands/

Without a connection it does nothing, and with a clean session, which is the default, the subscription is lost if the connection is lost: subscribe when isConnect() gives true, and again after every reconnection, as in the complete example.

unsubscribe(topic)

Stops being subscribed to the topic topic. It has to be the same text you subscribed with.

use MQTT

MQTT.subscribe("sensors/temperature")
// … later on
MQTT.unsubscribe("sensors/temperature")

callback(function)

Says which function receives the messages of every topic you're subscribed to. There's a single callback: if you call it again, the new one replaces the old one.

function can be a function (onMessage), a method of an object (station.onMessage) or the name of a global function in a string ("onMessage"). It has to take exactly two parameters, the topic and the message, both strings: if it has another number, the messages are dropped without warning. If function is none of those, callback returns an error value.

use MQTT

func onMessage(topic, message)
    println(topic, ": ", message)
end

MQTT.callback(onMessage)
MQTT.subscribe("sensors/#")

With a method, the callback runs with its this, and MQTT keeps the instance alive while it's registered:

use MQTT

class Station
    def init(name)
        this.name = name
        this.messages = 0
    end

    def onMessage(topic, message)
        this.messages = this.messages + 1
        println(this.name, " receives: ", message)
    end
end

var weather = Station("Weather")
MQTT.callback(weather.onMessage)
MQTT.subscribe("sensors/#")

When the callback runs

Messages arrive on their own and wait in a queue in the VM. The callback gets them one at a time, inside your program, like the Timer ones: at the VM's safe points (when a loop goes round and when calling or returning from a function), never inside pause(). That's why a program that waits for messages goes round with short pauses, and what arrives during a pause(10000) isn't handled until it ends. When the program ends, the callback is released.

isConnect()

Returns true if there's a connection to the broker, and false if not.

use MQTT

if (MQTT.isConnect()) then
    MQTT.publish("sensors/status", "active")
else
    println("No MQTT connection")
end

pubinfo()

Publishes the VM's version on the device topic, <user>/<id>/device, with the user and the identifier in the MQTT settings. With the user empty, which is the default, the topic starts with the slash: /<id>/device. The message is like {"HWVersion":"0.9.7.20261104 01"}. It returns true if it published it and false if there's no connection.

use MQTT

if (MQTT.pubinfo()) then
    println("Version published")
end

intStats()

Windows only. Returns a map with the counts of the VM's event queue, where MQTT messages wait their turn, and also GPIO interrupts, timer firings and sprite collisions:

Key	What it counts
enqueued	The events that have entered the queue
dequeued	The ones the VM has taken out to handle them
dropped	The ones lost because the queue was full
hasPending	true if there's one waiting right now

The counts start with the program, or at the last resetIntStats(). The queue holds 127 events: if dropped grows, the program can't keep up, and it gets more messages or interrupts than it handles.

It doesn't exist on the boards: calling it throws ERR_ATTRIBUTE_NOT_FOUND (Module has no function 'intStats'.). In a program that also goes to a board, check first with platform():

use MQTT

if (platform() == HW_WIN32) then
    println(MQTT.intStats())   // {hasPending: false, enqueued: 0, dequeued: 0, dropped: 0}
end

resetIntStats()

Windows only. Sets the counts of intStats() to zero and returns true. It doesn't exist on the boards, like intStats.

Complete example

A device that turns an LED on and off with the messages of the home/led topic, confirms each change and publishes a heartbeat every 10 seconds. It subscribes when it connects, and again if the connection is lost:

use MQTT, GPIO

const LED_PIN = 2
GPIO.pinmode(LED_PIN, GPIO.OUTPUT)

// A single callback for every topic
func onMessage(topic, message)
    println("[MQTT] ", topic, ": ", message)
    if (topic == "home/led") then
        if (message == "on") then
            GPIO.write(LED_PIN, HIGH)
        elsif (message == "off") then
            GPIO.write(LED_PIN, LOW)
        end
        MQTT.publish("home/led/status", message)   // confirms the change
    end
end

MQTT.callback(onMessage)

var subscribed = false
var lastBeat = 0
while (true)
    if (MQTT.isConnect()) then
        if (!subscribed) then
            MQTT.subscribe("home/led")   // when it connects, and again after losing the connection
            subscribed = true
        end
        if (clock() - lastBeat > 10000) then
            MQTT.publish("home/heartbeat", "alive")
            lastBeat = clock()
        end
    else
        subscribed = false
    end
    pause(50)   // short pauses: messages are handled between one and the next
end

Pitfalls and known issues

• A true that confirms nothing. Without a connection, publish, subscribe and unsubscribe do nothing and return true. Check isConnect().
• The program can start before the connection. On a board, always; on Windows, if the broker doesn't answer within 3 seconds. A subscribe at the start of the program is lost if the broker hasn't answered yet. Subscribe when isConnect() gives true.
• Subscriptions don't survive a reconnection. With the default clean session, when the connection comes back you have to subscribe again.
• Long pauses, waiting messages. While the program is in pause(), messages aren't handled: they wait in the queue, and if it fills up, they're lost.
• The callback, with two parameters. With any other number of parameters, messages are dropped without warning.
• One Client ID for each VM or board. Two with the same Client ID on the same broker keep kicking each other out. On Windows, every project comes with JARUWin11_01 by default: if you connect two at once, or use a public broker like test.mosquitto.org, give each project its own in Config › Platform › Windows.

All the known issues of JARU, together, are in Known issues.