198и. MQTT-клиент на STM32: публикация, подписка и повторное подключение

В разделе 198з прибор на F407 сам был сервером: браузер приходил к нему за страницей и данными. Для парка приборов удобнее обратная схема — прибор сам подключается к брокеру, публикует измерения и слушает команды (раздел 188.4). Здесь это сделано на клиенте MQTT, который входит в lwIP: src/apps/mqtt/mqtt.c, raw API, без ОС.

Что такое MQTT, из чего состоит пакет и зачем QoS и завещание — уже разобрано в разделах 37.2, 87.5 и 188.4, а клиент ESP-IDF — в 156.6. Здесь другое: как этим пользуется именно lwIP, что он на самом деле шлёт в провод, где его буферы кончаются и как прибор переживает обрыв кабеля и падение брокера. Всё проверено на настоящем брокере Mosquitto — со временами из прогона.

Опоры: исходники lwIP 2.1.2 из STM32CubeF4 1.28.0 (mqtt.c, mqtt.h, mqtt_priv.h, mqtt_opts.h), драйвер Ethernet и сетевой интерфейс из разделов 106а и 106в, журнал брокера eclipse-mosquitto:2. Файлы — _verify/mqtt198j/: mqtt198j_app.c (клиент прибора, общий для платы и ПК), host198j.c (прогон на ПК), mqtt_frame.c (свой кодировщик), sub198j.py (подписчик), run198j.py (сценарий), build_f407.py (сборка armclang). Сверяет _verify/mqtt198j_check.py.

Как это называется в документации

Оригинал	Перевод
keep alive	интервал «я жив»: сколько секунд прибор может молчать
Last Will and Testament (LWT)	завещание: что брокер опубликует, если прибор пропал
retained message	сохранённое сообщение: брокер отдаёт его каждому новому подписчику
packet identifier	номер пакета: связывает PUBLISH с QoS 1 и его PUBACK
remaining length	длина остатка: от 1 до 4 байт по 7 бит
clean session	чистая сессия: брокер не хранит ничего между подключениями
output ring buffer	кольцо передачи клиента lwIP
exponential backoff	повтор с нарастающей паузой

198и.1 Прибор, темы и стенд

Прибор f407-01 делает четыре вещи:

• раз в 30 с публикует измерение в lab/f407-01/meas с QoS 1;
• подписан на lab/f407-01/cmd/+: команда led включает светодиод, echo отвечает длиной принятого;
• ответы шлёт в lab/f407-01/resp с QoS 0;
• в lab/f407-01/state держит online с флагом сохранения, а при подключении оставляет брокеру завещание offline — тоже с сохранением.

Темы строятся от общего к частному: площадка, прибор, вид данных. Тогда диспетчер подписывается на lab/# и видит всё, а прибор — только на свои команды через +. Имя прибора в теме совпадает с его идентификатором клиента: по журналу брокера сразу видно, кто есть кто.

Платы с Ethernet в опыте нет, а QEMU не моделирует Ethernet F407. Поэтому тот же файл mqtt198j_app.c собран для Linux вместе с lwIP 2.1.2 и подключён к ядру через устройство TAP — как в разделе 106в. Контейнер клиента живёт в сетевом пространстве контейнера брокера: ядро с адресом 10.198.0.1 — это и есть брокер, lwIP — 10.198.0.2. Подписчик — программа на Python с голым сокетом, без библиотек MQTT. Каждый кадр, который lwIP отдал в TAP или получил из него, пишется в журнал «провода» с временем в миллисекундах. Файл /tmp/cut в контейнере клиента — «кабель выдернут»: кадры в обе стороны теряются, а связь для lwIP остаётся поднятой, как у настоящего кабеля, выдернутого из коммутатора где-то дальше по линии. На ПК шаг измерений — 40 с вместо 30, чтобы между ними было видно PINGREQ.

198и.2 Клиент lwIP: четыре вызова и обратные вызовы

