rumqtt

The Router responds to several key event types to manage client state:

New Connection (Event::Connect)

When a new connection is formed:

• The router validates if the total connection limit has been reached.
• If a clean session is NOT requested, the router retrieves any pending data and acks from the previous session.
• A connack is sent back to the client immediately (bypassing the readyqueue for efficiency).

Device Data (Event::DeviceData)

When a RemoteLink reads bytes from the Network, it notifies the router. The router parses these as MQTT packets:

• Publish Packets: Appended to the CommitLog, a puback is added to the ackslog, and the connection is added to Router::readyqueue to flush the ack.
• Subscribe Packets: The router validates the subscription and calls Router::prepare_consumption(), which adds the connection to the readyqueue.
• PubAck Packets: Registered with the QoS1 buffer in the Connection. Out-of-order or unsolicited acks trigger a disconnection.

Disconnect Event

When a client disconnects, Router::handle_disconnection() is called. The connection and its ackslog entries are removed. If a clean session was not requested, the Tracker is saved to a 'graveyard' to preserve state.

The rumqttc event loop (accessed via connection.iter() in sync or eventloop.poll() in async) is the core engine that orchestrates outgoing and incoming packets.

Key behaviors:

• Connection Management: It handles pings, detects half-open connections, and performs automatic reconnections as long as you continue polling the loop.
• Flow Control: It provides queue size based flow control on outgoing packets and natural backpressure to client APIs during network instability.
• Concurrency: It manages the state and handles packets concurrently.
• Customization: Because the loop is externally polled, users can intercept notifications to distribute messages by topic, stop the loop for graceful shutdown, or access internal state to modify options before a reconnection.

Critical Requirement: You must loop on connection.iter() or eventloop.poll(). If you stop polling or block inside this loop, the connection will not progress and will eventually fail.

The rumqttc event loop manages reconnections based on configurable options. You can choose between automatic reconnection or manual control. If you choose manual control, you can capture the MqttState returned by the event loop and pass it to a new connection to maintain state.

Available reconnection options include:

• Reconnect::Never: Disables automatic reconnection. Use this if you want to handle reconnection logic manually using the returned MqttState.
• Reconnect::Automatic: Enables the event loop to handle reconnections automatically.

// create an eventloop after the initial mqtt connection is successful
let eventloop = connect(mqttoptions) -> Result<EventLoop, Error>

// during intermittent reconnetions due to bad network, eventloop will
// behave as per configured reconnection options to the eventloop
let stream = eventloop.assemble(reconnection_options, inputs);

The rumqttc client library is designed around two primary components: the EventLoop and the AsyncClient.

• EventLoop: This is the core component that manages network interaction. It handles TCP/WebSockets connections (with or without TLS) and makes critical decisions such as acknowledging QoS 1/2 requests and receiving publishes. To keep the client operational, you must continuously poll the .poll() method on the EventLoop.
• AsyncClient: This is the interface used by your user application to send requests to the EventLoop. It communicates with the EventLoop via an internal channel.

Critical Requirement: You must ensure .poll() is called regularly. If the EventLoop is blocked or not polled frequently enough, requests from both the user and the broker will fail to process, which can lead to severe connection issues requiring a hard reset.

      ┌──────┐
      │Broker│
      └──┬▲──┘
         ││
      sub││pub (network)
         ││
     ┌───▼┴────┐                ┌───────────┐
     │EventLoop◄────────────────┤AsyncClient│
     └───┬─────┘                └─────▲─────┘
         └──────────────┐┌────────────┘
             .poll()    ││  - .ack()
                        ││  - .publish()
                        ││  - .subscribe()
                        ││  - .unsubscribe()
                ┌───────▼┴───────┐
                │User Application│
                └────────────────┘

To work with rumqttd, you should understand its three primary architectural abstractions: the Broker, the Router, and Links.

• Broker: The top-level entity that manages the lifecycle of the system. When you create a Broker, it automatically creates a Router. Starting a Broker spawns several background threads for the Router, metrics server, MQTT (v4/v5) servers, WebSocket servers (if enabled), Prometheus listener, and the ConsoleLink.
• Router: The central engine that controls data flow. It uses a reactor pattern to react to events in the input buffers (ibufs). It manages authorization, scheduling, and dispatching data between components by writing to output buffers (obufs).
• Link: Asynchronous, typically network-facing entities that facilitate communication between the Router and the outside world (e.g., devices, metrics consumers, or management consoles).