Práctica 5. El protocolo MQTT (I). Despliegue de clientes y servidores/brokers. Análisis de tráfico

Objetivos

• Familiarizarse con el uso de brokers y clientes de suscripción/publicación utilizando MQTT.

• Desplegar un sistema basado en MQTT local, incluyendo broker y clientes.

• Utilizar Eclipse Paho para integrar funcionalidad MQTT en programas Python.

• Familiarizarse con el uso de wildcards MQTT.

Publicación/suscripción contra un broker en la nube

En la primera parte de la práctica, utilizaremos un servidor/broker disponible en la nube para su uso libre por parte de los usuarios (test.mosquitto.org). Este servidor suele utilizarse con fines de pruebas básicas y depuración, y hay que ser consciente de que toda la información que en él se publica puede ser leída por cualquier suscriptor. Debe tenerse este dato en cuenta a la hora de publicar información sensible a través de MQTT cuando se use el servidor de pruebas.

El servidor escucha en los siguientes puertos:

• 1883: MQTT, sin encriptación.
• 8883: MQTT, con encriptación.
• 8884: MQTT, con encriptación, certificado de cliente requerido.
• 8080: MQTT sobre WebSockets, sin encriptación.
• 8081: MQTT sobre WebSockets, con encriptación.

Para realizar publicaciones/suscripciones contra el broker utilizaremos la distribución mosquitto del proyecto Eclipse IoT. Aunque mosquitto es principalmente una implementación de broker MQTT, nosotros la utilizaremos en este paso a modo de cliente, lo que nos permitirá suscribirnos o publicar sobre cualquier topic MQTT.

En primer lugar, instala mosquitto:

sudo apt-get update
sudo apt-get install mosquitto mosquitto-clients mosquitto-dev libmosquitto*

Si todo ha ido bien, deberías disponer de dos binarios listos para ejecución:

• mosquitto_sub: permite suscribirse a un determinado topic utilizando un broker.
• mosquitto_pub: permite publicar un mensaje asociado a un determinado topic utilizando un broker.

Suscribámonos al topic # en el broker, utilizando para ello la orden:

mosquitto_sub -h test.mosquitto.org  -t "#"

obteniendo?

A continuación, vamos a realizar un proceso de publicación/suscripcion con un topic conocido (por ejemplo, /MIOT/tunombre/). Para publicar un mensaje bajo dicho topic:

mosquitto_pub -h test.mosquitto.org -t "/MIOT/tunombre" -m "Hola, soy tunombre"

Despliegue de un broker local usando Eclipse Mosquitto

El uso de un servidor remoto presenta ventajas (facilidad de uso), pero una gran cantidad de inconvenientes (seguridad, imposibilidad de configuración avanzada, ...)

En esta sección, configuraremos un broker mosquitto para el despliegue de una infraestructura MQTT local o remota bajo nuestro control.

El arranque de un broker (servidor) mosquitto se realiza mediante el propio comando mosquitto:

mosquitto [-c config file] [ -d | --daemon ] [-p port number] [-v]

Sin embargo, en la mayoría de distribuciones Linux, el broker arranca por defecto y se ejecuta constantemente en segundo plano. Para comprobar el estado de funcionamiento del broker, basta con ejecutar:

sudo service mosquitto status

Observarás un mensaje que indica que el servicio está activo. Las opciones restart, start o stop te permitirán controlar el estado del broker en todo momento.

El broker mosquitto permite monitorizar sus propias estadísticas e información de estado utilizando el protocolo MQTT. Así, los topics $SYS retornan, bien periódicamente o bien cuando sucede un evento de interés, la información de estado del broker. Puedes consultar más detalles en la página de manual de mosquitto (comando man mosquitto), en el epígrafe BROKER STATUS.

Wildcards

Además de permitir el uso de topics completos para el proceso de suscripción, los topics pueden incluir wildcards o comodines en su estructura. + es la wildcard utilizada para obtener correspondencias con un único nivel de la jerarquía. Así, para un topic a/b/c/d, las siguientes suscripciones corresponderán con éxito:

• a/b/c/d
• +/b/c/d
• a/+/c/d
• a/+/+/d
• +/+/+/+

Pero no las siguientes:

• a/b/c
• b/+/c/d
• +/+/+

La segunda wildcard soportada es #, y permite corresponencias con cualquier nivel sucesivo de la jerarquía. Así, para un topic a/b/c/d, las siguientes suscripciones corresponderán con éxito:

• a/b/c/d
• #
• a/#
• a/b/#
• a/b/c/#
• +/b/c/#

Desarrollo de un clientes locales. Eclipse Paho

Los clientes mosquitto_pub y mosquitto_sub son básicamente herramientas de desarrollo y pruebas, pero resulta interesante conocer bibliotecas que permitan la integración de MQTT en programas existentes. Una de ellas es Eclipse Paho. Paho es una infraestructura desarrollada en el proyecto Eclipse IoT para dar soporte a implementaciones de protocolos de mensajería M2M e IoT, aunque, en este momento, su uso principal se centra exclusivamente en MQTT.

En nuestro caso, utilizaremos la versión Python de la biblioteca, instalable vía:

pip install paho-mqtt

Dispones de la documentación del módulo a través de este enlace.

El despliegue de un ejemplo sencillo para un cliente que se conecta a un broker y se suscribe al tópico $SYS, imprimiendo los mensajes recibidos, resultaría, utilizando Paho, en el siguiente código Python:

import paho.mqtt.client as mqtt

# Funcion callback invocada cuandl el cliente recibe un CONNACK desde el broker.
def on_connect(client, userdata, flags, rc):
    print("Connected with result code "+str(rc))

    # Suscribirse en on_connect() asegura que si se pierde la conexión y 
    # se reestablece, las suscripciones se renovarán.
    client.subscribe("$SYS/#")

# Funcion callback al recibir un mensaje de publicacion (PUBLISH) desde el 
# broker.
def on_message(client, userdata, msg):
    print(msg.topic+" "+str(msg.payload))

client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message

client.connect("mqtt.eclipse.org", 1883, 60)

# Llamada bloqueante que procesa el tráfico de red, invoca callbacks
# y maneja la reconexión al broker. 
client.loop_forever()

La clase cliente puede utilizarse para:

• Crear una instancia de cliente MQTT.
• Conectar a un broker usando las funciones de la familia connect*().
• Invocar a funciones de la familia loop*() para mantener el tráfico de datos con el servidor.
• Utilizar subscribe() para suscribirse a un topic y recibir mensajes.
• Utilizar publish() publicar mensajes en el broker.
• Utilizar disconnect() para desconectar del broker.

Los callbacks se invocarán automáticamente para permitir el procesamiento de eventos. De entre los más utilizados, destacan:

• ON_CONNECT: invocado cuando el broker responde a nuestra petición de conexión. Ejemplo:

def on_connect(client, userdata, flags, rc):
    print("Connection returned result: "+connack_string(rc))

• ON_DISCONNECT: invocado cuando el cliente se desconecta del broker. Ejemplo:

def on_disconnect(client, userdata, rc):
    if rc != 0:
        print("Unexpected disconnection.")

• ON_MESSAGE: invocado cuando se recibe un mensaje en un topic al que
• el cliente está suscrito. Ejemplo:

def on_message(client, userdata, message):
    print("Received message '" + str(message.payload) + "' on topic '"
        + message.topic + "' with QoS " + str(message.qos))

Para publicar de forma puntual sobre un broker (sin mantener una conexión establecida), es posible utilizar la siguiente secuencia de ordenes:

import paho.mqtt.publish as publish

publish.single("paho/test/single", "payload", hostname="mqtt.eclipse.org")

Del mismo modo, podemos suscribirnos de forma puntual mediante una llamada bloqueante a:

import paho.mqtt.subscribe as subscribe

msg = subscribe.simple("paho/test/simple", hostname="mqtt.eclipse.org")
print("%s %s" % (msg.topic, msg.payload))

Toda la información y documentación asociada al módulo puede consultarse aquí.