Весь клиент — четыре функции и три обратных вызова:

on_conn вызывается и при успехе (MQTT_CONNECT_ACCEPTED), и при разрыве: 256 — MQTT_CONNECT_DISCONNECTED (TCP закрыт или отвергнут), 257 — MQTT_CONNECT_TIMEOUT (брокер замолчал). В прогоне встретились оба.

Три подробности видны только в исходнике:

• mqtt_client_connect начинает с memset всей структуры клиента. Обратные вызовы входящих публикаций стираются вместе с ней, поэтому mqtt_set_inpub_callback и mqtt_subscribe вызываются в on_conn после каждого подключения, а не один раз при старте. Номера пакетов тоже начинаются заново: в прогоне после каждого переподключения подписка получает номер 1, online — 2, первое измерение — 3.
• Клиент можно держать статически. mqtt_client_new() берёт память из кучи lwIP, а описание mqtt_client_t лежит в mqtt_priv.h: хватает static mqtt_client_t cli;. На F407 она занимает 496 байт, на 64-битном ПК — 592 (указатели длиннее).
• Чистая сессия — всегда. Флаг clean session lwIP ставит сам и выключить его не даёт. Значит, после обрыва брокер ничего для прибора не хранит: подписку и неподтверждённые измерения восстанавливает сам прибор.

Сборка для F407 (build_f407.py, armclang -Os, свои файлы с -Wall -Wextra -Werror) — без предупреждений. Весь образ — 29 780 байт Flash и 28 924 байта ОЗУ. Из них lwIP с клиентом MQTT — 25 566 байт кода и 17 322 байта ОЗУ (куча 6 КБ и пулы), сам mqtt.c — 3 514 байт кода, приложение прибора — 1 820 байт кода и 589 байт ОЗУ. Две проверки на этапе сборки стоят в mqtt198j_app.c:

/* CONNECT с завещанием целиком помещается в кольцо передачи */
#define CONNECT_LEN (2u + 10u + 2u + sizeof(DEV_ID) - 1u + 2u + sizeof(T_STATE) - 1u + 2u + sizeof("offline") - 1u)
_Static_assert(CONNECT_LEN <= MQTT_OUTPUT_RINGBUF_SIZE, "CONNECT не влезает в MQTT_OUTPUT_RINGBUF_SIZE");
_Static_assert(sizeof(T_STATE) - 1u <= 255u, "lwIP хранит длину темы завещания в одном байте");

Вторая — о настоящем ограничении lwIP 2.1.2: длины темы и текста завещания он проверяет на 255 байт и хранит в u8_t, хотя протокол разрешает до 65 535.

198и.3 Пакеты по байтам: свой кодировщик против провода

Кодировщик MQTT 3.1.1 уже написан в разделе 87.5. Здесь mqtt_frame.c дополнен тем, что нужно прибору: CONNECT с завещанием, PUBLISH с QoS 1 и номером, SUBSCRIBE, PUBACK, PINGREQ. Сверка устроена так: разборщик журнала собирает поток TCP каждого соединения (повторы отбрасываются по номерам последовательности), режет его на пакеты MQTT и для каждого исходящего пакета составляет заявку — «CONNECT прибора», «PUBLISH QoS 1 retain 0 номер 3 в такую-то тему с таким-то телом». Свой кодировщик строит пакет по заявке, и байты сравниваются.

За прогон lwIP отправил 23 пакета MQTT, различных среди них 15 — все 15 совпали со своим кодировщиком байт в байт. Проверка ещё раз строит их кодировщиком на Python — тоже совпадение. Самые показательные:

Брокер в журнале повторил те же поля: p4, c1, k20 — протокол 4, чистая сессия, keep-alive 20; Will message specified (7 bytes) (r1, q1).

Входящую публикацию с QoS 1 lwIP подтверждает сам: PUBACK с номером, который назначил брокер (40 02 00 01), уходит после последнего куска данных. Приложению делать ничего не нужно — и повлиять на это оно тоже не может.

