Módulo MQTT

El módulo MQTT publica y recibe mensajes por MQTT, el protocolo de mensajería de muchos aparatos conectados, a través de un servidor que los reparte (el broker). Un programa publica mensajes en un tema, como casa/salon/temperatura, y recibe los de los temas a los que se suscribe con un callback.

La conexión con el broker no se abre desde el programa: la configuras fuera, en el perfil de la placa o, en Windows, en la configuración del proyecto (ver Plataformas).

use MQTT

Captura del módulo MQTT de JARU para conectar aplicaciones ESP32 a un broker y publicar mensajes IoT

Plataformas

	Windows	ESP32 / ESP32-S3	ESP32-P4	ESP32-S31
La conexión se configura en	Config › Platform › Windows, en el proyecto (Ajustes)	El perfil de la placa, en ESP32 Tools (Pantalla, sonido, SD y red)	Igual	Igual
Cuándo conecta	Al arrancar: espera al broker 3 segundos como mucho y, si no conecta, empieza el programa y lo sigue intentando	Por su cuenta, mientras el programa ya corre	Igual	Igual
Si se pierde la conexión	Vuelve a conectar sola	Vuelve a conectar sola, también si se cae la WiFi	Vuelve a conectar sola	Vuelve a conectar sola
Usuario y contraseña del broker	Los de la configuración	Igual	Igual	Igual
QoS de `publish` y `subscribe`	0	0	La de la configuración	La de la configuración
Retain en `publish`	No	El de la configuración	Igual	Igual
`intStats()` y `resetIntStats()`	Sí	No existen: dan `ERR_ATTRIBUTE_NOT_FOUND`	Igual	Igual

Funciones

Función	Qué hace	Devuelve
`MQTT.publish(tema, mensaje)`	Publica un mensaje en un tema	`true`
`MQTT.subscribe(tema)`	Se suscribe a un tema	`true`
`MQTT.unsubscribe(tema)`	Deja de estar suscrito a un tema	`true`
`MQTT.callback(funcion)`	Dice qué función recibe los mensajes	`true`
`MQTT.isConnect()`	Mira si hay conexión con el broker	`true` o `false`
`MQTT.pubinfo()`	Publica la versión de la VM	`true`, o `false` sin conexión
`MQTT.intStats()`	Solo en Windows: cómo va la cola de avisos de la VM	Un mapa
`MQTT.resetIntStats()`	Solo en Windows: pone a cero esas cuentas	`true`

publish, subscribe y unsubscribe no hacen nada si no hay conexión con el broker, y devuelven true igualmente: su true no confirma nada. Para saber si hay conexión, usa isConnect().

Si les pasas un número de argumentos que no es el que piden, devuelven un valor de error, que try no captura (ver Errores y excepciones). La excepción es intStats, que lanza ERR_INVALID_ARG_NUMBER.

Si el tema, o el mensaje de publish, no es una cadena, lanzan ERR_TYPE_MISMATCH, que try sí captura.

publish(tema, mensaje)

Publica mensaje en el tema tema. Los dos tienen que ser cadenas.

use MQTT

var temperatura = 25.5
MQTT.publish("sensores/temperatura", temperatura.toString())

// Un mensaje en JSON
MQTT.publish("aparato/estado", '{"activo": true, "bateria": 85}')

Si el mensaje es un número, lanza ERR_TYPE_MISMATCH: conviértelo antes en cadena con toString(), como temperatura en el ejemplo.

subscribe(tema)

Se suscribe al tema tema, una cadena: desde entonces, los mensajes que se publiquen en él llegan al callback. Admite los comodines de MQTT: + vale por un nivel del tema y #, al final, por todos los que siguen.

use MQTT

MQTT.subscribe("casa/salon/luz")
MQTT.subscribe("comandos/#")       // todos los temas que empiezan por comandos/

Sin conexión no hace nada, y con la sesión limpia, que es lo de serie, la suscripción se pierde si se pierde la conexión: suscríbete cuando isConnect() dé true, y otra vez después de cada reconexión, como en el ejemplo completo.

unsubscribe(tema)

Deja de estar suscrito al tema tema. Tiene que ser el mismo texto con el que te suscribiste.

use MQTT

MQTT.subscribe("sensores/temperatura")
// … más tarde
MQTT.unsubscribe("sensores/temperatura")

callback(funcion)

Dice qué función recibe los mensajes de todos los temas a los que estás suscrito. Hay un solo callback: si lo llamas otra vez, el nuevo sustituye al anterior.

funcion puede ser una función (alRecibir), un método de un objeto (estacion.alRecibir) o el nombre de una función global en una cadena ("alRecibir"). Tiene que recibir exactamente dos parámetros, el tema y el mensaje, los dos cadenas: si tiene otro número, los mensajes se descartan sin avisar. Si funcion no es nada de eso, callback devuelve un valor de error.

use MQTT

func alRecibir(tema, mensaje)
    println(tema, ": ", mensaje)
end

MQTT.callback(alRecibir)
MQTT.subscribe("sensores/#")

Con un método, el callback corre con su this, y MQTT mantiene viva la instancia mientras esté registrada:

use MQTT

