Implementando um cliente MQTT para sistemas reativos

MQTT-Reactive é um cliente MQTT v3.1.1 derivado da biblioteca MQTT-C de LiamBindle. O objetivo do MQTT-Reactive é fornecer um cliente MQTT portátil e sem bloqueio escrito em C para ser usado em sistemas embarcados reativos. Em primeiro lugar, este artigo explica o que é um sistema reativo. Em seguida, descreve como projetar uma estrutura de software adequada para esse tipo de sistema. Finalmente, o artigo mostra como usar a biblioteca MQTT-Reactive em um sistema reativo usando uma máquina de estado e o paradigma orientado a eventos. Para fazer isso, o artigo usa um dispositivo IoT real como um exemplo demonstrativo a partir do qual o artigo explica sua estrutura de software e comportamento baseado em estado usando diagramas UML, como máquina de estado, interação e estrutura. O artigo também fornece diretrizes para implementar o cliente MQTT-Reactive do dispositivo IoT em linguagem C.

Muitos sistemas embarcados são reativos, ou seja, eles reagem a eventos internos ou externos. Assim que essas reações forem concluídas, o software volta a aguardar o próximo evento. É por isso que os sistemas orientados a eventos são alternativamente chamados de sistemas reativos.

A programação orientada a eventos, ou programação reativa, é um dos paradigmas de programação mais adequados para obter um software flexível, previsível e sustentável para sistemas reativos. Neste paradigma, o fluxo do programa é determinado por eventos. Freqüentemente, a estrutura do software reativo é composta por várias unidades concorrentes, conhecidas como objetos ativos, que aguardam e processam diferentes tipos de eventos. Cada objeto ativo possui um thread de controle e uma fila de eventos por meio da qual processa seus eventos de entrada. Em sistemas reativos, os objetos ativos normalmente têm um comportamento baseado em estado definido em um gráfico de estado.

Para explorar como usar a biblioteca MQTT-Reactive em um sistema reativo com tarefas múltiplas e simultâneas e usando uma máquina de estado e o paradigma orientado a eventos, usamos um dispositivo IoT como exemplo.

A ideia de usar o protocolo MQTT nasceu enquanto um dispositivo IoT estava sendo desenvolvido para uma empresa ferroviária. Este dispositivo era um sistema reativo claro que era capaz de:

1. detectar e armazenar alterações de várias entradas digitais
2. adquirir, filtrar e armazenar vários sinais analógicos
3. enviar informações armazenadas para um servidor remoto periodicamente
4. enviar e receber informações por meio do protocolo MQTT pela rede GSM

MQTT foi escolhido porque é um protocolo de mensagens baseado em editor-assinante leve que é comumente usado em IoT e aplicativos de rede em que links de alta latência e baixa taxa de dados são esperados, como as redes GSM.

O recurso MQTT para o dispositivo IoT mencionado foi realizado usando uma versão modificada do MQTT-C de LiamBindle. Uma vez que o software desse dispositivo foi projetado como um software reativo, MQTT-C teve que ser modificado para se comunicar com o resto do sistema por meio da troca de eventos assíncronos. Esses eventos foram usados ​​para receber e enviar tráfego pela rede, bem como para conectar e publicar informações confidenciais em um servidor. A biblioteca de software resultante foi denominada MQTT-Reactive.

Máquina de estado

MQTT-Reactive foi usado por meio de uma máquina de estado conforme mostrado na Figura 1, que modela o comportamento básico de um cliente MQTT-Reactive. Era um objeto ativo chamado MqttMgr (MQTT Manager). As ações da máquina de estado na Figura 1 demonstram como a biblioteca MQTT-Reactive pode ser usada a partir de uma máquina de estado. Embora a linguagem C tenha sido usada como linguagem de ação na Figura 1, qualquer computador ou linguagem formal pode ser usada.

clique para ampliar a imagem

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

A máquina de estado na Figura 1 começa no estado WaitingForNetConnection. Depois que uma conexão de rede é estabelecida com um servidor, o WaitingForNetConnection recebe o evento Activate e, em seguida, a máquina de estado faz a transição para o estado WaitingForSync. Apenas neste estado as mensagens MQTT do estágio da máquina de estado podem ser enviadas ao broker, como CONNECT ou PUBLISH, por meio dos eventos Connect e Publish, respectivamente. O estado Sync usa um mecanismo especial da UML para adiar o evento Publish que é especificado pela palavra-chave defer incluída no compartimento interno do estado Sync. Se o evento Publish ocorrer quando Sync estiver no estado atual, ele será salvo (adiado) para processamento futuro até que o SM entre em um estado no qual o evento Publish não esteja em sua lista de eventos adiados, como WaitingForSync ou WaitingForNetConnection. Ao entrar em tais estados, a máquina de estado irá recuperar automaticamente qualquer evento de Publicação salvo e irá consumir ou descartar este evento de acordo com o estado de destino da transição.