198и.4 Буферы клиента и входящая публикация кусками

Размеры клиента задаются в lwipopts.h, значения по умолчанию — в mqtt_opts.h:

Исходящее: влезть в кольцо. Сразу после подключения прогон спрашивает у mqtt_publish, какое самое длинное тело в тему lab/f407-01/resp (16 символов) он примет с QoS 0. Ответ — 235 байт: пакет из 1 + 2 + 2 + 16 + 235 = 256 байт, ровно размер кольца. mqtt_publish вернул ERR_OK — а подписчик этого сообщения не получил, и в журнале провода его нет. Следующее, в 234 байта, дошло. Причина в кольце: при записи ровно 256 байт указатель записи догоняет указатель чтения, и кольцо выглядит пустым (198и.7). Практический вывод: самый длинный пакет должен быть хотя бы на байт меньше MQTT_OUTPUT_RINGBUF_SIZE.

Входящее: кусками по буферу приёма. Подписчик послал в cmd/echo 3 000 байт с QoS 1. Пакет занял 3 027 байт (длина остатка 3 024 — два байта, D0 17) и пришёл тремя сегментами TCP: 1 460, 1 460 и 107 байт. Но куски, которые видит приложение, режет не TCP, а буфер MQTT_VAR_HEADER_BUFFER_LEN: 25 кусков — 101 байт, 23 по 125 и последний 24. 128 − 3 байта заголовка = 125; в первом куске ещё тема lab/f407-01/cmd/echo (2 + 20 байт) и номер (2 байта): 125 − 24 = 101. Отсюда устройство обратного вызова: on_publish получает тему и полную длину, on_data — куски, пока не придёт флаг MQTT_DATA_FLAG_LAST:

/* ── входящая публикация: сначала тема и полная длина, потом данные кусками ── */
static void on_publish(void *arg, const char *topic, u32_t tot_len)
{
    (void)arg;
    rx_cmd = CMD_NONE;
    if (strncmp(topic, T_CMD_PFX, sizeof(T_CMD_PFX) - 1u) == 0) {
        const char *name = topic + sizeof(T_CMD_PFX) - 1u;
        rx_cmd = strcmp(name, "led") == 0 ? CMD_LED : strcmp(name, "echo") == 0 ? CMD_ECHO : CMD_NONE;
    }
    rx_got = 0;
    rx_frag = 0;
    app_log("rx topic=%s len=%lu", topic, (unsigned long)tot_len);
}

static void on_data(void *arg, const u8_t *data, u16_t len, u8_t flags)
{
    (void)arg;
    uint32_t have = rx_got < sizeof(rx_buf) - 1u ? rx_got : sizeof(rx_buf) - 1u;
    uint32_t room = sizeof(rx_buf) - 1u - have;     /* лишнее не храним, но считаем */
    memcpy(rx_buf + have, data, len < room ? len : room);
    rx_got += len;
    rx_frag++;
    app_log("frag %lu len=%u last=%u", (unsigned long)rx_frag, (unsigned)len, (unsigned)(flags & MQTT_DATA_FLAG_LAST));
    if (!(flags & MQTT_DATA_FLAG_LAST))
        return;                                     /* ждём последний кусок */
    rx_buf[rx_got < sizeof(rx_buf) - 1u ? rx_got : sizeof(rx_buf) - 1u] = 0;
    if (rx_cmd == CMD_LED) {
        led = strcmp(rx_buf, "on") == 0;
        respond(led ? "led=on" : "led=off", led ? 6u : 7u);
    } else if (rx_cmd == CMD_ECHO) {
        char r[16];
        memcpy(r, "got=", 4);
        respond(r, (u16_t)(4u + put_dec(r + 4, rx_got)));
    }
}

