MQTT and IoT

Why this matters

A flood-watch mission might involve three drones, ten water-level sensors, a ground station, a command-centre dashboard and a database. If every device connects point to point with every other, the number of connections grows fast, and each new device means changing every machine. MQTT solves this with a middleman that receives and distributes messages. UAT 314 module 1 introduced MQTT; this module goes deep enough to design and configure a real system.

IoT architecture

The Internet of Things (IoT) is a system in which many devices measure, send data and receive commands over a network. It usually has four layers.

End devices often have little power and computing capacity. A gateway collects data from devices on low-power long-range radio such as LoRaWAN and forwards it into the IP network. An example in the drone knowledge hub is ChirpStack, a LoRaWAN network server that publishes every device message as JSON over MQTT, on topics such as application/APPLICATION_ID/device/DEV_EUI/event/up.

Publish/subscribe and topics

In MQTT a sender (publisher) sends messages to a topic on a middleman called the broker, and a receiver (subscriber) tells the broker which topics it wants. The broker forwards each message to the matching subscribers. Publishers and subscribers do not need to know each other or be online at the same time, and new subscribers can be added without changing the publishers.

A topic is a multi-level string separated by /, like a file path. Subscribers can use two wildcards, as defined in MQTT 5.0:

• + matches exactly one level: drone/+/battery matches drone/3/battery but not drone/3/gps
• # matches all remaining levels, must come last, and also matches the parent: drone/# matches both drone/3/gps/fix and drone

def topic_matches(topic_filter, topic):
    f_parts, t_parts = topic_filter.split("/"), topic.split("/")
    for i, f in enumerate(f_parts):
        if f == "#":
            return True
        if i >= len(t_parts):
            return False
        if f != "+" and f != t_parts[i]:
            return False
    return len(f_parts) == len(t_parts)


cases = [("drone/+/battery", "drone/3/battery"), ("drone/+/battery", "drone/3/gps"), ("drone/#", "drone/3/gps/fix"),
         ("drone/+", "drone/3/battery"), ("drone/#", "drone"), ("fleet/+/+/alt", "fleet/a/7/alt")]
for f, t in cases:
    print(f"{f:<16} {t:<18} {topic_matches(f, t)}")

drone/+/battery  drone/3/battery    True
drone/+/battery  drone/3/gps        False
drone/#          drone/3/gps/fix    True
drone/+          drone/3/battery    False
drone/#          drone              True
fleet/+/+/alt    fleet/a/7/alt      True

QoS: three delivery guarantees

MQTT sets a Quality of Service (QoS) for each message.

Higher QoS costs more messages and time. In QoS 1, if the PUBACK is lost, the sender sends the PUBLISH again, so the receiver gets the same message twice. The receiving program must be ready for that.

received = []


def deliver_qos1(msg_id, payload, puback_lost):
    attempts = 0
    while True:
        attempts += 1
        received.append((msg_id, payload))
        if not puback_lost or attempts > 1:
            return attempts


print("attempts:", deliver_qos1(7, "RTL", puback_lost=True), "received:", received)

seen, actions = set(), []
for msg_id, payload in received:
    if msg_id in seen:
        continue
    seen.add(msg_id)
    actions.append(payload)
print("actions:", actions)

attempts: 2 received: [(7, 'RTL'), (7, 'RTL')]
actions: ['RTL']

Retained messages and Last Will

• Retained message: a message sent with the RETAIN flag is kept by the broker as the latest message for that topic, replacing any earlier retained one. A subscriber that joins later gets the latest value at once instead of waiting for the next update. It suits status such as online, or mission settings.
• Last Will: a device leaves a message with the broker when it connects. If the connection closes abnormally, the broker publishes that message, such as drone/1/status = offline, so the command centre knows the drone has dropped off.

class MiniBroker:
    def __init__(self):
        self.subs = []
        self.retained = {}

    def subscribe(self, name, topic_filter):
        self.subs.append((topic_filter, name))
        for topic, payload in self.retained.items():
            if topic_matches(topic_filter, topic):
                print(f"  {name} <- {topic} = {payload} (retained)")

    def publish(self, topic, payload, retain=False):
        if retain:
            self.retained[topic] = payload
        for topic_filter, name in self.subs:
            if topic_matches(topic_filter, topic):
                print(f"  {name} <- {topic} = {payload}")