A cada SyncTime milissegundos, a máquina de estado faz a transição para o estado composto de Sincronização, que faz o envio e recebimento real do tráfego da rede postando eventos de Recebimento e Envio para o gerenciador de rede. É uma entidade concorrente que lida com problemas de rede.

Mesmo que o MqttMgr introduzido só suporte os pacotes CONNECT e PUBLISH, ele pode suportar o pacote SUBSCRIBE com mudanças bastante simples.

As ações da máquina de estado acessam os parâmetros do evento consumido usando a palavra-chave params. Por exemplo, na transição a seguir, o evento Connect carrega dois parâmetros, clientId e keepAlive, cujos valores são usados ​​para atualizar os atributos do objeto MqttMgr correspondente:

 Conectar (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); 

Neste exemplo, o evento Connect (clientId, keepAlive) é o gatilho da transição e a chamada mqtt_connect () é parte da ação executada como resultado. Em outras palavras, quando o objeto MqttMgr recebe um evento Connect (clientId, keepAlive) com os parâmetros de 'publishing_client' e '400', Connect (“publishing_client”, 400), os atributos clientId e keepAlive do MqttMgr são atualizados com os valores ' publishing_client 'e' 400 'conseqüentemente.

Para criar e enviar eventos, as ações da máquina de estado usam a macro GEN (). Por exemplo, a instrução a seguir envia um evento Receber para o objeto Coletor, que é referido como um atributo do objeto MqttMgr por seu ponteiro Coletor:

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

O primeiro argumento da instrução GEN () é o objeto que recebe o evento, enquanto o segundo argumento é o evento que está sendo enviado, incluindo os argumentos do evento (se houver). Os argumentos devem concordar com os parâmetros do evento. Por exemplo, a instrução a seguir gera um evento ConnRefused (código) e o envia ao objeto Collector, passando o código retornado pelo corretor como um parâmetro de evento:

 GEN (me-> itsCollector, ConRefused (code)); 

A ideia de usar a palavra-chave params para acessar os parâmetros do evento consumido e a macro GEN () para gerar eventos a partir de ações foi adotada a partir do gerador de código do Rational Rhapsody Developer para fins puramente ilustrativos.

A ação padrão da máquina de estado na Figura 1 define o retorno de chamada que é chamado por MQTT-Reactive sempre que uma aceitação de conexão é recebida do corretor. Este retorno de chamada deve ser implementado no código MqttMgr. Este retorno de chamada deve gerar eventos ConnAccepted ou ConnRefused (código) para serem enviados ao objeto Collector conforme mostrado abaixo.

  estático   vazio   connack_response_callback  ( enum  MQTTConnackReturnCode return_code) {/*...*/  if  (return_code ==MQTT_CONNACK_ACCEPTED) {GEN (me-> itsCollector, ConnAccepted ()); }  mais  {GEN (me-> itsCollector, ConnRefused (return_code)); }} 

Implementação do modelo

O modelo na Figura 1 pode ser implementado em C ou C ++ usando sua ferramenta de software favorita ou apenas a implementação de sua própria máquina de estado. Existem diferentes ferramentas disponíveis na Internet para fazer isso, como a estrutura RKH, estrutura QP, Yakindu Statechart Tool ou Rational Rhapsody Developer, entre outras. Todos eles suportam Statecharts e linguagens C / C ++. Além disso, alguns deles incluem uma ferramenta para desenhar um diagrama Statechart e gerar código a partir dele.

Esta máquina de estado foi executada a partir de um objeto ativo chamado MqttMgr (MQTT Manager), que fornecia encapsulamento estrito do código MQTT-Reactive e era a única entidade com permissão para chamar qualquer função MQTT-Reactive ou acessar dados MQTT-Reactive. As outras entidades simultâneas no sistema, bem como quaisquer ISRs, só foram capazes de usar MQTT-Reactive indiretamente trocando eventos com MqttMgr. O uso desse mecanismo para sincronizar entidades concorrentes e compartilhar dados entre elas evita lidar com os perigos dos mecanismos de bloqueio tradicionais, como semáforos, mutex, atrasos ou sinalizadores de evento. Esses mecanismos podem causar mau funcionamento inesperado que é difícil e tedioso de diagnosticar e corrigir.

