Implementación de un cliente MQTT para sistemas reactivos

MQTT-Reactive es un cliente MQTT v3.1.1 derivado de la biblioteca MQTT-C de LiamBindle. El objetivo de MQTT-Reactive es proporcionar un cliente MQTT portátil y sin bloqueo escrito en C para ser utilizado en sistemas integrados reactivos. En primer lugar, este artículo explica qué es un sistema reactivo. Luego, describe cómo diseñar una estructura de software adecuada para ese tipo de sistema. Finalmente, el artículo muestra cómo usar la biblioteca MQTT-Reactive en un sistema reactivo mediante el uso de una máquina de estado y el paradigma impulsado por eventos. Para ello, el artículo utiliza un dispositivo IoT real como ejemplo demostrativo del cual el artículo explica su estructura de software y su comportamiento basado en estados mediante el uso de diagramas UML como máquina de estados, interacción y estructura. El artículo también proporciona pautas para implementar el cliente MQTT-Reactive del dispositivo IoT en lenguaje C.

Muchos sistemas integrados son reactivos, es decir, reaccionan a eventos internos o externos. Una vez que se completan estas reacciones, el software vuelve a esperar el próximo evento. Es por eso que los sistemas controlados por eventos se denominan alternativamente sistemas reactivos.

La programación dirigida por eventos, o programación reactiva, es uno de los paradigmas de programación más adecuados para lograr un software flexible, predecible y mantenible para sistemas reactivos. En este paradigma, el flujo del programa está determinado por eventos. Con frecuencia, la estructura del software reactivo se compone de varias unidades concurrentes, conocidas como objetos activos, que esperan y procesan diferentes tipos de eventos. Cada objeto activo posee un hilo de control y una cola de eventos a través de la cual procesa sus eventos entrantes. En los sistemas reactivos, los objetos activos tienen típicamente un comportamiento basado en el estado definido en un diagrama de estado.

Para explorar cómo usar la biblioteca MQTT-Reactive en un sistema reactivo con tareas múltiples y concurrentes y usando tanto una máquina de estado como el paradigma impulsado por eventos, usamos un dispositivo IoT como ejemplo.

La idea de utilizar el protocolo MQTT nació mientras se desarrollaba un dispositivo IoT para una empresa ferroviaria. Este dispositivo era un sistema reactivo claro que podía:

1. detectar y almacenar cambios de varias entradas digitales
2. adquirir, filtrar y almacenar varias señales analógicas
3. enviar información almacenada a un servidor remoto periódicamente
4. enviar y recibir información a través del protocolo MQTT a través de la red GSM

Se eligió MQTT porque es un protocolo de mensajería ligero basado en suscriptores y editores que se usa comúnmente en aplicaciones de redes y de IoT donde se esperan enlaces de alta latencia y baja velocidad de datos, como las redes GSM.

La capacidad MQTT para el dispositivo IoT mencionado se logró mediante el uso de una versión modificada de MQTT-C de LiamBindle. Dado que el software de ese dispositivo había sido diseñado como un software reactivo, MQTT-C tuvo que ser modificado para comunicarlo con el resto del sistema mediante el intercambio de eventos asincrónicos. Estos eventos se utilizaron para recibir y enviar tráfico a través de la red, así como para conectar y publicar información confidencial en un servidor. La biblioteca de software resultante se llamó MQTT-Reactive.

Máquina de estado

MQTT-Reactive se utilizó a través de una máquina de estado como se muestra en la Figura 1, que modela el comportamiento básico de un cliente MQTT-Reactive. Era un objeto activo llamado MqttMgr (MQTT Manager). Las acciones de la máquina de estado en la Figura 1 demuestran cómo se podría usar la biblioteca MQTT-Reactive desde una máquina de estado. Aunque el lenguaje C se usó como lenguaje de acción en la Figura 1, se puede usar cualquier computadora o lenguaje formal.

haz clic para ampliar la imagen

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

La máquina de estado en la Figura 1 comienza en el estado WaitingForNetConnection. Después de que se establece una conexión de red a un servidor, WaitingForNetConnection recibe el evento Activate y luego la máquina de estado cambia al estado WaitingForSync. Solo en este estado la máquina de estado puede preparar mensajes MQTT para enviarlos al intermediario, como CONNECT o PUBLISH a través de los eventos Connect y Publish, respectivamente. El estado de sincronización utiliza un mecanismo especial de UML para aplazar el evento de publicación especificado por la palabra clave defer incluida en el compartimento interno del estado de sincronización. Si el evento de publicación ocurre cuando Sync es el estado actual, se guardará (diferido) para su procesamiento futuro hasta que el SM entre en un estado en el que el evento de publicación no esté en su lista de eventos diferidos, como WaitingForSync o WaitingForNetConnection. Al ingresar a dichos estados, la máquina de estado recordará automáticamente cualquier evento de publicación guardado y luego consumirá o descartará este evento de acuerdo con el estado de destino de la transición.