class Estacion
    def init(nombre)
        this.nombre = nombre
        this.mensajes = 0
    end

    def alRecibir(tema, mensaje)
        this.mensajes = this.mensajes + 1
        println(this.nombre, " recibe: ", mensaje)
    end
end

var meteo = Estacion("Meteo")
MQTT.callback(meteo.alRecibir)
MQTT.subscribe("sensores/#")

Cuándo corre el callback

Los mensajes llegan por su cuenta y esperan en una cola de la VM. El callback los recibe de uno en uno, dentro de tu programa, como los de Timer: en los puntos seguros de la VM (al dar la vuelta un bucle y al llamar a una función o volver de ella), nunca dentro de pause(). Por eso un programa que espera mensajes da vueltas con pausas cortas, y lo que llega durante un pause(10000) no se atiende hasta que acaba. Al terminar el programa, el callback se suelta.

isConnect()

Devuelve true si hay conexión con el broker, y false si no.

use MQTT

if (MQTT.isConnect()) then
    MQTT.publish("sensores/estado", "activo")
else
    println("Sin conexión MQTT")
end

pubinfo()

Publica la versión de la VM en el tema del aparato, <usuario>/<id>/device, con el usuario y el identificador de la configuración MQTT. Con el usuario vacío, que es lo de serie, el tema empieza por la barra: /<id>/device. El mensaje es como {"HWVersion":"0.9.7.20261104 01"}. Devuelve true si lo ha publicado y false si no hay conexión.

use MQTT

if (MQTT.pubinfo()) then
    println("Versión publicada")
end

intStats()

Solo en Windows. Devuelve un mapa con las cuentas de la cola de avisos de la VM, donde esperan su turno los mensajes de MQTT y también las interrupciones de GPIO, los disparos de los timers y los choques de sprites:

Clave	Qué cuenta
`enqueued`	Los avisos que han entrado en la cola
`dequeued`	Los que ha sacado la VM para atenderlos
`dropped`	Los que se han perdido porque la cola estaba llena
`hasPending`	`true` si ahora mismo hay alguno esperando

Las cuentas empiezan con el programa, o en el último resetIntStats(). En la cola caben 127 avisos: si dropped crece, el programa no da abasto, y le llegan más mensajes o interrupciones de los que atiende.

En las placas no existe: llamarla lanza ERR_ATTRIBUTE_NOT_FOUND (Module has no function 'intStats'.). En un programa que también va a la placa, pregunta antes con platform():

use MQTT

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

resetIntStats()

Solo en Windows. Pone a cero las cuentas de intStats() y devuelve true. En las placas no existe, como intStats.

Ejemplo completo

Un aparato que enciende y apaga un LED con los mensajes del tema casa/led, confirma cada cambio y publica un latido cada 10 segundos. Se suscribe al conectar, y otra vez si se pierde la conexión:

use MQTT, GPIO

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

// Un solo callback para todos los temas
func alRecibir(tema, mensaje)
    println("[MQTT] ", tema, ": ", mensaje)
    if (tema == "casa/led") then
        if (mensaje == "on") then
            GPIO.write(PIN_LED, HIGH)
        elsif (mensaje == "off") then
            GPIO.write(PIN_LED, LOW)
        end
        MQTT.publish("casa/led/estado", mensaje)   // confirma el cambio
    end
end

MQTT.callback(alRecibir)

var suscrito = false
var ultimoLatido = 0
while (true)
    if (MQTT.isConnect()) then
        if (!suscrito) then
            MQTT.subscribe("casa/led")   // al conectar, y otra vez tras perder la conexión
            suscrito = true
        end
        if (clock() - ultimoLatido > 10000) then
            MQTT.publish("casa/latido", "vivo")
            ultimoLatido = clock()
        end
    else
        suscrito = false
    end
    pause(50)   // pausas cortas: los mensajes se atienden entre una y otra
end

Trampas y problemas conocidos

Un true que no confirma nada. Sin conexión, publish, subscribe y unsubscribe no hacen nada y devuelven true. Mira isConnect().
El programa puede empezar antes que la conexión. En la placa, siempre; en Windows, si el broker no contesta en 3 segundos. Un subscribe al principio del programa se pierde si el broker todavía no ha contestado. Suscríbete cuando isConnect() dé true.
Las suscripciones no sobreviven a una reconexión. Con la sesión limpia de serie, al volver la conexión hay que suscribirse otra vez.
Pausas largas, mensajes que esperan. Mientras el programa está en pause(), los mensajes no se atienden: esperan en la cola, y si se llena, se pierden.
El callback, con dos parámetros. Con otro número de parámetros, los mensajes se descartan sin avisar.
Un Client ID para cada VM o placa. Dos con el mismo Client ID en el mismo broker se echan la una a la otra. En Windows, todos los proyectos traen de serie JARUWin11_01: si conectas dos a la vez, o usas un broker público como test.mosquitto.org, pon a cada proyecto el suyo en Config › Platform › Windows.

Todos los problemas conocidos de JARU, juntos, están en Problemas conocidos.

Plataformas​

Funciones​

publish(tema, mensaje)​

subscribe(tema)​

unsubscribe(tema)​

callback(funcion)​

isConnect()​

pubinfo()​

intStats()​

resetIntStats()​

Ejemplo completo​

Trampas y problemas conocidos​