Construindo Cliente MQTT Não-Bloqueante para Sistemas Embarcados Reativos

MQTT-Reactive é um cliente MQTT v3.1.1 leve, não-bloqueante, construído sobre a biblioteca MQTT-C de LiamBindle. Projetado para sistemas embarcados reativos, ele entrega uma stack de comunicação portável, event-driven que pode ser integrada em qualquer aplicação baseada em C.

Neste artigo, primeiro definimos o que constitui um sistema reativo, então delineamos uma arquitetura de software que abraça concorrência e comportamento baseado em estado. Finalmente, percorremos um exemplo concreto — um dispositivo IoT para uma empresa ferroviária — mostrando como empregar MQTT-Reactive via máquina de estado e paradigma event-driven.

Sistemas reativos respondem a eventos internos ou externos, processam o evento e retornam a um estado ocioso aguardando o próximo gatilho. Este paradigma event-driven produz código previsível e mantenível, especialmente quando múltiplos objetos ativos (unidades concorrentes) manipulam streams de eventos distintos.

Nosso dispositivo de estudo de caso executou as seguintes tarefas:

1. Detectou e registrou mudanças em várias entradas digitais
2. Adquiriu, filtrou e armazenou múltiplos sinais analógicos
3. Transmitiu periodicamente dados armazenados a um servidor remoto
4. Trocou dados com um broker usando MQTT sobre GSM

MQTT foi escolhido por seu modelo leve publish/subscribe, ideal para links de baixa largura de banda e alta latência como GSM.

Como o firmware do dispositivo foi construído como um sistema reativo, a biblioteca MQTT-C original requeria adaptação para comunicar assincronamente via eventos. A biblioteca resultante — MQTT-Reactive — expõe uma API limpa que pode ser invocada de uma máquina de estado enquanto mantém o resto do sistema desacoplado.

Máquina de estado

MQTT-Reactive é impulsionado pelo objeto ativo MqttMgr, cujo comportamento é descrito pela máquina de estado na Figura 1. A máquina inicia em WaitingForNetConnection; uma vez que um link de rede está disponível, transiciona para WaitingForSync, onde pacotes CONNECT e PUBLISH podem ser estagiados. Eventos que chegam em Sync são adiados até que o sistema re-entre em um estado que os aceite.

A cada SyncTime milissegundos a máquina move para o estado composto Sync, que posta eventos Receive e Send para o gerenciador de rede. Embora o exemplo cubra CONNECT e PUBLISH, estender suporte a SUBSCRIBE é direto.

As ações da máquina de estado usam a keyword params para ler dados de evento e o macro GEN() para gerar eventos de saída. Por exemplo:

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

Nesta transição, o evento Connect carrega parâmetros clientId e keepAlive que são aplicados aos atributos do objeto MqttMgr. Da mesma forma, eventos são despachados para outros atores via:

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

A ação padrão define um callback que MQTT-Reactive invoca quando o broker reconhece um CONNECT. O callback deve gerar ou ConnAccepted ou ConnRefused eventos:

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 um cliente MQTT-Reactive (Fonte: VortexMakes)

Implementação do modelo

A máquina de estado pode ser codificada em C ou C++ puro usando qualquer das seguintes ferramentas: framework RKH, framework QP, Yakindu Statechart Tool, ou Rational Rhapsody Developer. Todas suportam Statecharts e geração de código para C/C++.

O objeto ativo MqttMgr encapsula todas as chamadas MQTT-Reactive, protegendo o resto do sistema de uso direto da API. Concorrência é alcançada trocando eventos, eliminando armadilhas de semáforos, mutexes, delays ou event-flags.

Atributos de MqttMgr são armazenados diretamente em sua estrutura, ex.:

struct MqttMgr { struct mqtt_client client; // instância do cliente MQTT LocalRecvAll localRecv; // buffer de recepção local /* itens de dados adicionais */ };

Estes membros são acessados via o ponteiro me:

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

Tabela 1. Atributos do MqttMgr

O sistema geral inclui três atores-chave: Collector (publica dados), NetMgr (gerencia a rede) e MqttMgr (gerencia MQTT).

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

O diagrama de sequência na Figura 3 ilustra o fluxo quando MqttMgr inicia uma sessão com um broker MQTT.

Figura 3. Conectando a um broker MQTT (Fonte: VortexMakes)

Após estabelecimento bem-sucedido da rede, Collector envia Connect(clientId, keepAlive) para MqttMgr. O broker pode responder com ConnAccepted ou ConnRefused, este último carregando um código de rejeição conforme MQTT v3.1.1 §3.2.2.3.

Figura 4. Broker rejeita uma requisição de conexão (Fonte: VortexMakes)

Publicação segue padrão similar. Collector emite eventos Publish(data, size, topic, qos). O payload de dados, tipicamente JSON, é montado usando a biblioteca leve jWrite.

Figura 5. Publicando dados para um broker (Fonte: VortexMakes)

Falhas de rede são sinalizadas por NetMgr através de eventos ReceiveFail ou SendFail, permitindo que MqttMgr reaja adequadamente.

Figura 6. Falhas na rede (Fonte: VortexMakes)

Tabela 2. Eventos envolvidos nos cenários

Conclusão

Encapsulando MQTT-Reactive dentro de um objeto ativo e orquestrando comunicação via eventos, a abordagem elimina construções bloqueantes e produz um design limpo e testável. O modelo de máquina de estado escala para múltiplas tarefas concorrentes, tornando-o bem adequado para aplicações embarcadas reativas modernas que requerem conectividade MQTT confiável.