Тема в on_publish — указатель внутрь буфера приёма, живой только на время вызова: запомнить нужно разобранную команду, а не указатель. Тело команды — не строка (раздел 37.2): ноль в конце ставит сам on_data, а всё, что длиннее 31 байта, только считается. Ответ got=3000 ушёл через 4 мс после команды. Отвечать можно прямо из on_data: mqtt_publish кладёт пакет в кольцо и отдаёт TCP, ответа не ждёт. На F407 on_data — 276 байт кода, on_publish — 108.

198и.5 QoS 1, keep-alive и обнаружение обрыва

Когда lwIP шлёт PINGREQ. Таймер клиента тикает раз в 5 с. Счётчик тиков обнуляется, когда TCP подтвердил очередную передачу; PINGREQ уходит, когда счётчик × 5 с достигнет keep-alive. При 20 с это происходит через 20–25 с после последней передачи, смотря где оказался тик. В прогоне — через 21,7 и 20,8 с. Брокер ждёт полтора keep-alive — 30 с, так что запас есть, но небольшой: при keep-alive 10 с и шаге 5 с PINGREQ мог бы уйти только через 15 с — ровно тогда, когда брокер уже отключает. Поэтому в mqtt198j_app.c стоит проверка на этапе сборки:

#define PING_PERIOD_S ((KEEPALIVE_S + MQTT_CYCLIC_TIMER_INTERVAL - 1u) / MQTT_CYCLIC_TIMER_INTERVAL * MQTT_CYCLIC_TIMER_INTERVAL \
                       + MQTT_CYCLIC_TIMER_INTERVAL)
_Static_assert(PING_PERIOD_S < KEEPALIVE_S * 3u / 2u, "брокер отключит прибор раньше, чем lwIP пошлёт PINGREQ");

Обрыв кабеля. Кабель «выдернут» на 70,2 с прогона. Последним, что дошло до брокера, был PINGREQ на 60,8 с. Дальше молчание с обеих сторон:

Брокер — через полтора keep-alive, как и обещано. lwIP — по своей «сторожевой» проверке: если брокер молчит дольше 1,5 keep-alive, клиент закрывает соединение; с шагом 5 с это происходит на седьмом тике — через 30–35 с. TCP сам по себе обрыва ещё долго не заметил бы (раздел 106.6).

Что стало с QoS 1. На 80 с, уже в обрыве, прибор опубликовал измерение n=2 — пакет ушёл в пустоту. Закрывая соединение, lwIP стёр все ждущие запросы молча: обратный вызов meas_done не вызывается ни с ошибкой, ни как-то ещё. Приложение узнаёт об этом только в on_conn и само возвращает такие измерения в очередь (lost_in_flight=1 в журнале). После подключения n=2 и накопленное n=3 ушли первыми, и подписчик получил все пять измерений прогона ровно по одному разу; n=2 — с опозданием на 68,7 с.

QoS 1 — это «дойдёт хотя бы раз» в пределах одного соединения. Повтор после переподключения — забота прибора, потому что сессия чистая. Цена — возможный дубль, если PUBACK потерялся уже после того, как брокер принял сообщение: подписчик должен уметь отбросить повтор, для этого в теле измерения есть номер n.

198и.6 Повторное подключение с нарастающей паузой

Правило простое: после неудачи подождать, потом попробовать ещё раз, и каждый раз ждать вдвое дольше — 1, 2, 4 … 32 с, с добавкой до четверти паузы, чтобы сотня приборов после сбоя не ломилась к брокеру в одну и ту же секунду. После успеха пауза снова 1 с.

/* следующая пауза: 1, 2, 4 … 32 с плюс случайная добавка до четверти */
static uint32_t next_pause(void)
{
    uint32_t p = pause_ms;
    rnd ^= rnd << 13; rnd ^= rnd >> 17; rnd ^= rnd << 5;
    pause_ms = pause_ms * 2u > PAUSE_MAX_MS ? PAUSE_MAX_MS : pause_ms * 2u;
    return p + rnd % (p / 4u + 1u);
}

