Создание неблокирующего MQTT клиента для реактивных встроенных систем

MQTT-Reactive — легковесный, неблокирующий MQTT v3.1.1 клиент, построенный на библиотеке MQTT-C Лиама Биндл. Проектированный для реактивных встроенных систем, он доставляет переносимый, событийно-ориентированный коммуникационный стек, который может быть интегрирован в любое C-базированное приложение.

В этой статье мы сначала определяем, что составляет реактивную систему, затем описываем программную архитектуру, охватывающую конкурентность и поведение на основе состояний. Наконец, мы проходим через конкретный пример — IoT-устройство для железнодорожной компании — показывая, как использовать MQTT-Reactive через конечный автомат и событийно-ориентированную парадигму.

Реактивные системы отвечают на внутренние или внешние события, обрабатывают событие и возвращаются в простое состояние в ожидании следующего триггера. Эта событийно-ориентированная парадигма дает предсказуемый, поддерживаемый код, особенно когда множественные активные объекты (конкурентные единицы) обрабатывают отдельные потоки событий.

Наше устройство-пример выполняло следующие задачи:

1. Обнаруживало и логировало изменения в нескольких цифровых входах
2. Приобретало, фильтровало и хранило множественные аналоговые сигналы
3. Периодически передавало сохраненные данные на удаленный сервер
4. Обменивалось данными с брокером используя MQTT через GSM

MQTT был выбран за свою легковесную publish/subscribe модель, идеальную для низкопропускных, высоколатентных связей вроде GSM.

Поскольку прошивка устройства была построена как реактивная система, оригинальная библиотека MQTT-C требовала адаптации для асинхронного общения через события. Получившаяся библиотека — MQTT-Reactive — экспонирует чистый API, который может быть вызван из конечного автомата, сохраняя остальную систему развязанной.

Конечный автомат

MQTT-Reactive управляется активным объектом MqttMgr, поведение которого описано конечным автоматом на Рисунке 1. Машина начинается в WaitingForNetConnection; как только сетевое соединение доступно, она переходит в WaitingForSync, где пакеты CONNECT и PUBLISH могут быть поставлены в очередь. События, прибывающие в Sync, откладываются до повторного входа системы в состояние, принимающее их.

Каждые SyncTime миллисекунд машина переходит в составное состояние Sync, которое посылает события Receive и Send менеджеру сети. Хотя пример покрывает CONNECT и PUBLISH, расширение поддержки до SUBSCRIBE просто.

Действия конечного автомата используют ключевое слово params для чтения данных события и макрос GEN() для генерации исходящих событий. Например:

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

В этом переходе событие Connect несет параметры clientId и keepAlive, которые применяются к атрибутам объекта MqttMgr. Подобно этому, события диспетчеризируются к другим актерам через:

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

Дефолтное действие устанавливает коллбэк, который MQTT-Reactive вызывает, когда брокер подтверждает CONNECT. Коллбэк должен сгенерировать либо ConnAccepted, либо 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)); } }

Рисунок 1. Конечный автомат MQTT-Reactive клиента (Источник: VortexMakes)

Реализация модели

Конечный автомат может быть закодирован на чистом C или C++ с использованием любого из следующих инструментов: RKH framework, QP framework, Yakindu Statechart Tool или Rational Rhapsody Developer. Все поддерживают Statecharts и генерацию кода для C/C++.

Активный объект MqttMgr инкапсулирует все вызовы MQTT-Reactive, защищая остальную систему от прямого использования API. Конкурентность достигается обменом событиями, что устраняет подводные камни семафоров, мьютексов, задержек или event-флагов.

Атрибуты MqttMgr хранятся напрямую в его структуре, например:

struct MqttMgr { struct mqtt_client client; // экземпляр MQTT клиента LocalRecvAll localRecv; // локальный буфер приема /* дополнительные элементы данных */ };

Эти члены доступны через указатель me:

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

Таблица 1. Атрибуты MqttMgr

Общая система включает три ключевых актера: Collector (публикует данные), NetMgr (управляет сетью) и MqttMgr (управляет MQTT).

Рисунок 2. Черновик структуры IoT системы (Источник: VortexMakes)

Диаграмма последовательности на Рисунке 3 иллюстрирует поток, когда MqttMgr инициирует сессию с MQTT брокером.

Рисунок 3. Подключение к MQTT брокеру (Источник: VortexMakes)

После успешного установления сети, Collector посылает Connect(clientId, keepAlive) в MqttMgr. Брокер может ответить с ConnAccepted или ConnRefused, последний несет код отказа согласно MQTT v3.1.1 §3.2.2.3.

Рисунок 4. Брокер отклоняет запрос соединения (Источник: VortexMakes)

Публикация следует похожему паттерну. Collector выдает события Publish(data, size, topic, qos). Данные полезной нагрузки, обычно JSON, собираются с помощью легковесной библиотеки jWrite.

Рисунок 5. Публикация данных брокеру (Источник: VortexMakes)

Сетевые сбои сигнализируются NetMgr через события ReceiveFail или SendFail, позволяя MqttMgr реагировать соответственно.

Рисунок 6. Сбои в сети (Источник: VortexMakes)

Таблица 2. События, участвующие в сценариях

Заключение

Инкапсулируя MQTT-Reactive внутри активного объекта и оркестрируя коммуникацию через события, подход устраняет блокирующие конструкции и дает чистый, тестируемый дизайн. Модель конечного автомата масштабируется до множественных конкурентных задач, делая ее хорошо подходящей для современных реактивных встроенных приложений, требующих надежного MQTT соединения.