Cada milisegundos de SyncTime, la máquina de estado pasa al estado compuesto de sincronización, que realiza el envío y la recepción reales del tráfico de la red mediante la publicación de eventos de recepción y envío en el administrador de la red. Es una entidad concurrente que se ocupa de los problemas de la red.

Aunque el MqttMgr introducido solo admite los paquetes CONNECT y PUBLISH, podría admitir el paquete SUBSCRIBE con cambios bastante simples.

Las acciones de la máquina de estado acceden a los parámetros del evento consumido mediante el uso de la palabra clave params. Por ejemplo, en la siguiente transición, el evento Connect lleva dos parámetros, clientId y keepAlive, cuyos valores se utilizan para actualizar los atributos del objeto MqttMgr correspondiente:

 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 este ejemplo, el evento Connect (clientId, keepAlive) es el desencadenante de la transición y la llamada mqtt_connect () es parte de la acción que se ejecuta como resultado. En otras palabras, cuando el objeto MqttMgr recibe un evento Connect (clientId, keepAlive) con los parámetros de 'publishing_client' y '400', Connect ("publishing_client", 400), los atributos clientId y keepAlive de MqttMgr se actualizan con los valores ' publishing_client 'y' 400 'en consecuencia.

Para crear y enviar eventos, las acciones de la máquina de estado utilizan la macro GEN (). Por ejemplo, la siguiente declaración envía un evento de recepción al objeto Collector, al que se hace referencia como un atributo del objeto MqttMgr por su puntero de colector:

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

El primer argumento de la instrucción GEN () es el objeto que recibe el evento, mientras que el segundo argumento es el evento que se envía, incluidos los argumentos del evento (si los hay). Los argumentos deben coincidir con los parámetros del evento. Por ejemplo, la siguiente declaración genera un evento ConnRefused (código) y lo envía al objeto Collector pasando el código devuelto por el corredor como parámetro de evento:

 GEN (me-> itsCollector, ConRefused (código)); 

La idea de utilizar la palabra clave params para acceder a los parámetros del evento consumido y la macro GEN () para generar eventos a partir de acciones se adoptó del generador de código de Rational Rhapsody Developer con fines puramente ilustrativos.

La acción predeterminada de la máquina de estado en la Figura 1 establece la devolución de llamada a la que llama MQTT-Reactive cada vez que se recibe una aceptación de conexión del intermediario. Esta devolución de llamada debe implementarse dentro del código MqttMgr. Esta devolución de llamada debe generar eventos ConnAccepted o ConnRefused (código) para enviarse al objeto Collector como se muestra a continuación.

  estático   anular   connack_response_callback  ( enumeración  MQTTConnackReturnCode return_code) {/*...*/  si  (return_code ==MQTT_CONNACK_ACCEPTED) {GEN (me-> itsCollector, ConnAccepted ()); }  más  {GEN (me-> itsCollector, ConnRefused (return_code)); }} 

Implementación del modelo

El modelo de la Figura 1 podría implementarse en C o C ++ utilizando su herramienta de software favorita o simplemente su propia implementación de máquina de estado. Existen diferentes herramientas disponibles en Internet para hacer eso, como el framework RKH, el framework QP, Yakindu Statechart Tool, o Rational Rhapsody Developer, entre otros. Todos ellos admiten Statecharts y lenguajes C / C ++. Además, algunos de ellos incluyen una herramienta para dibujar un diagrama Statechart y generar código a partir de él.

Esta máquina de estado se ejecutó desde un objeto activo llamado MqttMgr (MQTT Manager), que proporcionó una encapsulación estricta del código MQTT-Reactive y fue la única entidad autorizada para llamar a cualquier función MQTT-Reactive o acceder a datos MQTT-Reactive. Las otras entidades concurrentes en el sistema, así como cualquier ISR, solo pudieron usar MQTT-Reactive indirectamente intercambiando eventos con MqttMgr. El uso de este mecanismo para sincronizar entidades concurrentes y compartir datos entre ellas evita lidiar con los peligros de los mecanismos de bloqueo tradicionales como semáforos, mutex, retrasos o banderas de eventos. Esos mecanismos podrían causar fallas inesperadas que son difíciles y tediosas de diagnosticar y corregir.