Два важных места. Повтор не запускается прямо из on_conn: lwIP вызывает его изнутри своих обработчиков TCP (через mqtt_close), и открывать новое соединение оттуда незачем — повтор ставится таймером sys_timeout, тем же, что отмеряет паузу. И CONNACK ждётся своим таймером 5 с: собственный срок lwIP MQTT_CONNECT_TIMOUT — 100 с, а пока кабеля нет, SYN просто теряется и TCP ничего не сообщает. По своему таймеру прибор зовёт mqtt_disconnect — on_conn после него не вызывается, повтор ставится сразу.

Два обрыва прогона ведут себя по-разному:

Задержка после восстановления — плата за нарастающую паузу: чем дольше лежал брокер, тем дольше прибор думает перед следующей попыткой. Потолок паузы (здесь 32 с) задаёт худший случай.

Штатное отключение. На 230,3 с прибор вызвал mqtt_disconnect, и через 1 мс подписчик получил offline — завещание. В журнале провода пакета DISCONNECT (E0 00) нет: lwIP 2.1.2 просто закрывает TCP, и для брокера это обрыв (connection closed by client). Если прибор уходит спать и не хочет пугать диспетчера, перед отключением он сам публикует в state что-то вроде sleep с сохранением.

198и.7 Грабли

Прибор завис, получив длинную команду. В lwIP 2.1.2 в mqtt_message_received стоит проверка msg_idx < MQTT_VAR_HEADER_BUFFER_LEN, а msg_idx считает все принятые байты пакета. Любая входящая публикация длиннее буфера приёма её нарушает. Прогон с проверками lwIP как есть: команда в 3 000 байт — и assert: client->msg_idx < MQTT_VAR_HEADER_BUFFER_LEN, процесс снят (код 134). На плате LWIP_PLATFORM_ASSERT из arch/cc.h — вечный цикл, прибор просто замер. Сам разбор кусками при этом верен (198и.4), неверна только проверка. Выход в опыте — собрать mqtt.c с -DLWIP_NOASSERT, остальной стек — с проверками. Другой выход — не принимать публикаций длиннее буфера: тема плюс тело меньше 128 байт.

mqtt_publish вернул ERR_OK, а сообщения нет. Пакет длиной ровно MQTT_OUTPUT_RINGBUF_SIZE проходит проверку места (<=), но после записи кольцо выглядит пустым — и пакет пропадает без следа. В прогоне — тело 235 байт при теме 16 символов. Держите самый длинный пакет хотя бы на байт меньше кольца или увеличьте кольцо.

После переподключения команды не приходят. mqtt_client_connect обнуляет структуру клиента, и обратные вызовы входящих публикаций пропадают. mqtt_set_inpub_callback и подписка — в on_conn, при каждом подключении.

Измерение потерялось при обрыве, хотя QoS 1. lwIP стирает ждущие запросы при закрытии соединения, не вызывая их обратных вызовов. Очередь неподтверждённого — в приложении, флаг «в полёте» сбрасывается в on_conn при разрыве.

Прибор «на связи» ещё полминуты после обрыва. Так и есть: и брокер, и lwIP узнают о молчании только через полтора keep-alive (в опыте 30,2 и 34,2 с). Быстрее — только меньший keep-alive, с проверкой из 198и.5.

Зависло на подключении при выдернутом кабеле. Срок CONNACK у lwIP — 100 с (MQTT_CONNECT_TIMOUT). Нужен свой таймер: в опыте 5 с.

Старое соединение шлёт пакеты после закрытия. mqtt_close закрывает TCP, но неподтверждённые данные остаются у стека: через 34 с после восстановления кабеля lwIP повторил по старому соединению измерение n=2 и PINGREQ, брокер ответил RST. Вреда нет — брокер это соединение уже забыл, — но в перехвате трафика не пугайтесь «лишнего» PUBLISH.

Все приборы переподключаются в одну секунду. Без случайной добавки к паузе после перезапуска брокера парк приборов стучится синхронно. Добавка до четверти паузы разводит их.