broker = MiniBroker()
broker.subscribe("gcs", "drone/+/battery")
broker.publish("drone/1/status", "online", retain=True)
broker.publish("drone/1/battery", "76%")
print("dashboard joins late:")
broker.subscribe("dashboard", "drone/#")
print("drone 1 link drops, broker sends its will:")
broker.publish("drone/1/status", "offline", retain=True)

  gcs <- drone/1/battery = 76%
dashboard joins late:
  dashboard <- drone/1/status = online (retained)
drone 1 link drops, broker sends its will:
  dashboard <- drone/1/status = offline

The dashboard that joins late gets the online status at once because it was retained, but not the 76% battery value, because that message was not retained. This toy broker only illustrates the principle.

MQTT 5.0 (an OASIS standard from 2019) adds important features over 3.1.1, such as reason codes that explain failures, session expiry, which sets how long the broker remembers a subscriber’s subscriptions, and the message expiry interval, which discards messages older than a limit so that expired commands are not delivered.

Data freshness and queues

A delivered message is not necessarily still useful. Data age is the time from measurement to use. If messages enter a queue faster than the receiver can process them ($\lambda >\mu$), the queue keeps growing and new messages wait longer and longer. These formulas come from the MQTT and Mosquitto knowledge unit of the drone knowledge hub.

lam, mu, q0 = 50, 40, 0
for t in (10, 60, 300):
    q = max(0, q0 + (lam - mu) * t)
    print(f"after {t:>3} s  queue {q:>4} messages  a new message waits about {q / mu:.1f} s")

after  10 s  queue  100 messages  a new message waits about 2.5 s
after  60 s  queue  600 messages  a new message waits about 15.0 s
after 300 s  queue 3000 messages  a new message waits about 75.0 s

Mosquitto and security

Eclipse Mosquitto is a widely used open-source MQTT broker. The latest release at the time of writing is 2.1.2 (February 2026). It comes with mosquitto_pub and mosquitto_sub for testing. IANA registers port 1883 for plain MQTT and 8883 for MQTT over TLS.

Since Mosquitto 2.0, a broker with no listener configured accepts only local (loopback) connections, and once a listener is configured allow_anonymous defaults to false. These defaults prevent accidentally exposing a broker to the network. Real deployments should add three more layers:

1. TLS encrypts data in transit over port 8883, preventing eavesdropping.
2. One account per device: each drone has its own username and password or certificate.
3. Per-topic rights (ACL): drone 1 may publish only to drone/1/#, so it cannot pose as drone 2.

An example client with paho-mqtt version 2, which requires CallbackAPIVersion (this code needs a real broker, so the automatic check does not run it):

import os
import paho.mqtt.client as mqtt


def on_connect(client, userdata, flags, reason_code, properties):
    print("connected:", reason_code)
    client.subscribe("drone/+/battery", qos=1)


def on_message(client, userdata, msg):
    print(msg.topic, msg.payload.decode())


client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="gcs-01", protocol=mqtt.MQTTv5)
client.username_pw_set("gcs-01", os.environ["MQTT_PASSWORD"])
client.tls_set(ca_certs="ca.crt")
client.will_set("gcs/gcs-01/status", "offline", qos=1, retain=True)
client.on_connect = on_connect
client.on_message = on_message
client.connect("broker.field.lan", 8883)
client.loop_forever()

The password is read from an environment variable, not written in the code, following UAT 104 module 5.

Module lab

Common mistakes

Summary

• IoT has devices, gateways, broker and cloud, and applications; MQTT links every layer with publish/subscribe
• + matches one level; # matches all remaining levels and must come last
• QoS 0 may lose, QoS 1 may duplicate, QoS 2 delivers exactly once but is slowest; retained messages give new subscribers the latest value, and Last Will reports dropped devices
• Configure Mosquitto with TLS, separate accounts and ACLs, and always check data age, because queues can make data stale without any loss

Check your understanding

1. Does site/+/drone/# match site/a/drone/2/gps?
2. Does drone/+ match drone/2/battery? Why?
3. Which QoS should a “land” command use, and how should the command be designed?
4. Messages arrive at 30 per second and are processed at 25 per second, starting from an empty queue. After 2 minutes, how long is the queue and about how long does a new message wait?
5. Why should a broker not be open on port 1883 on a network others can join?