198и. An MQTT client on the STM32: publish, subscribe and reconnect

In section 198з the F407 device was the server itself: the browser came to it for the page and the data. For a fleet of devices the reverse scheme is more convenient: the device connects to a broker on its own, publishes measurements and listens for commands (section 188.4). Here this is done with the MQTT client that ships with lwIP: src/apps/mqtt/mqtt.c, raw API, no OS.

What MQTT is, what a packet consists of, and why QoS and the last will exist has already been covered in sections 37.2, 87.5 and 188.4, and the ESP-IDF client in 156.6. This section is about something else: how lwIP specifically uses all this, what it actually puts on the wire, where its buffers run out, and how the device survives a pulled cable and a broker going down. Everything is tested against a real Mosquitto broker, with the timings taken from the run.

Sources: lwIP 2.1.2 from STM32CubeF4 1.28.0 (mqtt.c, mqtt.h, mqtt_priv.h, mqtt_opts.h), the Ethernet driver and network interface from sections 106а and 106в, the log of the eclipse-mosquitto:2 broker. Files are in _verify/mqtt198j/: mqtt198j_app.c (the device client, shared by the board and the PC), host198j.c (the PC run), mqtt_frame.c (our own encoder), sub198j.py (the subscriber), run198j.py (the scenario), build_f407.py (the armclang build). The check is _verify/mqtt198j_check.py.

Terms used in the documentation

Term	Meaning
keep alive	the "I am alive" interval: how many seconds the device may stay silent
Last Will and Testament (LWT)	the last will: what the broker publishes if the device disappears
retained message	a stored message: the broker hands it to every new subscriber
packet identifier	the packet number: links a QoS 1 PUBLISH to its PUBACK
remaining length	the length of the rest: 1 to 4 bytes of 7 bits each
clean session	the broker keeps nothing between connections
output ring buffer	the transmit ring of the lwIP client
exponential backoff	retrying with a growing pause

198и.1 The device, the topics and the test bench

The device f407-01 does four things:

• every 30 s it publishes a measurement to lab/f407-01/meas with QoS 1;
• it is subscribed to lab/f407-01/cmd/+: the led command turns on the LED, echo replies with the length of what it received;
• it sends replies to lab/f407-01/resp with QoS 0;
• in lab/f407-01/state it keeps online with the retain flag, and when connecting it leaves the broker the last will offline, also retained.

Topics go from general to specific: site, device, kind of data. Then a dispatcher subscribes to lab/# and sees everything, while the device subscribes only to its own commands through +. The device name in the topic matches its client identifier: the broker log shows at once who is who.

There is no board with Ethernet in this experiment, and QEMU does not model the F407 Ethernet. So the same file mqtt198j_app.c is built for Linux together with lwIP 2.1.2 and connected to the kernel through a TAP device, as in section 106в. The client container lives in the network namespace of the broker container: the kernel with the address 10.198.0.1 is the broker, lwIP is 10.198.0.2. The subscriber is a Python program on a bare socket, without MQTT libraries. Every frame that lwIP sent to TAP or received from it is written to the "wire" log with a time in milliseconds. The file /tmp/cut in the client container means "cable pulled": frames in both directions are lost, while the link stays up for lwIP, as with a real cable pulled out of a switch somewhere further down the line. On the PC the measurement step is 40 s instead of 30, so that a PINGREQ is visible between measurements.

198и.2 The lwIP client: four calls and the callbacks

The whole client is four functions and three callbacks:

on_conn is called both on success (MQTT_CONNECT_ACCEPTED) and on a break: 256 is MQTT_CONNECT_DISCONNECTED (TCP closed or refused), 257 is MQTT_CONNECT_TIMEOUT (the broker went silent). Both occurred in the run.

Three details are visible only in the source code:

• mqtt_client_connect starts with a memset of the whole client structure. The incoming-publication callbacks are wiped together with it, so mqtt_set_inpub_callback and mqtt_subscribe are called in on_conn after every connection, not once at startup. Packet numbers also start over: in the run, after each reconnection the subscription gets number 1, online gets 2, the first measurement gets 3.
• The client can be kept static. mqtt_client_new() takes memory from the lwIP heap, while the definition of mqtt_client_t is in mqtt_priv.h: static mqtt_client_t cli; is enough. On the F407 it takes 496 bytes, on a 64-bit PC 592 (pointers are longer).
• The session is always clean. lwIP sets the clean session flag itself and does not let you clear it. So after a break the broker keeps nothing for the device: the device itself restores the subscription and the unacknowledged measurements.

The F407 build (build_f407.py, armclang -Os, our own files with -Wall -Wextra -Werror) has no warnings. The whole image is 29780 bytes of Flash and 28924 bytes of RAM. Of that, lwIP with the MQTT client is 25566 bytes of code and 17322 bytes of RAM (a 6 KB heap and the pools), mqtt.c itself is 3514 bytes of code, the device application is 1820 bytes of code and 589 bytes of RAM. Two build-time checks are in 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 хранит длину темы завещания в одном байте");

The second one reflects a real limitation of lwIP 2.1.2: it checks the lengths of the last-will topic and message against 255 bytes and stores them in a u8_t, although the protocol allows up to 65535.

198и.3 Packets byte by byte: our own encoder against the wire

An MQTT 3.1.1 encoder was already written in section 87.5. Here mqtt_frame.c is extended with what the device needs: CONNECT with a last will, PUBLISH with QoS 1 and a packet number, SUBSCRIBE, PUBACK, PINGREQ. The comparison works like this: the log parser assembles the TCP stream of each connection (retransmissions are dropped by sequence numbers), cuts it into MQTT packets and, for each outgoing packet, writes a request: "the device's CONNECT", "PUBLISH QoS 1 retain 0 number 3 to such-and-such topic with such-and-such body". Our encoder builds the packet from the request, and the bytes are compared.

During the run lwIP sent 23 MQTT packets, 15 of them distinct, and all 15 matched our encoder byte for byte. The check builds them once more with a Python encoder, also a match. The most telling ones:

The broker log repeated the same fields: p4, c1, k20 means protocol 4, clean session, keep-alive 20; Will message specified (7 bytes) (r1, q1).

lwIP acknowledges an incoming QoS 1 publication on its own: a PUBACK with the number assigned by the broker (40 02 00 01) goes out after the last chunk of data. The application has nothing to do here, and it cannot influence this either.

198и.4 The client buffers and an incoming publication in chunks

The client sizes are set in lwipopts.h, the defaults are in mqtt_opts.h:

Outgoing: fit into the ring. Right after connecting, the run asks mqtt_publish what the longest body for the topic lab/f407-01/resp (16 characters) it will accept with QoS 0 is. The answer is 235 bytes: a packet of 1 + 2 + 2 + 16 + 235 = 256 bytes, exactly the size of the ring. mqtt_publish returned ERR_OK, yet the subscriber never got this message, and it is not in the wire log. The next one, 234 bytes, arrived. The reason is the ring: when exactly 256 bytes are written, the write pointer catches up with the read pointer, and the ring looks empty (198и.7). The practical conclusion: the longest packet must be at least one byte smaller than MQTT_OUTPUT_RINGBUF_SIZE.

Incoming: in chunks of the receive buffer. The subscriber sent 3000 bytes to cmd/echo with QoS 1. The packet took 3027 bytes (remaining length 3024, two bytes, D0 17) and arrived in three TCP segments: 1460, 1460 and 107 bytes. But the chunks the application sees are cut not by TCP but by the MQTT_VAR_HEADER_BUFFER_LEN buffer: 25 chunks, 101 bytes, 23 of 125 and the last one 24. 128 − 3 header bytes = 125; the first chunk also holds the topic lab/f407-01/cmd/echo (2 + 20 bytes) and the number (2 bytes): 125 − 24 = 101. Hence the shape of the callback: on_publish gets the topic and the full length, on_data gets the chunks until the MQTT_DATA_FLAG_LAST flag arrives:

/* ── входящая публикация: сначала тема и полная длина, потом данные кусками ── */
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)));
    }
}

The topic in on_publish is a pointer into the receive buffer, valid only for the duration of the call: remember the parsed command, not the pointer. The command body is not a string (section 37.2): on_data adds the terminating zero itself, and anything longer than 31 bytes is only counted. The reply got=3000 went out 4 ms after the command. You can reply directly from on_data: mqtt_publish puts the packet into the ring and hands it to TCP, it does not wait for a reply. On the F407 on_data is 276 bytes of code, on_publish is 108.

198и.5 QoS 1, keep-alive and detecting a break

When lwIP sends PINGREQ. The client timer ticks every 5 s. The tick counter is reset when TCP acknowledges the latest transmission; PINGREQ goes out when counter × 5 s reaches the keep-alive. With 20 s this happens 20–25 s after the last transmission, depending on where the tick falls. In the run, after 21.7 and 20.8 s. The broker waits one and a half keep-alives, 30 s, so there is a margin, but a small one: with a keep-alive of 10 s and a step of 5 s PINGREQ could go out only after 15 s, exactly when the broker is already disconnecting. That is why mqtt198j_app.c has a build-time check:

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

A pulled cable. The cable is "pulled" at 70.2 s into the run. The last thing that reached the broker was a PINGREQ at 60.8 s. After that, silence on both sides:

The broker reacts after one and a half keep-alives, as promised. lwIP reacts through its own "watchdog" check: if the broker is silent for longer than 1.5 keep-alive, the client closes the connection; with a 5 s step this happens on the seventh tick, after 30–35 s. TCP by itself would not notice the break for a long time (section 106.6).

What happened to QoS 1. At 80 s, already during the break, the device published measurement n=2, and the packet went into the void. When closing the connection, lwIP wiped all pending requests silently: the meas_done callback is not called, neither with an error nor in any other way. The application learns about it only in on_conn and puts such measurements back in the queue itself (lost_in_flight=1 in the log). After reconnecting, n=2 and the accumulated n=3 went out first, and the subscriber received all five measurements of the run exactly once each; n=2 arrived late by 68.7 s.

QoS 1 means "will arrive at least once" within one connection. Retrying after a reconnection is the device's job, because the session is clean. The price is a possible duplicate if the PUBACK was lost after the broker had already accepted the message: the subscriber must be able to discard the repeat, and for that the measurement body contains the number n.

198и.6 Reconnecting with a growing pause

The rule is simple: after a failure, wait, then try again, and each time wait twice as long: 1, 2, 4 … 32 s, plus up to a quarter of the pause on top, so that a hundred devices do not storm the broker in the same second after a failure. After a success the pause is 1 s again.