198и.8 Итоги

• Клиент MQTT из lwIP — четыре вызова и три обратных вызова, без ОС; структура клиента статическая, 496 байт на F407. Весь образ с lwIP — 29 780 байт Flash и 28 924 байта ОЗУ, из них mqtt.c — 3 514 байт кода.
• Свой кодировщик совпал с тем, что lwIP отправил брокеру: 15 разных пакетов из 23, байт в байт; CONNECT — 49 байт с флагами 0x2E.
• Кольцо передачи 256 байт: тело 235 байт (пакет 256) принимается и пропадает, 234 — доходит.
• Команда в 3 000 байт пришла 25 кусками: 101, 23 × 125, 24 — по буферу приёма 128 байт, а не по сегментам TCP. С проверками lwIP 2.1.2 она же роняет стек.
• PINGREQ — через 20,8–21,7 с тишины при keep-alive 20 с. Обрыв кабеля: завещание через 20,8 с, lwIP сдался через 24,8 с.
• QoS 1 держится только внутри соединения: измерение, ушедшее в обрыв, lwIP стёр молча, приложение повторило его и доставило через 68,7 с.
• Нарастающая пауза: 4 неудачи и подключение через 8,5 с после восстановления кабеля и через 10,5 с после запуска брокера.
• mqtt_disconnect не шлёт DISCONNECT: подписчик получает завещание через 1 мс.

Полные исходники

mqtt198j_app.c — MQTT-клиент прибора поверх lwIP raw API.

/* 198и: MQTT-клиент прибора поверх lwIP raw API (NO_SYS). Один и тот же файл
   собирается armclang для F407 и gcc для прогона на ПК с настоящим брокером. */
#include <stdint.h>
#include <string.h>
#include "lwip/apps/mqtt.h"
#include "lwip/apps/mqtt_priv.h"        /* описание mqtt_client_t: клиент статический, не из кучи */
#include "lwip/timeouts.h"
#include "mqtt198j_app.h"

/* PINGREQ у lwIP уходит, когда шаг таймера насчитал keep_alive без передачи:
   через keep_alive + шаг (округлённо вверх до шага). Брокер ждёт 1,5 keep_alive. */
#define PING_PERIOD_S ((KEEPALIVE_S + MQTT_CYCLIC_TIMER_INTERVAL - 1u) / MQTT_CYCLIC_TIMER_INTERVAL * MQTT_CYCLIC_TIMER_INTERVAL \
                       + MQTT_CYCLIC_TIMER_INTERVAL)
_Static_assert(PING_PERIOD_S < KEEPALIVE_S * 3u / 2u, "брокер отключит прибор раньше, чем lwIP пошлёт PINGREQ");

/* CONNECT с завещанием целиком помещается в кольцо передачи */
#define CONNECT_LEN (2u + 10u + 2u + sizeof(DEV_ID) - 1u + 2u + sizeof(T_STATE) - 1u + 2u + sizeof("offline") - 1u)
_Static_assert(CONNECT_LEN <= MQTT_OUTPUT_RINGBUF_SIZE, "CONNECT не влезает в MQTT_OUTPUT_RINGBUF_SIZE");
_Static_assert(sizeof(T_STATE) - 1u <= 255u, "lwIP хранит длину темы завещания в одном байте");

enum { CMD_NONE, CMD_LED, CMD_ECHO };

struct outbox {                          /* измерение, пока брокер не ответил PUBACK */
    uint32_t n;
    uint8_t used, in_flight;
};

static mqtt_client_t cli;
static ip_addr_t broker_ip;
static const struct mqtt_connect_client_info_t ci = {
    .client_id = DEV_ID, .keep_alive = KEEPALIVE_S,
    .will_topic = T_STATE, .will_msg = "offline", .will_qos = 1, .will_retain = 1,
};
static struct outbox box[OUTBOX_N];
static uint32_t meas_n, pause_ms = PAUSE_MIN_MS, tries, rnd = 0x198u;
static uint8_t rx_cmd, led;
static char rx_buf[32];
static uint32_t rx_got, rx_frag;

