Building a Non‑Blocking MQTT Client for Reactive Embedded Systems

MQTT‑Reactive is a lightweight, non‑blocking MQTT v3.1.1 client built on LiamBindle’s MQTT‑C library. Designed for reactive embedded systems, it delivers a portable, event‑driven communication stack that can be integrated into any C‑based application.

In this article we first define what constitutes a reactive system, then outline a software architecture that embraces concurrency and state‑based behavior. Finally, we walk through a concrete example—an IoT device for a railway company—showing how to employ MQTT‑Reactive via a state machine and event‑driven paradigm.

Reactive systems respond to internal or external events, process the event, and return to an idle state awaiting the next trigger. This event‑driven paradigm yields predictable, maintainable code, especially when multiple active objects (concurrent units) handle distinct event streams.

Our case study device performed the following tasks:

1. Detected and logged changes in several digital inputs
2. Acquired, filtered, and stored multiple analog signals
3. Periodically transmitted stored data to a remote server
4. Exchanged data with a broker using MQTT over GSM

MQTT was chosen for its lightweight, publish/subscribe model, ideal for low‑bandwidth, high‑latency links such as GSM.

Because the device’s firmware was built as a reactive system, the original MQTT‑C library required adaptation to communicate asynchronously via events. The resulting library—MQTT‑Reactive—exposes a clean API that can be invoked from a state machine while keeping the rest of the system decoupled.

State machine

MQTT‑Reactive is driven by the MqttMgr active object, whose behavior is described by the state machine in Figure 1. The machine starts in WaitingForNetConnection; once a network link is available, it transitions to WaitingForSync, where CONNECT and PUBLISH packets can be staged. Events that arrive in Sync are deferred until the system re‑enters a state that accepts them.

Every SyncTime milliseconds the machine moves to the Sync composite state, which posts Receive and Send events to the network manager. Though the example covers CONNECT and PUBLISH, extending support to SUBSCRIBE is straightforward.

The state machine’s actions use the params keyword to read event data and the GEN() macro to generate outgoing events. For example:

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

In this transition, the Connect event carries clientId and keepAlive parameters that are applied to the MqttMgr object’s attributes. Likewise, events are dispatched to other actors via:

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

The default action sets a callback that MQTT‑Reactive invokes when the broker acknowledges a CONNECT. The callback must generate either ConnAccepted or ConnRefused events:

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

Figure 1. State machine of an MQTT‑Reactive client (Source: VortexMakes)

Model implementation

The state machine can be coded in plain C or C++ using any of the following tools: RKH framework, QP framework, Yakindu Statechart Tool, or Rational Rhapsody Developer. All support Statecharts and code generation for C/C++.

The MqttMgr active object encapsulates all MQTT‑Reactive calls, shielding the rest of the system from direct API usage. Concurrency is achieved by exchanging events, which eliminates the pitfalls of semaphores, mutexes, delays, or event‑flags.

Attributes of MqttMgr are stored directly in its structure, e.g.:

struct MqttMgr {
    struct mqtt_client client;  // MQTT client instance
    LocalRecvAll localRecv;     // local receive buffer
    /* additional data items */
};

These members are accessed via the me pointer:

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

Table 1. MqttMgr attributes

The overall system includes three key actors: Collector (publishes data), NetMgr (manages the network), and MqttMgr (manages MQTT).

Figure 2. Draft of IoT system structure (Source: VortexMakes)

The sequence diagram in Figure 3 illustrates the flow when MqttMgr initiates a session with an MQTT broker.

Figure 3. Connecting to an MQTT broker (Source: VortexMakes)

Upon successful network establishment, Collector sends Connect(clientId, keepAlive) to MqttMgr. The broker may reply with ConnAccepted or ConnRefused, the latter carrying a rejection code as per MQTT v3.1.1 §3.2.2.3.

Figure 4. Broker rejects a connection request (Source: VortexMakes)

Publishing follows a similar pattern. Collector issues Publish(data, size, topic, qos) events. The data payload, typically JSON, is assembled using the lightweight jWrite library.

Figure 5. Publishing data to a broker (Source: VortexMakes)

Network failures are signaled by NetMgr through ReceiveFail or SendFail events, enabling MqttMgr to react accordingly.

Figure 6. Failures in network (Source: VortexMakes)

Table 2. Events involved in the scenarios

Conclusion

By encapsulating MQTT‑Reactive within an active object and orchestrating communication via events, the approach eliminates blocking constructs and yields a clean, testable design. The state‑machine model scales to multiple concurrent tasks, making it well‑suited for modern reactive embedded applications that require reliable MQTT connectivity.