El objeto activo MqttMgr encapsula sus atributos como un conjunto de elementos de datos. Un elemento de datos designa una variable con un nombre y un tipo, donde el tipo es en realidad un tipo de datos. Un elemento de datos para el objeto MqttMgr se asigna a un miembro de la estructura del objeto. El nombre y el tipo del miembro son los mismos que los de los datos del objeto. Por ejemplo, el atributo de cliente del tipo de objeto MqttMgr está incrustado por valor como un miembro de datos dentro de la estructura MqttMgr:

  estructura   MqttMgr  { / * ... * /   estructura   mqtt_client   cliente ;  / * atributo cliente * /  LocalRecvAll localRecv;  / * atributo localRecv * /  }; 

Se accede a los datos del objeto MqttMgr y se modifican directamente sin utilizar operaciones de acceso o mutación. Por ejemplo, se accede a client y localRecv a través del puntero me, que apunta a una instancia de MqttMgr.

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

MqttMgr tiene la lista de atributos que se muestra en la Tabla 1.

Tabla 1. Atributos de MqttMgr

La estructura de la Figura 2 es útil para tener en cuenta las relaciones entre los actores involucrados. Ellos son:el objeto Recopilador, que quiere enviar información al corredor; el objeto NetMgr, que se ocupa de la red; y el objeto MqttMgr.

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

El diagrama de secuencia de la Figura 3 muestra cómo el objeto MqttMgr interactúa con el resto del sistema cuando es necesario abrir una sesión con un servidor MQTT. En este diagrama, el estado de MqttMgr y los mensajes asincrónicos intercambiados se muestran entre los actores Collector, MqttMgr y NetMgr.

Figura 3. Conexión a un corredor de MQTT (Fuente:VortexMakes)

Una vez que el objeto NetMgr establece una conexión de red a un intermediario, el primer paquete enviado desde MqttMgr a un servidor MQTT debe ser un paquete CONNECT. Por lo tanto, el actor recopilador envía un evento Connect (clientId, keepAlive) al actor MqttMgr. Este evento tiene que llevar el identificador del cliente y el intervalo de tiempo de actividad. Si el servidor acepta la solicitud de conexión, el actor MqttMgr enviará un evento ConnAccepted al actor recopilador para notificar esta situación. A partir de ese momento, el actor recopilador puede publicar mensajes de información para ese corredor.

Si el servidor rechaza la solicitud de conexión, el actor MqttMgr enviará un evento ConnRefused al actor recopilador. Este evento lleva consigo un código que notifica la causa del rechazo como se muestra en la Figura 4. Consulte MQTT v3.1.1 sección 3.2.2.3.

Figura 4. El corredor rechaza una solicitud de conexión (Fuente:VortexMakes)

La Figura 5 muestra el flujo de interacción cuando se publica un mensaje. Para ello, el actor recopilador envía un evento Publish (data, size, topic, qos), que lleva la información a publicar (data), la longitud de la información en bytes (size), el nombre del tema al que se publicará la información (tema) y el nivel de seguridad para entregar este mensaje (qos). En el dispositivo IoT mencionado anteriormente, la información publicada se formateó utilizando la especificación JSON. Es un formato estándar abierto que contiene objetos de datos con pares atributo-valor en texto legible por humanos. Este formato se logró usando jWrite, que es una biblioteca simple y liviana escrita en C.

Figura 5. Publicación de datos en un corredor (Fuente:VortexMakes)

La Figura 6 muestra un escenario en el que falla la recepción y envío de mensajes MQTT a la red. Si el administrador de red no puede recibir tráfico de la red, enviará un ReceiveFail al actor MqttMgr. De manera similar, si el administrador de red no puede enviar datos a la red, enviará un SendFail al actor MqttMgr.

Figura 6. Fallos en la red (Fuente:VortexMakes)

La Tabla 2 resume los eventos involucrados en los escenarios mostrados.

Tabla 2. Eventos

Conclusión

Al evitar los peligros de los mecanismos de bloqueo tradicionales, como semáforos, mutex, retrasos o indicadores de eventos, la biblioteca MQTT-Reactive, la máquina de estado y la arquitectura de software propuestas en este artículo permiten que los sistemas integrados reactivos implementen un cliente MQTT en una novela. camino. Se logra encapsulando el código MQTT-Reactive dentro de una unidad de simultaneidad llamada objeto activo , cuyo comportamiento basado en estados se define en la máquina de estados sugerida. Este objeto activo se comunica con el resto del sistema mediante el intercambio de eventos asincrónicos utilizados:no solo para recibir y enviar tráfico a través de la red, sino también para conectar y publicar información a un servidor para aplicaciones de Internet de las Cosas.