static void try_connect(void *arg);

static uint32_t put_dec(char *p, uint32_t v)
{
    char t[10];
    uint32_t k = 0, i = 0;
    do { t[k++] = (char)('0' + v % 10u); v /= 10u; } while (v);
    while (k) p[i++] = t[--k];
    return i;
}

/* следующая пауза: 1, 2, 4 … 32 с плюс случайная добавка до четверти */
static uint32_t next_pause(void)
{
    uint32_t p = pause_ms;
    rnd ^= rnd << 13; rnd ^= rnd >> 17; rnd ^= rnd << 5;
    pause_ms = pause_ms * 2u > PAUSE_MAX_MS ? PAUSE_MAX_MS : pause_ms * 2u;
    return p + rnd % (p / 4u + 1u);
}

static void schedule_retry(void)
{
    uint32_t p = next_pause();
    sys_untimeout(try_connect, NULL);
    sys_timeout(p, try_connect, NULL);
    app_log("retry pause=%lu", (unsigned long)p);
}

static void meas_done(void *arg, err_t err)
{
    struct outbox *b = arg;
    if (err == ERR_OK) {
        app_log("puback meas n=%lu", (unsigned long)b->n);
        b->used = 0;
    } else {
        app_log("timeout meas n=%lu err=%d", (unsigned long)b->n, err);
    }
    b->in_flight = 0;
}

static void send_outbox(void)
{
    char pl[24];
    for (uint32_t i = 0; i < OUTBOX_N; i++) {
        struct outbox *b = &box[i];
        if (!b->used || b->in_flight)
            continue;
        uint32_t k = 0;
        memcpy(pl, "n=", 2); k = 2u + put_dec(pl + 2, b->n);
        memcpy(pl + k, " t=23.", 6); k += 6u; k += put_dec(pl + k, b->n % 10u);
        err_t e = mqtt_publish(&cli, T_MEAS, pl, (u16_t)k, 1, 0, meas_done, b);
        app_log("pub meas n=%lu id=%u err=%d", (unsigned long)b->n, (unsigned)cli.pkt_id_seq, e);
        if (e == ERR_OK)
            b->in_flight = 1;
    }
}

static void meas_tick(void *arg)
{
    (void)arg;
    struct outbox *slot = NULL;
    meas_n++;
    for (uint32_t i = 0; i < OUTBOX_N && !slot; i++)
        if (!box[i].used) slot = &box[i];
    if (!slot) {                                    /* очередь полна: теряем самое старое */
        slot = &box[0];
        for (uint32_t i = 1; i < OUTBOX_N; i++)
            if (box[i].n < slot->n) slot = &box[i];
        app_log("outbox full, drop n=%lu", (unsigned long)slot->n);
    }
    slot->n = meas_n; slot->used = 1; slot->in_flight = 0;
    if (mqtt_client_is_connected(&cli))
        send_outbox();
    sys_timeout(MEAS_MS, meas_tick, NULL);
}

static void respond(const char *s, u16_t len)
{
    err_t e = mqtt_publish(&cli, T_RESP, s, len, 0, 0, NULL, NULL);
    app_log("pub resp len=%u err=%d", (unsigned)len, e);
}

/* ── входящая публикация: сначала тема и полная длина, потом данные кусками ── */
static void on_publish(void *arg, const char *topic, u32_t tot_len)
{
    (void)arg;
    rx_cmd = CMD_NONE;
    if (strncmp(topic, T_CMD_PFX, sizeof(T_CMD_PFX) - 1u) == 0) {
        const char *name = topic + sizeof(T_CMD_PFX) - 1u;
        rx_cmd = strcmp(name, "led") == 0 ? CMD_LED : strcmp(name, "echo") == 0 ? CMD_ECHO : CMD_NONE;
    }
    rx_got = 0;
    rx_frag = 0;
    app_log("rx topic=%s len=%lu", topic, (unsigned long)tot_len);
}

