Construyendo cliente MQTT no bloqueante para sistemas embebidos reactivos

MQTT-Reactive es un cliente ligero, no bloqueante MQTT v3.1.1 construido sobre la librería MQTT-C de Liam Bindle. Diseñado para sistemas embebidos reactivos, entrega una pila de comunicación portable, event-driven que puede integrarse en cualquier aplicación basada en C.

En este artículo primero definimos qué constituye un sistema reactivo, luego esbozamos una arquitectura de software que abraza concurrencia y comportamiento basado en estados. Finalmente, recorremos un ejemplo concreto —un dispositivo IoT para una compañía ferroviaria— mostrando cómo emplear MQTT-Reactive vía máquina de estado y paradigma event-driven.

Los sistemas reactivos responden a eventos internos o externos, procesan el evento y retornan a estado idle esperando el siguiente disparador. Este paradigma event-driven produce código predecible, mantenible, especialmente cuando múltiples objetos activos (unidades concurrentes) manejan flujos de eventos distintos.

Nuestro dispositivo de caso de estudio realizó las siguientes tareas:

1. Detectó y registró cambios en varias entradas digitales
2. Adquirió, filtró y almacenó múltiples señales analógicas
3. Transmitió periódicamente datos almacenados a un servidor remoto
4. Intercambió datos con un broker usando MQTT sobre GSM

MQTT fue elegido por su modelo ligero publish/subscribe, ideal para enlaces de bajo ancho de banda y alta latencia como GSM.

Como el firmware del dispositivo se construyó como sistema reactivo, la librería MQTT-C original requería adaptación para comunicar asincrónicamente vía eventos. La librería resultante —MQTT-Reactive— expone una API limpia invocable desde una máquina de estado manteniendo el resto del sistema desacoplado.

Máquina de estado

MQTT-Reactive es impulsado por el MqttMgr active object, cuyo comportamiento se describe por la máquina de estado en Figura 1. La máquina inicia en WaitingForNetConnection; una vez que un enlace de red está disponible, transiciona a WaitingForSync, donde paquetes CONNECT y PUBLISH pueden prepararse. Eventos que llegan en Sync se diferan hasta que el sistema re-ingresa a un estado que los acepte.

Cada SyncTime milisegundos la máquina se mueve al estado compuesto Sync, que publica eventos Receive y Send al gestor de red. Aunque el ejemplo cubre CONNECT y PUBLISH, extender soporte a SUBSCRIBE es directo.

Las acciones de la máquina de estado usan la palabra clave params para leer datos de evento y la macro GEN() para generar eventos salientes. Por ejemplo:

Connect(clientId, keepAlive)/ me->clientId = params->clientId; me->keepAlive = params->keepAlive; me->operRes = mqtt_connect(&me->client, me->clientId, NULL, NULL, 0, NULL, NULL, 0, me->keepAlive);

En esta transición, el evento Connect lleva parámetros clientId y keepAlive que se aplican a los atributos del objeto MqttMgr. Asimismo, eventos se despachan a otros actores vía:

GEN(me->itsCollector, Receive());

La acción por defecto establece un callback que MQTT-Reactive invoca cuando el broker acusa recibo de un CONNECT. El callback debe generar eventos ConnAccepted o ConnRefused:

static void connack_response_callback(enum MQTTConnackReturnCode return_code) { if (return_code == MQTT_CONNACK_ACCEPTED) { GEN(me->itsCollector, ConnAccepted()); } else { GEN(me->itsCollector, ConnRefused(return_code)); } }

Figura 1. Máquina de estado de un cliente MQTT-Reactive (Fuente: VortexMakes)

Implementación del modelo

La máquina de estado puede codificarse en C o C++ plano usando cualquiera de estas herramientas: framework RKH, framework QP, Yakindu Statechart Tool, o Rational Rhapsody Developer. Todas soportan Statecharts y generación de código para C/C++.

El MqttMgr active object encapsula todas las llamadas MQTT-Reactive, blindando el resto del sistema de uso directo de API. La concurrencia se logra intercambiando eventos, eliminando los escollos de semáforos, mutexes, delays o event-flags.

Atributos de MqttMgr se almacenan directamente en su estructura, ej.:

struct MqttMgr { struct mqtt_client client; // instancia cliente MQTT LocalRecvAll localRecv; // buffer recepción local /* items de datos adicionales */ };

Estos miembros se acceden vía el puntero me:

mqtt_recvMsgError(&me->client, &me->localRecv);

Tabla 1. Atributos MqttMgr

El sistema general incluye tres actores clave: Collector (publica datos), NetMgr (gestiona la red), y MqttMgr (gestiona MQTT).

Figura 2. Borrador de estructura de sistema IoT (Fuente: VortexMakes)

El diagrama de secuencia en Figura 3 ilustra el flujo cuando MqttMgr inicia una sesión con un broker MQTT.

Figura 3. Conectando a un broker MQTT (Fuente: VortexMakes)

Tras establecimiento exitoso de red, Collector envía Connect(clientId, keepAlive) a MqttMgr. El broker puede responder con ConnAccepted o ConnRefused, este último llevando código de rechazo per MQTT v3.1.1 §3.2.2.3.

Figura 4. Broker rechaza una petición de conexión (Fuente: VortexMakes)

Publicar sigue patrón similar. Collector emite eventos Publish(data, size, topic, qos). La carga de datos, típicamente JSON, se ensambla usando la ligera librería jWrite.

Figura 5. Publicando datos a un broker (Fuente: VortexMakes)

Fallos de red son señalados por NetMgr a través de eventos ReceiveFail o SendFail, permitiendo a MqttMgr reaccionar en consecuencia.

Figura 6. Fallos en red (Fuente: VortexMakes)

Tabla 2. Eventos involucrados en los escenarios

Conclusión

Encapsulando MQTT-Reactive dentro de un active object y orquestando comunicación vía eventos, el enfoque elimina constructos bloqueantes y produce un diseño limpio, testeable. El modelo de máquina de estado escala a múltiples tareas concurrentes, haciéndolo idóneo para aplicaciones embebidas reactivas modernas que requieren conectividad MQTT fiable.