/* следующая пауза: 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);
}

Two important points. The retry is not started directly from on_conn: lwIP calls it from inside its TCP handlers (through mqtt_close), and there is no reason to open a new connection from there. The retry is scheduled by the sys_timeout timer, the same one that measures the pause. And CONNACK is awaited with our own 5 s timer: lwIP's own deadline MQTT_CONNECT_TIMOUT is 100 s, and while the cable is out, the SYN is simply lost and TCP reports nothing. When its own timer fires, the device calls mqtt_disconnect; on_conn is not called after it, so the retry is scheduled immediately.

The two breaks in the run behave differently:

The delay after recovery is the price of the growing pause: the longer the broker was down, the longer the device thinks before the next attempt. The pause ceiling (32 s here) sets the worst case.

A normal disconnect. At 230.3 s the device called mqtt_disconnect, and 1 ms later the subscriber received offline, the last will. There is no DISCONNECT packet (E0 00) in the wire log: lwIP 2.1.2 simply closes TCP, and for the broker this is a break (connection closed by client). If the device goes to sleep and does not want to alarm the dispatcher, it publishes something like sleep with retain to state itself before disconnecting.

198и.7 Pitfalls

The device hung after receiving a long command. In lwIP 2.1.2, mqtt_message_received contains the check msg_idx < MQTT_VAR_HEADER_BUFFER_LEN, while msg_idx counts all received bytes of the packet. Any incoming publication longer than the receive buffer violates it. A run with the lwIP checks as they are: a 3000-byte command, and assert: client->msg_idx < MQTT_VAR_HEADER_BUFFER_LEN, the process is killed (code 134). On the board LWIP_PLATFORM_ASSERT from arch/cc.h is an endless loop, and the device simply freezes. The chunked parsing itself is correct (198и.4); only the check is wrong. The way out in the experiment is to build mqtt.c with -DLWIP_NOASSERT and the rest of the stack with the checks. Another way is not to accept publications longer than the buffer: topic plus body under 128 bytes.

mqtt_publish returned ERR_OK, but the message is missing. A packet exactly MQTT_OUTPUT_RINGBUF_SIZE long passes the space check (<=), but after it is written the ring looks empty, and the packet vanishes without a trace. In the run: a body of 235 bytes with a 16-character topic. Keep the longest packet at least one byte smaller than the ring, or enlarge the ring.

Commands stop arriving after a reconnection. mqtt_client_connect zeroes the client structure, and the incoming-publication callbacks are lost. mqtt_set_inpub_callback and the subscription belong in on_conn, on every connection.

A measurement was lost during a break, despite QoS 1. lwIP wipes pending requests when closing a connection without calling their callbacks. The queue of unacknowledged data lives in the application, and the "in flight" flag is cleared in on_conn on a break.

The device is "online" for another half minute after a break. That is how it is: both the broker and lwIP learn about the silence only after one and a half keep-alives (30.2 and 34.2 s in the experiment). Faster only with a smaller keep-alive, with the check from 198и.5.

Stuck connecting with the cable pulled. The lwIP CONNACK deadline is 100 s (MQTT_CONNECT_TIMOUT). You need your own timer: 5 s in the experiment.

The old connection sends packets after being closed. mqtt_close closes TCP, but unacknowledged data stay with the stack: 34 s after the cable was restored, lwIP retransmitted measurement n=2 and a PINGREQ over the old connection, and the broker answered with RST. No harm done, since the broker has already forgotten that connection, but do not be alarmed by an "extra" PUBLISH in a traffic capture.

All devices reconnect in the same second. Without a random addition to the pause, a fleet of devices knocks on the broker in sync after a broker restart. An addition of up to a quarter of the pause spreads them out.

198и.8 Summary

• The lwIP MQTT client is four calls and three callbacks, without an OS; the client structure is static, 496 bytes on the F407. The whole image with lwIP is 29780 bytes of Flash and 28924 bytes of RAM, of which mqtt.c is 3514 bytes of code.
• Our own encoder matched what lwIP sent to the broker: 15 distinct packets out of 23, byte for byte; CONNECT is 49 bytes with flags 0x2E.
• A 256-byte transmit ring: a 235-byte body (a 256-byte packet) is accepted and vanishes, 234 gets through.
• A 3000-byte command arrived in 25 chunks: 101, 23 × 125, 24, cut by the 128-byte receive buffer, not by TCP segments. With the lwIP 2.1.2 checks enabled the same command crashes the stack.
• PINGREQ after 20.8–21.7 s of silence with a keep-alive of 20 s. Pulled cable: the last will after 20.8 s, lwIP gave up after 24.8 s.
• QoS 1 holds only within a connection: lwIP silently wiped the measurement sent into the break, and the application repeated it and delivered it 68.7 s late.
• Growing pause: 4 failures and a connection 8.5 s after the cable was restored and 10.5 s after the broker was started.
• mqtt_disconnect does not send DISCONNECT: the subscriber gets the last will after 1 ms.

Full sources

mqtt198j_app.c — the device MQTT client on top of the 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);
}

Next: section 198к "Mini-project: an oscilloscope on the board — ADC, DMA, edge trigger and a display on the PC".