static void on_data(void *arg, const u8_t *data, u16_t len, u8_t flags)
{
    (void)arg;
    uint32_t have = rx_got < sizeof(rx_buf) - 1u ? rx_got : sizeof(rx_buf) - 1u;
    uint32_t room = sizeof(rx_buf) - 1u - have;     /* лишнее не храним, но считаем */
    memcpy(rx_buf + have, data, len < room ? len : room);
    rx_got += len;
    rx_frag++;
    app_log("frag %lu len=%u last=%u", (unsigned long)rx_frag, (unsigned)len, (unsigned)(flags & MQTT_DATA_FLAG_LAST));
    if (!(flags & MQTT_DATA_FLAG_LAST))
        return;                                     /* ждём последний кусок */
    rx_buf[rx_got < sizeof(rx_buf) - 1u ? rx_got : sizeof(rx_buf) - 1u] = 0;
    if (rx_cmd == CMD_LED) {
        led = strcmp(rx_buf, "on") == 0;
        respond(led ? "led=on" : "led=off", led ? 6u : 7u);
    } else if (rx_cmd == CMD_ECHO) {
        char r[16];
        memcpy(r, "got=", 4);
        respond(r, (u16_t)(4u + put_dec(r + 4, rx_got)));
    }
}

static void on_sub(void *arg, err_t err)
{
    (void)arg;
    app_log("suback err=%d", err);
}

static void on_state(void *arg, err_t err)
{
    (void)arg;
    app_log("puback state err=%d", err);
}

static void connack_wait_over(void *arg)
{
    (void)arg;
    app_log("no connack in %lu ms", (unsigned long)CONNACK_MS);
    mqtt_disconnect(&cli);                          /* обратного вызова не будет — повтор сами */
    schedule_retry();
}

static void on_conn(mqtt_client_t *c, void *arg, mqtt_connection_status_t st)
{
    (void)arg;
    sys_untimeout(connack_wait_over, NULL);
    if (st == MQTT_CONNECT_ACCEPTED) {
        app_log("connected try=%lu", (unsigned long)tries);
        tries = 0;
        pause_ms = PAUSE_MIN_MS;
        mqtt_set_inpub_callback(c, on_publish, on_data, NULL);
        mqtt_subscribe(c, T_CMD_SUB, 1, on_sub, NULL);
        mqtt_publish(c, T_STATE, "online", 6, 1, 1, on_state, NULL);
        send_outbox();                              /* то, что не дошло до обрыва */
        return;
    }
    uint32_t lost = 0;
    for (uint32_t i = 0; i < OUTBOX_N; i++) {       /* lwIP стёр запросы молча: вернуть их в очередь */
        lost += box[i].in_flight;
        box[i].in_flight = 0;
    }
    app_log("status %d lost_in_flight=%lu", (int)st, (unsigned long)lost);
    schedule_retry();                               /* не из обратного вызова и не сразу */
}

static void try_connect(void *arg)
{
    (void)arg;
    tries++;
    err_t e = mqtt_client_connect(&cli, &broker_ip, MQTT_PORT, on_conn, NULL, &ci);
    app_log("connect try=%lu err=%d", (unsigned long)tries, e);
    if (e != ERR_OK) {
        schedule_retry();
        return;
    }
    sys_timeout(CONNACK_MS, connack_wait_over, NULL);
}

mqtt_client_t *app_client(void)
{
    return &cli;
}

void app_start(const ip_addr_t *broker)
{
    ip_addr_copy(broker_ip, *broker);
    sys_timeout(MEAS_MS, meas_tick, NULL);
    try_connect(NULL);
}

Дальше: раздел 198к «Мини-проект: осциллограф на плате — АЦП, DMA, запуск по фронту и экран на ПК».