O objeto ativo MqttMgr encapsula seus atributos como um conjunto de itens de dados. Um item de dados designa uma variável com um nome e um tipo, onde o tipo é na verdade um tipo de dados. Um item de dados para o objeto MqttMgr é mapeado para um membro da estrutura do objeto. O nome e o tipo do membro são iguais aos dos dados do objeto. Por exemplo, o atributo cliente do tipo de objeto MqttMgr é incorporado por valor como um membro de dados dentro da estrutura MqttMgr:

  estrutura   MqttMgr  { / * ... * /   estrutura   mqtt_client   cliente ;  / * attribute client * /  LocalRecvAll localRecv;  / * attribute localRecv * /  }; 

Os dados do objeto MqttMgr são acessados ​​e modificados diretamente, sem o uso de operações de acessador ou modificador. Por exemplo, client e localRecv são acessados ​​por meio do ponteiro me, que aponta para uma instância de MqttMgr.

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

O MqttMgr possui a lista de atributos mostrada na Tabela 1.

Tabela 1. Atributos MqttMgr


A estrutura da Figura 2 é útil para ter em mente as relações entre os atores envolvidos. São eles:o objeto Coletor, que deseja enviar informações para a corretora; o objeto NetMgr, que trata da rede; e o objeto MqttMgr.

Figura 2. Rascunho da estrutura do sistema IoT (Fonte:VortexMakes)

O diagrama de sequência na Figura 3 mostra como o objeto MqttMgr interage com o resto do sistema quando é necessário abrir uma sessão com um servidor MQTT. Neste diagrama, o estado MqttMgr e as mensagens assíncronas trocadas são exibidos entre os atores Collector, MqttMgr e NetMgr.

Figura 3. Conectando-se a um corretor MQTT (Fonte:VortexMakes)

Depois que uma conexão de rede é estabelecida pelo objeto NetMgr para um broker, o primeiro pacote enviado de MqttMgr para um servidor MQTT deve ser um pacote CONNECT. Assim, o ator Collector envia um evento Connect (clientId, keepAlive) para o ator MqttMgr. Este evento deve conter o identificador do cliente e o intervalo de tempo de manutenção de funcionamento. Se o servidor aceitar a solicitação de conexão, o ator MqttMgr enviará um evento ConnAccepted ao ator Coletor para notificar essa situação. A partir de então, o ator Collector pode publicar mensagens informativas para esse broker.

Se o servidor rejeitar a solicitação de conexão, o ator MqttMgr enviará um evento ConnRefused ao ator Coletor. Este evento carrega consigo um código que notifica a causa da rejeição, conforme mostrado na Figura 4. Consulte MQTT v3.1.1 seção 3.2.2.3.

Figura 4. O corretor rejeita uma solicitação de conexão (Fonte:VortexMakes)

A Figura 5 mostra o fluxo de interação quando uma mensagem é publicada. Para fazer isso, o ator Collector envia um evento Publish (data, size, topic, qos), que carrega as informações a serem publicadas (data), o comprimento das informações em bytes (size), o nome do tópico para o qual as informações serão publicadas (tópico) e o nível de garantia para entregar esta mensagem (qos). No dispositivo IoT mencionado anteriormente, as informações publicadas foram formatadas usando a especificação JSON. É um formato padrão aberto que contém objetos de dados com pares de valor de atributo em texto legível por humanos. Este formato foi realizado usando jWrite, que é uma biblioteca simples e leve escrita em C.

Figura 5. Publicação de dados para um corretor (Fonte:VortexMakes)

A Figura 6 mostra um cenário em que a recepção e o envio de mensagens MQTT para a rede falham. Se o gerenciador de rede não puder receber tráfego da rede, ele enviará um ReceiveFail para o ator MqttMgr. Da mesma forma, se o gerente da rede não puder enviar dados para a rede, ele enviará um SendFail para o ator MqttMgr.

Figura 6. Falhas na rede (Fonte:VortexMakes)

A Tabela 2 resume os eventos envolvidos nos cenários mostrados.

Tabela 2. Eventos


Conclusão

Ao evitar os perigos dos mecanismos de bloqueio tradicionais - como semáforos, mutex, atrasos ou sinalizadores de evento - a biblioteca MQTT-Reactive, a máquina de estado e a arquitetura de software proposta neste artigo permitem que sistemas embarcados reativos implementem um cliente MQTT em um romance caminho. É conseguido encapsulando o código MQTT-Reactive dentro de uma unidade de simultaneidade chamada objeto ativo , cujo comportamento baseado em estado é definido na máquina de estado sugerida. Este objeto ativo se comunica com o resto do sistema por meio da troca de eventos assíncronos usados:não apenas para receber e enviar tráfego pela rede, mas também para conectar e publicar informações em um servidor para aplicativos da Internet das Coisas.