📡 MQTT – Architecture, Brokers, IoT, Sécurité & Real-Time Ingestion

Dans une taxonomie de développement web ou backend, MQTT se place dans la famille Messaging / Message Brokers / Event-driven architecture. Techniquement, MQTT est un protocole, mais dans un portail technique il est souvent classé près des brokers qui l’implémentent : Mosquitto, EMQX, HiveMQ, VerneMQ, RabbitMQ MQTT plugin.

Arbre de classification recommandé

Web / Dev
                            └── Backend
                            └── Architecture distribuée
                            └── Messaging & Event-driven architecture
                            └── Message Brokers
                            └── MQTT
                            ├── Mosquitto
                            ├── EMQX
                            ├── HiveMQ
                            ├── VerneMQ
                            └── RabbitMQ MQTT Plugin

Badges utiles pour une fiche IDEO-Lab

# Messaging
                            # Message Broker
                            # Pub/Sub
                            # IoT
                            # Edge Computing
                            # Real-Time
                            # Telemetry
                            # M2M
                            # MQTT 3.1.1
                            # MQTT 5
                            # Lightweight Protocol
                            # Event-Driven Architecture

MQTT repose sur un modèle très simple : les clients ne s’appellent pas directement. Ils passent par un broker. Un producteur publie un message sur un topic. Les consommateurs s’abonnent à des topics exacts ou à des motifs avec wildcards.

Flux publish / subscribe

1. Le device se connecte au broker
                            CONNECT client_id=device-042 username=gw_paris

                            2. Le dashboard s’abonne à un topic
                            SUBSCRIBE factory/line1/+/temperature

                            3. Le device publie une mesure
                            PUBLISH factory/line1/motor7/temperature
                            Payload: {"value": 72.4, "unit": "C", "ts": "2026-04-25T10:00:00Z"}

                            4. Le broker route le message
                            Tous les abonnés compatibles reçoivent le message

                            5. Les consommateurs traitent
                            - affichage dashboard
                            - stockage time-series
                            - alerte si seuil dépassé
                            - pipeline ML ou prédiction

Topologie logique

Topic exact :
                            factory/line1/motor7/temperature

                            Wildcard + :
                            factory/line1/+/temperature
                            => reçoit :
                            factory/line1/motor7/temperature
                            factory/line1/motor8/temperature

                            Wildcard # :
                            factory/line1/#
                            => reçoit tout sous factory/line1

                            Attention :
                            # est puissant mais dangereux en production
                            si utilisé trop largement.

Architecture industrielle typique

┌─────────────────────────┐
                            │ Machines / PLC / Sensors│
                            └────────────┬────────────┘
                            │ Modbus / OPC-UA / Serial / GPIO
                            ▼
                            ┌─────────────────────────┐
                            │ Edge Gateway            │
                            │ - normalise les données │
                            │ - enrichit les messages │
                            │ - buffer local          │
                            │ - publie MQTT           │
                            └────────────┬────────────┘
                            │ MQTT/TLS
                            ▼
                            ┌─────────────────────────┐
                            │ Broker MQTT local/cloud │
                            └───────┬─────────┬───────┘
                            │         │
                            ▼         ▼
                            ┌────────────┐ ┌──────────────┐
                            │ Historian  │ │ Dashboard    │
                            │ TimeSeries │ │ Supervision  │
                            └────────────┘ └──────────────┘
                            │
                            ▼
                            ┌─────────────────────────┐
                            │ Analytics / IA / Alertes│
                            └─────────────────────────┘

Checklist de conception

Avant de mettre MQTT en production :

                            [ ] Nommer les topics officiellement
                            [ ] Définir les ACL par client/groupe
                            [ ] Activer TLS si réseau non totalement privé
                            [ ] Choisir QoS par type de message
                            [ ] Ajouter message_id pour idempotence
                            [ ] Versionner les payloads
                            [ ] Monitorer broker, connexions, messages/sec
                            [ ] Prévoir retained messages pour états courants
                            [ ] Définir Last Will pour online/offline
                            [ ] Tester reconnexion, coupures réseau, doublons
                            [ ] Documenter conventions et exemples

Ce que MQTT ne fait pas seul

• Il ne garantit pas automatiquement une sémantique métier “exactement une fois”.
• Il ne remplace pas une base time-series ou un event store.
• Il ne valide pas le schéma du payload par défaut.
• Il ne protège pas sans configuration TLS/auth/ACL.
• Il ne résout pas les problèmes de gouvernance des topics.

Standards & documentation

MQTT official:
                            https://mqtt.org/

                            MQTT 5.0 OASIS standard:
                            https://www.oasis-open.org/standard/mqtt-v5-0-os/

                            MQTT 5.0 specification:
                            https://docs.oasis-open.org/mqtt/mqtt/v5.0/mqtt-v5.0.html

                            MQTT 3.1.1 OASIS / ISO information:
                            https://www.oasis-open.org/2016/07/19/oasis-mqtt-internet-of-things-standard-now-approved-by-iso-iec-jtc1/

                            Eclipse Sparkplug:
                            https://sparkplug.eclipse.org/specification/

Brokers MQTT

Eclipse Mosquitto:
                            https://mosquitto.org/

                            EMQX:
                            https://www.emqx.com/

                            HiveMQ:
                            https://www.hivemq.com/

                            VerneMQ:
                            https://vernemq.com/

                            RabbitMQ MQTT plugin:
                            https://www.rabbitmq.com/docs/mqtt

Cloud IoT & intégrations

AWS IoT Core:
                            https://aws.amazon.com/iot-core/

                            Azure IoT Hub:
                            https://azure.microsoft.com/products/iot-hub/

                            Google Cloud IoT alternatives / partner ecosystem:
                            https://cloud.google.com/solutions/iot

                            Node-RED:
                            https://nodered.org/

                            Telegraf MQTT Consumer:
                            https://docs.influxdata.com/telegraf/

                            Paho MQTT clients:
                            https://www.eclipse.org/paho/

Librairies clientes courantes

Le broker MQTT est le serveur qui accepte les connexions clientes, gère les topics, contrôle qui a le droit de publier ou de s’abonner, route les messages, maintient les sessions et expose des métriques. Dans une architecture sérieuse, il est considéré comme un composant critique d’infrastructure.

Broker comme point de contrôle

Client CONNECT
                            │
                            ▼
                            ┌──────────────────────────────┐
                            │ Broker MQTT                  │
                            ├──────────────────────────────┤
                            │ 1. Accepte ou refuse TLS     │
                            │ 2. Authentifie le client     │
                            │ 3. Charge les ACL            │
                            │ 4. Attache la session        │
                            │ 5. Reçoit publish/subscribe  │
                            │ 6. Route les messages        │
                            │ 7. Journalise / expose stats │
                            └──────────────────────────────┘

Comparaison rapide de brokers

Le topic est l’adresse logique d’un message MQTT. Il est hiérarchique, séparé par des slashs. Le broker ne comprend pas forcément le sens métier du topic, mais il l’utilise pour router les messages vers les abonnements correspondants.

Exemple de topic industriel

factory/fr/paris/site-01/line-03/motor-07/temperature

                            factory     = domaine
                            fr          = pays ou région
                            paris       = zone
                            site-01     = site industriel
                            line-03     = ligne de production
                            motor-07    = équipement
                            temperature = mesure

Wildcards MQTT

Routage par topic

PUBLISH :
                            factory/fr/paris/site-01/line-03/motor-07/temperature

                            SUBSCRIBERS :

                            1) factory/fr/paris/site-01/line-03/motor-07/temperature
                            => reçoit uniquement ce moteur précis

                            2) factory/fr/paris/site-01/line-03/+/temperature
                            => reçoit toutes les températures des équipements de line-03

                            3) factory/fr/paris/site-01/#
                            => reçoit tout le site

                            4) factory/#
                            => reçoit tout le domaine factory
                            => dangereux si volumétrie élevée

Bonnes conventions

Exemple de séparation propre

devices/gw-001/status
                            devices/gw-001/telemetry/temperature
                            devices/gw-001/telemetry/vibration
                            devices/gw-001/events/alarm
                            devices/gw-001/commands/reboot
                            devices/gw-001/commands/config/update

Une session MQTT représente l’état logique associé à un client : son Client ID, ses abonnements, ses messages QoS en attente et les échanges non acquittés. La gestion de session est essentielle pour les clients intermittents.

Cycle de vie d’une session

CONNECT client_id="gateway-paris-001"
                            clean_start=false
                            session_expiry=86400

                            │
                            ▼
                            Broker vérifie si session existante
                            │
                            ├── Oui :
                            │     ├── restaure subscriptions
                            │     ├── conserve messages QoS en attente
                            │     └── reprend les échanges inflight
                            │
                            └── Non :
                            ├── crée nouvelle session
                            ├── attend subscriptions
                            └── initialise l’état client

Profils de session recommandés

Erreur fréquente

Deux clients avec le même Client ID
                            │
                            ▼
                            Le broker déconnecte l’ancien client
                            │
                            ▼
                            Symptôme :
                            - connexions instables
                            - devices qui se déconnectent
                            - messages perdus ou sessions écrasées

                            Solution :
                            - générer Client ID unique et stable
                            - contrôler l’onboarding device
                            - auditer les connexions du broker

Retained messages

Un message retained est conservé par le broker comme dernière valeur connue d’un topic. Lorsqu’un nouveau subscriber s’abonne, il reçoit immédiatement cette dernière valeur, sans attendre la prochaine publication.

Topic :
                            devices/gw-034/status

                            Payload retained :
                            {
                            "state": "online",
                            "firmware": "1.8.2",
                            "ip": "10.0.4.12",
                            "ts": "2026-04-25T10:00:00Z"
                            }

Last Will and Testament

Le Last Will est un message que le client prépare au moment de sa connexion. Si le client disparaît brutalement, le broker publie ce message à sa place. C’est le mécanisme standard pour signaler qu’un device est offline.

CONNECT gateway-paris-001
                            will_topic   = devices/gateway-paris-001/status
                            will_payload = {"state": "offline", "reason": "connection_lost"}
                            will_qos     = 1
                            will_retain  = true

                            Après connexion normale :
                            PUBLISH retained devices/gateway-paris-001/status
                            {"state": "online"}

                            Si coupure brutale :
                            Broker publie automatiquement :
                            {"state": "offline", "reason": "connection_lost"}

Diagramme état online/offline

Gateway démarre
                            │
                            ├── CONNECT avec Last Will offline
                            │
                            ├── PUBLISH retained online
                            │
                            ▼
                            Device visible comme ONLINE
                            │
                            ├── Déconnexion propre :
                            │       DISCONNECT
                            │       Pas de Last Will
                            │
                            └── Crash / réseau coupé :
                            Broker publie Last Will
                            Device visible comme OFFLINE

Flux QoS 0 : fire and forget

Publisher              Broker               Subscriber
                            │                    │                       │
                            │ PUBLISH QoS 0      │                       │
                            ├───────────────────►│                       │
                            │                    │ PUBLISH QoS 0         │
                            │                    ├──────────────────────►│
                            │                    │                       │

                            Garantie :
                            - au plus une fois
                            - pas d’ack protocolaire
                            - rapide, léger
                            - perte possible

Flux QoS 1 : au moins une fois

Publisher              Broker
                            │                    │
                            │ PUBLISH QoS 1      │
                            ├───────────────────►│
                            │ PUBACK             │
                            ◄────────────────────┤

                            Garantie :
                            - au moins une fois
                            - doublons possibles
                            - nécessite idempotence côté consommateur

Flux QoS 2 : exactement une fois protocolaire

Publisher              Broker
                            │                    │
                            │ PUBLISH QoS 2      │
                            ├───────────────────►│
                            │ PUBREC             │
                            ◄────────────────────┤
                            │ PUBREL             │
                            ├───────────────────►│
                            │ PUBCOMP            │
                            ◄────────────────────┤

                            Garantie :
                            - échange en 4 étapes
                            - plus coûteux
                            - utile seulement si vraiment justifié

Pipeline complet d’ingestion

Device
                            │
                            │ PUBLISH telemetry QoS 1
                            ▼
                            Broker MQTT
                            │
                            │ route vers subscribers
                            ▼
                            Ingestion worker
                            │
                            ├── validate payload
                            ├── check idempotency key
                            ├── enrich metadata
                            ├── write hot state
                            ├── write historical record
                            └── trigger alert if needed

Dans un backend Django, MQTT ne doit pas être mélangé directement aux vues HTTP. Une architecture propre isole l’ingestion MQTT dans un service ou un management command dédié, puis pousse le travail vers Redis/Celery ou vers une file interne.

[MQTT Broker]
                            │
                            │ SUBSCRIBE devices/+/telemetry/#
                            ▼
                            [ingestion_service.py]
                            │
                            ├── connect MQTT
                            ├── validate topic
                            ├── validate payload schema
                            ├── normalize timestamp
                            ├── enrich with Device/Site metadata
                            ├── build idempotency key
                            ├── push job to Celery / Redis Stream
                            └── ack / log / retry policy

                            [Celery Workers]
                            │
                            ├── write current state to Redis
                            ├── write history to PostgreSQL / TimescaleDB
                            ├── trigger alerting rules
                            ├── notify WebSocket dashboard
                            └── store failures in admin-visible table

                            [Django Admin / API]
                            │
                            ├── devices
                            ├── certificates
                            ├── ACL / topic map
                            ├── ingestion failures
                            ├── replay tools
                            └── monitoring dashboard

Découpage recommandé des responsabilités

Exemple d’idempotency key

idempotency_key =
                            sha256(
                            client_id + "|" +
                            topic + "|" +
                            payload.message_id + "|" +
                            payload.ts
                            )

                            But :
                            - éviter double insertion
                            - supporter QoS 1
                            - rejouer sans casser la DB
                            - tracer les doublons

En production, le broker est critique. Selon la volumétrie et le niveau de service, on peut utiliser un broker simple, un cluster MQTT, ou une architecture edge avec broker local et bridge vers un broker central/cloud.

Architecture Edge + Cloud

┌───────────────────────────────┐
                            │ Site industriel local          │
                            │                               │
                            │  Sensors / PLC / Machines      │
                            │          │                    │
                            │          ▼                    │
                            │  Edge Gateway                  │
                            │          │                    │
                            │          ▼                    │
                            │  Local MQTT Broker             │
                            │          │                    │
                            │          ├── Local dashboard   │
                            │          ├── Local rules       │
                            │          └── Bridge MQTT/TLS   │
                            └──────────┼────────────────────┘
                            │ Internet / VPN / 4G
                            ▼
                            ┌───────────────────────────────┐
                            │ Central MQTT Broker / Cloud    │
                            │          │                    │
                            │          ├── Data Lake         │
                            │          ├── Time-series DB    │
                            │          ├── Alerting          │
                            │          └── Global dashboard  │
                            └───────────────────────────────┘

Bridge MQTT : points à surveiller

• Préfixes de topics entre local et central.
• Éviter les boucles de bridge.
• Limiter les wildcards trop larges.
• Définir la politique de reconnexion et buffer.
• Signer/chiffrer les connexions entre sites.
• Monitorer le lag entre edge et cloud.

Une architecture MQTT sans métriques est dangereuse : le broker peut sembler fonctionner alors qu’il accumule des sessions, rejette des messages, sature CPU, perd des connexions ou subit un abonnement wildcard trop coûteux.

Dashboard MQTT recommandé

MQTT Production Dashboard
                            ├── Broker health
                            │   ├── CPU / RAM / FD usage
                            │   ├── uptime
                            │   └── cluster status
                            │
                            ├── Clients
                            │   ├── connected clients
                            │   ├── connect/disconnect rate
                            │   ├── top clients by messages
                            │   └── auth failures
                            │
                            ├── Traffic
                            │   ├── messages in/sec
                            │   ├── messages out/sec
                            │   ├── bytes in/out
                            │   └── dropped messages
                            │
                            ├── Sessions
                            │   ├── persistent sessions
                            │   ├── queued messages
                            │   └── inflight messages
                            │
                            └── Topics
                            ├── hottest prefixes
                            ├── wildcard subscriptions
                            └── retained count

Alertes utiles

Checklist architecture avant production

[ ] Broker choisi selon volumétrie réelle
                            [ ] Topics documentés et gouvernés
                            [ ] ACL publish/subscribe définies
                            [ ] TLS activé si réseau non totalement privé
                            [ ] Client IDs stables pour devices critiques
                            [ ] Last Will configuré pour gateways/devices
                            [ ] Retained réservé aux états courants
                            [ ] QoS choisi par type de message
                            [ ] Idempotence backend en place
                            [ ] Payloads versionnés
                            [ ] Monitoring broker activé
                            [ ] Alertes connect/disconnect configurées
                            [ ] Stratégie HA ou backup définie
                            [ ] Tests de coupure réseau réalisés
                            [ ] Tests de reconnexion réalisés
                            [ ] Tests de duplication QoS 1 réalisés

Architecture saine en une phrase

Diagramme anti-chaos

Mauvais :
                            devices/all/data
                            └── payloads différents, pas de version, pas d’ACL fine

                            Bon :
                            devices/{site_id}/{device_id}/telemetry/{metric}
                            devices/{site_id}/{device_id}/status
                            devices/{site_id}/{device_id}/events/{event_type}
                            devices/{site_id}/{device_id}/commands/{command_name}

                            Résultat :
                            - ACL simples
                            - dashboards filtrables
                            - ingestion plus robuste
                            - debugging plus rapide
                            - évolution plus propre

Le paquet CONNECT démarre la session MQTT. Il transporte les paramètres de session, l’identité du client, le keepalive, les informations d’authentification, et éventuellement le message Last Will.

Handshake de connexion

Client                                      Broker
                            │                                           │
                            │ CONNECT                                   │
                            │  client_id = gateway-paris-001            │
                            │  clean_start = false                      │
                            │  keepalive = 60                           │
                            │  will_topic = devices/gw-001/status       │
                            │  username / password or certificate       │
                            ├──────────────────────────────────────────►│
                            │                                           │
                            │                         Authentification  │
                            │                         ACL / quotas      │
                            │                         session restore   │
                            │                                           │
                            │ CONNACK                                   │
                            │  success / failure                        │
                            │  session_present                          │
                            │  reason_code                              │
                            ◄──────────────────────────────────────────┤

Codes d’échec typiques

PUBLISH transporte le message applicatif. Il contient au minimum un topic et un payload. Selon le QoS, il contient aussi un Packet Identifier. Selon MQTT 5, il peut également contenir des properties comme Message Expiry, Response Topic, Correlation Data, Payload Format Indicator ou User Properties.

Exemple de payload propre

Topic:
                            factory/fr/paris/site-01/line-03/motor-07/telemetry/temperature

                            Payload:
                            {
                            "schema_version": "1.0",
                            "message_id": "01HX9F4Z7C3X",
                            "device_id": "motor-07",
                            "metric": "temperature",
                            "value": 72.4,
                            "unit": "C",
                            "ts": "2026-04-25T10:00:00Z"
                            }

Structure logique PUBLISH

PUBLISH Packet
                            ├── Fixed Header
                            │   ├── Packet Type = PUBLISH
                            │   ├── DUP flag
                            │   ├── QoS level
                            │   ├── Retain flag
                            │   └── Remaining Length
                            │
                            ├── Variable Header
                            │   ├── Topic Name
                            │   ├── Packet Identifier if QoS 1 or 2
                            │   └── Properties if MQTT 5
                            │
                            └── Payload
                            └── Application bytes

Flags importants

QoS 0 : au plus une fois

Publisher                           Broker
                            │                                  │
                            │ PUBLISH QoS 0                    │
                            ├─────────────────────────────────►│
                            │                                  │

                            Caractéristiques :
                            - pas d’accusé protocolaire
                            - pas de retransmission MQTT
                            - très rapide
                            - perte possible
                            - idéal pour mesures fréquentes non critiques

QoS 1 : au moins une fois

Publisher                           Broker
                            │                                  │
                            │ PUBLISH QoS 1 packet_id=42        │
                            ├─────────────────────────────────►│
                            │ PUBACK packet_id=42               │
                            ◄─────────────────────────────────┤

                            Si PUBACK est perdu :
                            - le publisher peut renvoyer PUBLISH
                            - le broker ou le subscriber peut recevoir un doublon

                            Conséquence :
                            - le backend doit être idempotent

QoS 2 : exactement une fois au niveau protocole

Publisher                           Broker
                            │                                  │
                            │ PUBLISH QoS 2 packet_id=77        │
                            ├─────────────────────────────────►│
                            │ PUBREC packet_id=77               │
                            ◄─────────────────────────────────┤
                            │ PUBREL packet_id=77               │
                            ├─────────────────────────────────►│
                            │ PUBCOMP packet_id=77              │
                            ◄─────────────────────────────────┤

                            Caractéristiques :
                            - handshake en 4 étapes
                            - plus d’état côté client/broker
                            - plus coûteux
                            - utile seulement pour messages critiques

Table de décision QoS

Le paquet SUBSCRIBE permet à un client de demander au broker de recevoir les messages correspondant à un ou plusieurs topic filters. Le broker répond avec SUBACK, qui indique le QoS réellement accordé ou un échec.

Exemple classique

SUBSCRIBE
                            topic_filter = factory/fr/paris/site-01/+/temperature
                            requested_qos = 1

                            SUBACK
                            granted_qos = 1

                            Résultat :
                            Le client reçoit toutes les températures sous site-01
                            sur exactement un niveau après site-01.

Shared subscriptions

Sans shared subscription :

                            Broker
                            ├── message A -> worker-1
                            ├── message A -> worker-2
                            └── message A -> worker-3

                            Chaque worker reçoit le même message.

                            Avec shared subscription :

                            $share/ingestion-workers/factory/+/telemetry

                            Broker
                            ├── message A -> worker-1
                            ├── message B -> worker-2
                            ├── message C -> worker-3
                            └── message D -> worker-1

                            Les messages sont répartis dans le groupe.

Usage backend

MQTT garde une connexion ouverte. Pour éviter les connexions mortes silencieuses, le client annonce une valeur de Keep Alive dans CONNECT. Si aucun autre paquet n’est échangé pendant cette période, le client envoie un PINGREQ. Le broker répond avec PINGRESP.

CONNECT keepalive=60

                            Pendant 60 secondes :
                            - si PUBLISH, SUBSCRIBE ou autre paquet circule,
                            pas besoin de PINGREQ.

                            Si silence :
                            Client -> PINGREQ
                            Broker -> PINGRESP

                            Si pas de PINGRESP :
                            Client considère la connexion morte et reconnecte.

Diagramme keepalive

Client                                  Broker
                            │                                       │
                            │ CONNECT keepalive=60                  │
                            ├──────────────────────────────────────►│
                            │ CONNACK                               │
                            ◄──────────────────────────────────────┤
                            │                                       │
                            │ ... aucun trafic pendant 60s ...      │
                            │                                       │
                            │ PINGREQ                               │
                            ├──────────────────────────────────────►│
                            │ PINGRESP                              │
                            ◄──────────────────────────────────────┤
                            │                                       │
                            │ Connexion considérée vivante          │

Keepalive vs Last Will

DISCONNECT permet de fermer proprement une session MQTT. Lorsqu’un client envoie DISCONNECT, le broker sait que la fermeture est volontaire et ne publie pas le Last Will. En MQTT 5, DISCONNECT peut aussi transporter un reason code et des properties.

Exemple logique

Cas 1 : arrêt volontaire

                            Client:
                            PUBLISH devices/gw-001/status {"state":"shutting_down"} retain=true
                            DISCONNECT

                            Broker:
                            ferme proprement
                            ne publie pas Last Will

                            Cas 2 : crash brutal

                            Client:
                            disparaît sans DISCONNECT

                            Broker:
                            détecte la perte
                            publie Last Will offline

MQTT 5 DISCONNECT reason codes

Exemple MQTT 5 : commande avec expiration

PUBLISH
                            topic = devices/gw-001/commands/reboot
                            qos = 1
                            payload = {
                            "command_id": "cmd-20260425-001",
                            "action": "reboot"
                            }

                            MQTT 5 properties:
                            message_expiry_interval = 30
                            response_topic = devices/gw-001/commands/ack
                            correlation_data = cmd-20260425-001
                            user_properties:
                            tenant_id = customer-42
                            trace_id = trace-abc-789

                            Effet :
                            - si la gateway reçoit trop tard, le message est périmé
                            - la réponse revient sur un topic connu
                            - le backend corrèle commande et ack

MQTT 3.1.1 vs MQTT 5

Scénario normal

[CONNECT]
                            client_id=gateway-paris-001
                            keepalive=60
                            clean_start=false
                            will=devices/gateway-paris-001/status

                            [CONNACK]
                            success=true
                            session_present=true

                            [SUBSCRIBE]
                            topic=devices/gateway-paris-001/commands/#
                            qos=1

                            [SUBACK]
                            granted_qos=1

                            [PUBLISH]
                            topic=devices/gateway-paris-001/status
                            retain=true
                            qos=1
                            payload={"state":"online"}

                            [PUBACK]
                            packet_id=10

                            [PINGREQ]
                            [PINGRESP]

                            [DISCONNECT]
                            reason=normal

Ce que tu vérifies en premier

Runbook debug MQTT

1. Vérifier la connexion TCP/TLS
                            - port 1883 / 8883
                            - certificat
                            - DNS
                            - firewall

                            2. Vérifier CONNECT
                            - Client ID unique
                            - keepalive
                            - clean start / session expiry
                            - username/password/token
                            - Last Will

                            3. Lire CONNACK
                            - success
                            - reason code
                            - session present

                            4. Vérifier SUBSCRIBE/SUBACK
                            - topic filter exact
                            - wildcards correctes
                            - QoS accordé
                            - ACL subscribe

                            5. Vérifier PUBLISH
                            - topic exact
                            - QoS
                            - retain
                            - payload valide
                            - packet id si QoS 1/2

                            6. Vérifier les ACK
                            - PUBACK pour QoS 1
                            - PUBREC/PUBREL/PUBCOMP pour QoS 2

                            7. Vérifier keepalive
                            - PINGREQ/PINGRESP
                            - timeouts
                            - reconnect loops

                            8. Vérifier backend
                            - idempotence
                            - validation payload
                            - logs ingestion
                            - erreurs DB

Commandes utiles Mosquitto

Subscribe debug :
                            mosquitto_sub -h broker.example.com -p 8883 \
                            --cafile ca.crt \
                            -u user -P password \
                            -t 'factory/fr/#' -v

                            Publish test :
                            mosquitto_pub -h broker.example.com -p 8883 \
                            --cafile ca.crt \
                            -u user -P password \
                            -t 'factory/fr/test/device-1/status' \
                            -m '{"state":"online"}' \
                            -q 1 -r

                            Debug local :
                            mosquitto_sub -h localhost -t '#' -v

Avec QoS 0, le client envoie un paquet PUBLISH et n’attend aucun accusé de réception. C’est le mode le plus rapide et le moins coûteux. En contrepartie, si le réseau coupe, si le broker est indisponible ou si le client subscriber n’est pas là, le message peut être perdu.

Bon usage QoS 0

• Température envoyée toutes les 5 ou 10 secondes.
• Position GPS très fréquente.
• Heartbeat à haute fréquence.
• Métriques observabilité non critiques.
• Données remplacées rapidement par une mesure plus récente.

Flux QoS 0

Publisher                           Broker
                            │                                  │
                            │ PUBLISH QoS 0                    │
                            ├─────────────────────────────────►│
                            │                                  │
                            │ Aucun PUBACK                     │
                            │ Aucun retry MQTT                 │
                            │                                  │

                            Conséquence :
                            - rapide
                            - simple
                            - pas de garantie de réception
                            - pas d’état inflight à maintenir

Exemple de décision

Flux :
                            devices/gw-001/telemetry/temperature
                            Fréquence :
                            1 message toutes les 5 secondes
                            Criticité :
                            faible à moyenne
                            Historique :
                            time-series compressée

                            Choix :
                            QoS 0 si perte acceptable
                            QoS 1 si chaque mesure doit être auditée

QoS 1 est souvent le meilleur compromis en production. Le publisher envoie PUBLISH, le destinataire répond PUBACK. Si l’accusé n’arrive pas, le message peut être retransmis. Cela améliore la fiabilité, mais introduit une contrainte majeure : le consommateur doit tolérer les doublons.

Bon usage QoS 1

• Alarme machine.
• Événement métier.
• Changement d’état online/offline.
• Message de configuration important.
• Commande non destructrice mais à tracer.
• Mesures qui doivent être conservées avec audit.

Flux QoS 1

Publisher                           Broker
                            │                                  │
                            │ PUBLISH packet_id=42 QoS 1        │
                            ├─────────────────────────────────►│
                            │                                  │
                            │ PUBACK packet_id=42               │
                            ◄─────────────────────────────────┤

                            Cas problématique :
                            1. Broker reçoit PUBLISH
                            2. Broker traite le message
                            3. PUBACK se perd sur le réseau
                            4. Publisher renvoie PUBLISH avec DUP flag
                            5. Backend peut revoir le même événement

Implication backend

Payload conseillé :
                            {
                            "message_id": "01HX9F4Z7C3X",
                            "device_id": "gw-001",
                            "sequence": 18422,
                            "metric": "temperature",
                            "value": 72.4,
                            "ts": "2026-04-25T10:00:00Z"
                            }

                            Déduplication :
                            UNIQUE(device_id, sequence)
                            ou
                            UNIQUE(message_id)

QoS 2 utilise un échange en quatre étapes pour éviter la double livraison au niveau MQTT. C’est le niveau de garantie le plus fort du protocole, mais aussi le plus coûteux en latence, état client/broker et complexité opérationnelle.

Flux QoS 2 détaillé

Publisher                           Broker
                            │                                  │
                            │ PUBLISH packet_id=77 QoS 2        │
                            ├─────────────────────────────────►│
                            │                                  │
                            │ PUBREC packet_id=77               │
                            ◄─────────────────────────────────┤
                            │                                  │
                            │ PUBREL packet_id=77               │
                            ├─────────────────────────────────►│
                            │                                  │
                            │ PUBCOMP packet_id=77              │
                            ◄─────────────────────────────────┤

                            But :
                            - éviter double livraison MQTT
                            - synchroniser l’état publisher/broker
                            - nettoyer l’état après PUBCOMP

Quand utiliser QoS 2 ?

La duplication n’est pas un bug exceptionnel : elle fait partie du modèle de livraison “at least once”. Un message peut être envoyé plusieurs fois si un accusé est perdu, si le client reconnecte, si une session reprend ou si le backend rejoue un buffer.

Le flag DUP ne suffit pas

Le flag DUP indique une retransmission MQTT. Il ne doit pas être utilisé comme unique mécanisme de déduplication métier, car la duplication peut venir d’autres couches : buffer local, bridge, worker, retry DB, redémarrage applicatif.

Timeline d’un doublon QoS 1

1. Device publie :
                            PUBLISH packet_id=42 message_id=A

                            2. Broker reçoit le message.

                            3. Broker transmet au subscriber.

                            4. Subscriber écrit en base.

                            5. PUBACK ou ack côté chemin retour est perdu.

                            6. Device retransmet :
                            PUBLISH packet_id=42 DUP=1 message_id=A

                            7. Backend revoit le même message.

                            Sans idempotence :
                            double insertion

                            Avec idempotence :
                            message reconnu comme déjà traité

Stratégies anti-doublons

Bonnes clés possibles :
                            - message_id global dans le payload
                            - device_id + sequence_number
                            - device_id + metric_name + timestamp
                            - command_id pour commandes
                            - MQTT 5 user property trace_id

                            Stockage :
                            - table inbox_mqtt_message
                            - contrainte unique applicative
                            - statut processed / failed / ignored
                            - timestamp first_seen / last_seen
                            - duplicate_count

L’idempotence signifie qu’un même message traité plusieurs fois produit le même état final que s’il avait été traité une seule fois. C’est indispensable avec QoS 1 et très recommandé même avec QoS 2, car les retries peuvent aussi exister hors MQTT.

Modèle conceptuel

Message MQTT
                            │
                            ▼
                            Validate schema
                            │
                            ▼
                            Build idempotency_key
                            │
                            ▼
                            Check inbox table / cache
                            │
                            ├── already processed
                            │       └── ignore or update duplicate_count
                            │
                            └── new message
                            ├── process in transaction
                            ├── write target table
                            ├── mark as processed
                            └── commit

Candidats de clé d’idempotence

Exemple Django : table inbox

class MQTTInboxMessage(models.Model):
                            idempotency_key = models.CharField(max_length=128)
                            topic = models.CharField(max_length=512)
                            client_id = models.CharField(max_length=128)
                            payload_hash = models.CharField(max_length=64)

                            status = models.CharField(max_length=32, default="pending")
                            first_seen_at = models.DateTimeField(auto_now_add=True)
                            last_seen_at = models.DateTimeField(auto_now=True)
                            duplicate_count = models.IntegerField(default=0)

                            error_message = models.TextField(blank=True, default="")

                            class Meta:
                            constraints = [
                            models.UniqueConstraint(
                            fields=["idempotency_key"],
                            name="uniq_mqtt_inbox_idempotency_key"
                            )
                            ]

Pseudo-code traitement

def handle_mqtt_message(topic, payload, client_id):
                            data = validate_payload(payload)
                            key = build_idempotency_key(topic, data, client_id)

                            with transaction.atomic():
                            inbox, created = MQTTInboxMessage.objects.get_or_create(
                            idempotency_key=key,
                            defaults={
                            "topic": topic,
                            "client_id": client_id,
                            "payload_hash": sha256(payload),
                            "status": "pending",
                            }
                            )

                            if not created:
                            inbox.duplicate_count += 1
                            inbox.save(update_fields=["duplicate_count", "last_seen_at"])
                            return "duplicate_ignored"

                            write_business_data(data)
                            inbox.status = "processed"
                            inbox.save(update_fields=["status"])

                            return "processed"

La backpressure apparaît quand les publishers envoient plus vite que les consumers, workers ou bases de données ne peuvent absorber. MQTT peut bufferiser certains messages selon sessions/QoS, mais ce n’est pas une solution infinie. Il faut une stratégie explicite.

Mécanismes utiles

• Receive Maximum en MQTT 5 pour limiter l’inflight.
• Shared subscriptions pour répartir la charge entre workers.
• Redis Streams / Kafka comme buffer durable côté backend.
• Batch database writes pour réduire coût SQL.
• Priorisation entre alarmes critiques et métriques secondaires.
• Message Expiry pour éviter de traiter des commandes périmées.

Architecture avec buffer

MQTT Broker
                            │
                            │ shared subscription
                            ▼
                            Ingestion Workers
                            │
                            ├── validate
                            ├── deduplicate
                            └── push to buffer
                            │
                            ▼
                            Redis Stream / Kafka
                            │
                            ▼
                            Processing Workers
                            │
                            ├── batch write PostgreSQL
                            ├── update Redis hot state
                            ├── trigger alerts
                            └── push WebSocket dashboard

Stratégie de dégradation

Si surcharge :
                            1. protéger les alarmes critiques
                            2. ralentir ou rejeter métriques non critiques
                            3. augmenter workers consommateurs
                            4. batcher les écritures DB
                            5. réduire QoS sur flux non critiques
                            6. augmenter capacité broker si nécessaire
                            7. surveiller queue depth et lag
                            8. exposer état degraded dans l’admin

Pour les commandes critiques, MQTT QoS confirme la livraison protocolaire, mais pas l’exécution métier. Une commande peut être livrée au device, mais échouer côté machine : droit insuffisant, sécurité locale, état incompatible, timeout, opérateur requis.

Pattern commande / ack

Backend publish :
                            devices/gw-001/commands/reboot

                            Payload :
                            {
                            "command_id": "cmd-20260425-0001",
                            "action": "reboot",
                            "requested_by": "operator-7",
                            "expires_at": "2026-04-25T10:01:00Z"
                            }

                            Device ack :
                            devices/gw-001/commands/ack

                            Payload :
                            {
                            "command_id": "cmd-20260425-0001",
                            "status": "accepted",
                            "received_at": "2026-04-25T10:00:05Z"
                            }

                            Device result :
                            devices/gw-001/commands/result

                            Payload :
                            {
                            "command_id": "cmd-20260425-0001",
                            "status": "done",
                            "completed_at": "2026-04-25T10:00:28Z"
                            }

Machine d’état commande

CREATED
                            │
                            ├── published to MQTT
                            ▼
                            SENT
                            │
                            ├── device received
                            ▼
                            ACKED
                            │
                            ├── device starts execution
                            ▼
                            RUNNING
                            │
                            ├── success
                            ▼
                            DONE

                            Branches erreur :
                            SENT -> EXPIRED
                            SENT -> REJECTED
                            ACKED -> FAILED
                            RUNNING -> TIMEOUT
                            ANY -> CANCELLED

Symptôme → diagnostic

Checklist QoS avant mise en production

[ ] Chaque flux a un QoS documenté
                            [ ] Les raisons du QoS sont écrites
                            [ ] Les payloads importants ont message_id
                            [ ] Les devices critiques ont sequence_number
                            [ ] Le backend est idempotent pour QoS 1
                            [ ] Les commandes ont command_id
                            [ ] Les commandes ont ack et result
                            [ ] Les expirations sont définies pour commandes
                            [ ] Les shared subscriptions sont testées
                            [ ] Les queues offline ont des limites
                            [ ] Le lag d’ingestion est mesuré
                            [ ] Les doublons sont visibles dans l’admin
                            [ ] Les drops broker sont alertés
                            [ ] Les tests de coupure réseau sont faits
                            [ ] Les tests de retry sont faits
                            [ ] Les tests de crash worker sont faits

Test de chaos simple

Scénario :
                            1. Publier 1000 messages QoS 1 avec message_id.
                            2. Couper le worker backend après 300 messages.
                            3. Redémarrer le worker.
                            4. Vérifier :
                            - aucune double insertion métier
                            - duplicate_count augmente si retry
                            - lag revient à zéro
                            - erreurs visibles dans l’admin
                            - aucune commande critique exécutée deux fois

Convention recommandée

{tenant}/{site}/{area}/{asset}/{stream}/{signal}

                            Exemples :
                            acme/paris/building-a/boiler-07/telemetry/temperature
                            acme/paris/building-a/boiler-07/telemetry/pressure
                            acme/paris/building-a/boiler-07/status/availability
                            acme/paris/building-a/boiler-07/event/alarm
                            acme/paris/building-a/boiler-07/command/setpoint
                            acme/paris/building-a/boiler-07/ack/setpoint

Règles pratiques de nommage

Bon vs mauvais

Mauvais :
                            ACME/Paris/Bâtiment A/Chaudière 07/Température
                            device123
                            data
                            telemetry/all
                            factory/#/temperature

                            Bon :
                            acme/paris/building-a/boiler-07/telemetry/temperature
                            acme/paris/building-a/boiler-07/status/availability
                            acme/paris/building-a/boiler-07/event/alarm
                            acme/paris/building-a/boiler-07/command/setpoint

Les wildcards MQTT servent uniquement dans les subscriptions. Elles permettent de recevoir plusieurs topics correspondant à un motif. Elles sont très puissantes, mais dangereuses si elles sont trop larges.

Exemples

Subscription :
                            acme/+/+/+/telemetry/temperature

                            Reçoit :
                            acme/paris/line-01/motor-01/telemetry/temperature
                            acme/madrid/line-04/motor-09/telemetry/temperature

                            Ne reçoit pas :
                            acme/paris/line-01/motor-01/telemetry/vibration
                            acme/paris/line-01/motor-01/status/availability

Subscription :
                            acme/paris/#

                            Reçoit :
                            acme/paris/line-01/motor-01/telemetry/temperature
                            acme/paris/line-01/motor-01/status/availability
                            acme/paris/line-02/gw-07/event/alarm
                            acme/paris/building-a/boiler-07/command/setpoint

Risque des wildcards trop larges

Subscription dangereuse :
                            #

                            Effet :
                            - reçoit tout le broker
                            - peut aspirer énormément de trafic
                            - peut exposer des données sensibles
                            - peut saturer un client lent
                            - peut rendre le debug trompeur

                            À réserver :
                            - debug local
                            - compte admin temporaire
                            - environnement de test
                            - bridge très contrôlé

Bonnes pratiques wildcards

En architecture multi-tenant, le premier niveau du topic sert souvent à isoler les clients, organisations, environnements ou domaines métier. Cela facilite les ACL, le monitoring, les quotas, les bridges et les exports.

Modèle recommandé

{env}/{tenant}/{site}/{area}/{asset}/{stream}/{signal}

                            Exemples :
                            prod/acme/paris/line-02/motor-09/telemetry/temperature
                            prod/acme/paris/line-02/motor-09/status/availability
                            prod/acme/paris/line-02/motor-09/event/alarm

                            staging/acme/paris/line-02/motor-09/telemetry/temperature

ACL multi-tenant simplifiées

Tenant ACME backend :
                            allow subscribe: prod/acme/+/+/+/telemetry/#
                            allow subscribe: prod/acme/+/+/+/event/#
                            allow publish:   prod/acme/+/+/+/command/#

                            Tenant GLOBEX backend :
                            allow subscribe: prod/globex/+/+/+/telemetry/#
                            deny subscribe:  prod/acme/#

                            Device ACME gw-123 :
                            allow publish:   prod/acme/paris/line-02/gw-123/telemetry/#
                            allow publish:   prod/acme/paris/line-02/gw-123/status/#
                            allow subscribe: prod/acme/paris/line-02/gw-123/command/#
                            deny subscribe:  prod/acme/#

Stratégies d’isolation

Un topic design propre distingue les messages de mesure, les états courants et les événements. Ces trois familles ont des comportements différents : fréquence, stockage, retained, QoS, consommation et alerting.

Exemples propres

Telemetry :
                            acme/paris/line-02/motor-09/telemetry/temperature
                            acme/paris/line-02/motor-09/telemetry/vibration
                            acme/paris/line-02/motor-09/telemetry/rpm

                            Status :
                            acme/paris/line-02/motor-09/status/availability
                            acme/paris/line-02/motor-09/status/mode
                            acme/paris/line-02/motor-09/status/health

                            Events :
                            acme/paris/line-02/motor-09/event/alarm
                            acme/paris/line-02/motor-09/event/maintenance
                            acme/paris/line-02/motor-09/event/error

Diagramme : nature des flux

Telemetry
                            ├── nombreuses mesures
                            ├── stockage time-series
                            ├── agrégation possible
                            └── perte parfois acceptable

                            Status
                            ├── état courant
                            ├── retained utile
                            ├── dashboard immédiat
                            └── dernière valeur importante

                            Event
                            ├── événement daté
                            ├── audit / historique
                            ├── pas forcément retained
                            └── idempotence importante

                            Command
                            ├── ordre envoyé au device
                            ├── command_id obligatoire
                            ├── expiration recommandée
                            └── ack/result requis

Payloads types

Telemetry payload :
                            {
                            "schema_version": "1.0",
                            "message_id": "msg-001",
                            "value": 72.4,
                            "unit": "C",
                            "ts": "2026-04-25T10:00:00Z"
                            }

                            Status payload :
                            {
                            "schema_version": "1.0",
                            "state": "online",
                            "since": "2026-04-25T09:58:00Z",
                            "firmware": "1.8.2"
                            }

                            Event payload :
                            {
                            "schema_version": "1.0",
                            "event_id": "evt-001",
                            "severity": "critical",
                            "code": "OVERHEAT",
                            "ts": "2026-04-25T10:00:00Z"
                            }

Les commandes doivent être traitées différemment de la télémétrie. Une commande est une intention d’action. Elle doit avoir un identifiant, une expiration, un statut, un accusé de réception et idéalement un résultat d’exécution.

Topics recommandés

Commande :
                            acme/paris/line-02/gw-123/command/reboot
                            acme/paris/line-02/gw-123/command/setpoint
                            acme/paris/line-02/gw-123/command/config-update

                            Ack :
                            acme/paris/line-02/gw-123/ack/reboot
                            acme/paris/line-02/gw-123/ack/setpoint
                            acme/paris/line-02/gw-123/ack/config-update

                            Résultat :
                            acme/paris/line-02/gw-123/result/reboot
                            acme/paris/line-02/gw-123/result/setpoint

Payload commande

{
                            "schema_version": "1.0",
                            "command_id": "cmd-20260425-0001",
                            "action": "setpoint",
                            "value": 21.5,
                            "unit": "C",
                            "requested_by": "backend-rules-engine",
                            "expires_at": "2026-04-25T10:01:00Z"
                            }

Flux command / ack / result

Backend
                            │
                            │ PUBLISH command/setpoint
                            ▼
                            Broker MQTT
                            │
                            ▼
                            Device
                            │
                            ├── vérifie command_id
                            ├── vérifie expiration
                            ├── vérifie état local
                            ├── publie ack/accepted ou ack/rejected
                            └── exécute si accepté
                            │
                            ▼
                            publie result/done ou result/failed

Machine d’état commande

CREATED
                            │
                            ▼
                            PUBLISHED
                            │
                            ├── ACK accepted
                            ▼
                            ACKED
                            │
                            ├── execution started
                            ▼
                            RUNNING
                            │
                            ├── success
                            ▼
                            DONE

                            Branches :
                            PUBLISHED -> EXPIRED
                            PUBLISHED -> REJECTED
                            ACKED     -> FAILED
                            RUNNING   -> TIMEOUT
                            ANY       -> CANCELLED

Le flag retained demande au broker de conserver le dernier message publié sur un topic. Quand un nouveau subscriber s’abonne, il reçoit immédiatement cette dernière valeur. C’est parfait pour les états courants, mais dangereux si utilisé comme historique.

Exemple retained status

Topic :
                            acme/paris/line-02/gw-123/status/availability

                            Retained payload :
                            {
                            "schema_version": "1.0",
                            "state": "online",
                            "since": "2026-04-25T09:58:00Z",
                            "firmware": "1.8.2"
                            }

Retained : comportement

1. Device publie retained :
                            status/availability = online

                            2. Broker conserve cette dernière valeur.

                            3. Dashboard démarre plus tard.

                            4. Dashboard SUBSCRIBE status/availability.

                            5. Broker envoie immédiatement :
                            online

                            6. Dashboard n’a pas besoin d’attendre
                            le prochain message device.

Retained vs queue vs historique

Les ACL MQTT doivent être pensées dès le design des topics. Un bon namespace rend les règles simples. Un mauvais namespace oblige à écrire des ACL complexes, fragiles et dangereuses.

Exemple ACL device

Device gw-123 :

                            allow publish:
                            acme/paris/line-02/gw-123/telemetry/#
                            acme/paris/line-02/gw-123/status/#
                            acme/paris/line-02/gw-123/event/#

                            allow subscribe:
                            acme/paris/line-02/gw-123/command/#

                            deny subscribe:
                            acme/#

                            deny publish:
                            acme/paris/line-02/other-device/#

Exemple ACL backend ingestion

Backend ingestion :

                            allow subscribe:
                            acme/+/+/+/telemetry/#
                            acme/+/+/+/status/#
                            acme/+/+/+/event/#

                            deny publish:
                            acme/+/+/+/command/#

                            allow publish only for command service:
                            acme/+/+/+/command/#

Table des rôles MQTT

Diagramme ACL

Device gw-123
                            ├── publish telemetry/status/event
                            └── subscribe command

                            Backend ingestion
                            └── subscribe telemetry/status/event

                            Command service
                            ├── publish command
                            └── subscribe ack/result

                            Dashboard
                            └── subscribe périmètre utilisateur

                            Admin
                            └── debug limité, temporaire, audité

Les systèmes MQTT vivent longtemps : devices, gateways et dashboards peuvent être mis à jour à des rythmes différents. Il faut donc une stratégie de versioning. Le plus souvent, on versionne le payload. On versionne le topic seulement si la rupture est structurelle.

Payload versionné

{
                            "schema_version": "2.0",
                            "message_id": "msg-20260425-001",
                            "device_id": "motor-09",
                            "metric": "temperature",
                            "value": 72.4,
                            "unit": "C",
                            "quality": "good",
                            "ts": "2026-04-25T10:00:00Z"
                            }

Migration sans rupture

Phase 1 :
                            Device publie ancien payload v1
                            Backend accepte v1

                            Phase 2 :
                            Device publie v1 + nouveaux champs compatibles
                            Backend accepte v1 enrichi

                            Phase 3 :
                            Device publie schema_version v2
                            Backend accepte v1 et v2

                            Phase 4 :
                            Dashboard compatible v2
                            Monitoring des consommateurs restants v1

                            Phase 5 :
                            Décommission v1 documentée

Breaking changes typiques

Checklist qualité topic design

[ ] Le topic permet de filtrer par tenant
                            [ ] Le topic permet de filtrer par site
                            [ ] Le topic permet de filtrer par asset
                            [ ] Le topic distingue telemetry/status/event/command
                            [ ] Les commands ont ack/result
                            [ ] Les commands ne sont pas retained
                            [ ] Les status importants peuvent être retained
                            [ ] Les events sont historisés en DB
                            [ ] Le payload contient schema_version
                            [ ] Le payload contient message_id si important
                            [ ] Les noms sont lowercase ASCII
                            [ ] Les ACL sont simples à écrire
                            [ ] Les wildcards larges sont évitées
                            [ ] Les conventions sont documentées
                            [ ] Les migrations v1/v2 sont prévues

Design final recommandé

prod/{tenant}/{site}/{area}/{asset}/telemetry/{metric}
                            prod/{tenant}/{site}/{area}/{asset}/status/{state_name}
                            prod/{tenant}/{site}/{area}/{asset}/event/{event_type}
                            prod/{tenant}/{site}/{area}/{asset}/command/{command_name}
                            prod/{tenant}/{site}/{area}/{asset}/ack/{command_name}
                            prod/{tenant}/{site}/{area}/{asset}/result/{command_name}

Mosquitto est souvent le meilleur point de depart. Il est simple a installer, tres leger, tres documente, parfait pour un lab, une VM, une gateway locale, un Raspberry Pi, ou une petite production controlee.

Configuration TLS + password + ACL

# /etc/mosquitto/conf.d/prod.conf

                            listener 8883
                            protocol mqtt

                            cafile /etc/mosquitto/certs/ca.crt
                            certfile /etc/mosquitto/certs/server.crt
                            keyfile /etc/mosquitto/certs/server.key

                            require_certificate false
                            allow_anonymous false

                            password_file /etc/mosquitto/passwd
                            acl_file /etc/mosquitto/acl

                            persistence true
                            persistence_location /var/lib/mosquitto/

                            log_type error
                            log_type warning
                            log_type notice
                            log_type information

ACL Mosquitto typique

# /etc/mosquitto/acl

                            user gw_123
                            topic write acme/paris/line-02/gw-123/telemetry/#
                            topic write acme/paris/line-02/gw-123/status/#
                            topic write acme/paris/line-02/gw-123/event/#
                            topic read  acme/paris/line-02/gw-123/command/#

                            user ingestion_worker
                            topic read acme/+/+/+/telemetry/#
                            topic read acme/+/+/+/status/#
                            topic read acme/+/+/+/event/#

                            user command_service
                            topic write acme/+/+/+/command/#
                            topic read  acme/+/+/+/ack/#
                            topic read  acme/+/+/+/result/#

Commandes utiles

# Creer un utilisateur
                            sudo mosquitto_passwd -c /etc/mosquitto/passwd gw_123

                            # Ajouter un utilisateur
                            sudo mosquitto_passwd /etc/mosquitto/passwd ingestion_worker

                            # Tester subscribe
                            mosquitto_sub -h localhost -p 8883 \
                            --cafile ca.crt \
                            -u ingestion_worker -P password \
                            -t 'acme/+/+/+/telemetry/#' -v

                            # Tester publish
                            mosquitto_pub -h localhost -p 8883 \
                            --cafile ca.crt \
                            -u gw_123 -P password \
                            -t 'acme/paris/line-02/gw-123/status/availability' \
                            -m '{"state":"online"}' -q 1 -r

EMQX est un broker MQTT moderne oriente scalabilite, clustering, MQTT 5, dashboards, plugins d’authentification, integrations externes et rules engine. Il devient logique quand le broker n’est plus seulement un point de passage, mais une plateforme IoT.

Quand EMQX devient pertinent

• Nombre eleve de connexions simultanees.
• Besoin de cluster actif.
• Besoin de dashboard broker integre.
• Auth externe : HTTP, JWT, LDAP, SQL, OAuth selon edition/plugins.
• Rules engine pour pousser vers Kafka, SQL, HTTP, S3 ou autres integrations.
• Quotas, rate limits, observabilite et MQTT 5 avance.
• Multi-tenant ou segmentation forte des clients.

Architecture EMQX typique

                 ┌─────────────────────┐
                            Devices ────────►│  TCP/TLS Load Balancer│
                            └──────────┬──────────┘
                            │
                            ┌─────────────────┼─────────────────┐
                            ▼                 ▼                 ▼
                            ┌─────────────┐   ┌─────────────┐   ┌─────────────┐
                            │ EMQX Node 1 │   │ EMQX Node 2 │   │ EMQX Node 3 │
                            └──────┬──────┘   └──────┬──────┘   └──────┬──────┘
                            │                 │                 │
                            └─────────────────┼─────────────────┘
                            ▼
                            ┌─────────────────┐
                            │ Rules / Bridges │
                            └───────┬─────────┘
                            │
                            ┌───────────────────┼────────────────────┐
                            ▼                   ▼                    ▼
                            Kafka              PostgreSQL              HTTP
                            S3                 ClickHouse              Webhooks

Points de vigilance

HiveMQ est tres oriente production enterprise : clustering, support MQTT 5, observabilite, extension SDK, integrations, securite et support professionnel. C’est un choix naturel dans des environnements ou le broker devient un composant critique avec SLA, support et gouvernance.

Forces de HiveMQ

• Orientation enterprise et production critique.
• Support fort de MQTT 5.
• Cluster et haute disponibilite.
• Extension SDK pour personnaliser authentification, autorisation, integration.
• Observabilite et integrations monitoring.
• Approche professionnelle pour grands deploiements IoT.

Architecture HiveMQ enterprise

Devices
                            │
                            ▼
                            Load Balancer TCP/TLS
                            │
                            ├── HiveMQ Node A
                            ├── HiveMQ Node B
                            └── HiveMQ Node C
                            │
                            ├── Extension Auth / ACL
                            ├── Extension Integration
                            ├── Metrics / Monitoring
                            └── Data pipeline
                            ├── Kafka
                            ├── Database
                            ├── Webhook
                            └── Analytics

Quand choisir HiveMQ

VerneMQ est un broker MQTT distribue, historiquement apprecie pour les architectures cluster et la robustesse de l’ecosysteme Erlang. Il peut etre interessant si tu veux un broker open-source distribue, avec plugins et capacite de cluster.

Quand VerneMQ est pertinent

• Tu veux un broker distribue open-source.
• Tu as besoin de cluster MQTT.
• Tu acceptes d’exploiter un composant Erlang.
• Tu veux tester une alternative a EMQX/HiveMQ.
• Tu as des besoins de plugins ou d’auth custom.

Topologie cluster VerneMQ

             ┌─────────────────────┐
                            Devices ────►│ Load Balancer MQTT  │
                            └──────────┬──────────┘
                            │
                            ┌───────────────┼───────────────┐
                            ▼               ▼               ▼
                            ┌──────────────┐ ┌──────────────┐ ┌──────────────┐
                            │ VerneMQ N1   │ │ VerneMQ N2   │ │ VerneMQ N3   │
                            └──────┬───────┘ └──────┬───────┘ └──────┬───────┘
                            │                │                │
                            └────────────────┼────────────────┘
                            ▼
                            Cluster state / routing
                            │
                            ▼
                            Backend subscribers / bridges

Points d’exploitation

RabbitMQ n’est pas d’abord un broker MQTT pur. C’est un broker de messaging enterprise AMQP avec exchanges, queues, routing keys, dead-letter, acknowledgements et plugins. Le plugin MQTT permet a des clients MQTT de publier/recevoir via RabbitMQ, ce qui est tres utile si ton SI exploite deja RabbitMQ.

Cas d’usage pertinent

• RabbitMQ existe deja dans l’architecture.
• Les devices doivent publier en MQTT.
• Les services internes consomment via AMQP.
• Tu veux utiliser exchanges, queues, dead-letter, retry.
• Tu veux brancher ingestion, stockage, alerting et WebSocket via queues separees.

Architecture RabbitMQ + MQTT

MQTT Devices
                            │
                            │ MQTT publish
                            ▼
                            RabbitMQ MQTT Listener
                            │
                            ▼
                            AMQP Exchange
                            │
                            ├── Queue storage
                            │       └── PostgreSQL / TimescaleDB worker
                            │
                            ├── Queue alerts
                            │       └── rule engine
                            │
                            ├── Queue websocket
                            │       └── realtime dashboard
                            │
                            └── Dead-letter queue
                            └── failed messages / replay

Forces et limites

Les services cloud IoT ne sont pas seulement des brokers MQTT. Ils ajoutent souvent un registre devices, une gestion d’identite, certificats, policies, rules engine, integration avec stockage, serverless, monitoring et securite cloud.

AWS IoT Core : pipeline typique

Devices
                            │
                            │ MQTT/TLS + certificates
                            ▼
                            AWS IoT Core
                            │
                            ├── IoT Policy
                            ├── Device Registry
                            ├── Rules Engine
                            │       ├── Lambda
                            │       ├── Kinesis
                            │       ├── DynamoDB
                            │       ├── S3
                            │       └── SNS/SQS
                            │
                            └── Monitoring / Logs

Azure IoT Hub : pipeline typique

Devices
                            │
                            │ MQTT/TLS
                            ▼
                            Azure IoT Hub
                            │
                            ├── Device Identity
                            ├── Device-to-cloud messages
                            ├── Cloud-to-device messages
                            ├── Device Twin
                            ├── Routes
                            │       ├── Event Hubs
                            │       ├── Storage
                            │       ├── Service Bus
                            │       └── Functions
                            │
                            └── Monitoring / Defender / Logs

Cloud managed vs self-hosted

La haute disponibilite depend du broker choisi. Un Mosquitto simple peut etre suffisant pour un edge local, mais un service central critique a souvent besoin d’un cluster, d’un load balancer, de monitoring, de sauvegardes de configuration et de procedures de failover.

Architecture edge avec bridge

Site industriel
                            │
                            ├── Sensors / PLC / Machines
                            │
                            ├── Edge Gateway
                            │
                            └── Local MQTT Broker
                            │
                            ├── Local dashboard
                            ├── Local rules
                            └── Bridge MQTT/TLS
                            │
                            ▼
                            Central Broker / Cloud
                            │
                            ├── Data lake
                            ├── Time-series DB
                            ├── Alerting
                            └── Global dashboard

Architecture cluster broker

Devices
                            │
                            ▼
                            TCP/TLS Load Balancer
                            │
                            ├── Broker Node 1
                            ├── Broker Node 2
                            └── Broker Node 3
                            │
                            ├── Session / subscription management
                            ├── Metrics / logs
                            ├── Auth / ACL
                            └── Bridges / connectors
                            │
                            ├── Kafka
                            ├── PostgreSQL
                            ├── Redis
                            ├── S3
                            └── Webhooks

Checklist HA broker

[ ] Load balancer TCP/TLS configure
                            [ ] Health checks broker
                            [ ] Certificats TLS renouvelables
                            [ ] Config broker versionnee
                            [ ] ACL sauvegardees
                            [ ] Persistence configuree si necessaire
                            [ ] Retained messages strategy
                            [ ] Quotas clients
                            [ ] Monitoring CPU/RAM/FD
                            [ ] Metrics messages in/out
                            [ ] Alertes connect/disconnect
                            [ ] Test failover node
                            [ ] Test reconnexion clients
                            [ ] Test bridge loop prevention
                            [ ] Plan de rollback config

Matrice de choix rapide

Scorecard broker

Noter chaque broker de 1 a 5 :

                            [ ] Simplicite installation
                            [ ] Support MQTT 5
                            [ ] TLS / certificats
                            [ ] ACL fines
                            [ ] Auth externe
                            [ ] Cluster / HA
                            [ ] Observabilite
                            [ ] Rules engine
                            [ ] Integrations data
                            [ ] Cout licence
                            [ ] Cout operations
                            [ ] Competence equipe
                            [ ] Documentation
                            [ ] Support vendor
                            [ ] Portabilite

Runbook d’evaluation en 2 jours

Jour 1 : fonctionnalite
                            1. Installer broker
                            2. Activer TLS
                            3. Creer 3 users : device, ingestion, command
                            4. Configurer ACL
                            5. Tester publish telemetry
                            6. Tester subscribe backend
                            7. Tester retained status
                            8. Tester Last Will
                            9. Tester QoS 1 et doublons
                            10. Tester logs et metrics

                            Jour 2 : production readiness
                            1. Redemarrer broker
                            2. Verifier persistence
                            3. Simuler coupure device
                            4. Simuler subscriber lent
                            5. Charger 10k messages
                            6. Tester wildcard controle
                            7. Tester rotation certificats
                            8. Exporter configuration
                            9. Configurer alertes
                            10. Rediger decision finale

Conclusion pratique

URLs utiles

Mosquitto:
                            https://mosquitto.org/

                            EMQX:
                            https://www.emqx.com/

                            HiveMQ:
                            https://www.hivemq.com/

                            VerneMQ:
                            https://vernemq.com/

                            RabbitMQ MQTT plugin:
                            https://www.rabbitmq.com/docs/mqtt

                            AWS IoT Core:
                            https://docs.aws.amazon.com/iot/

                            Azure IoT Hub:
                            https://learn.microsoft.com/azure/iot-hub/

TLS server authentication

With standard TLS, the client validates the broker certificate. This protects the connection against passive sniffing and man-in-the-middle attacks, provided the client correctly validates the broker certificate chain and hostname.

Client validates broker certificate:

                            Device
                            │
                            │ TLS handshake
                            │ verify broker certificate
                            │ verify CA
                            │ verify hostname if applicable
                            ▼
                            MQTT Broker

                            Client authentication may still be:
                            - username/password
                            - JWT token
                            - API key
                            - external auth plugin

mTLS mutual authentication

With mTLS, both sides validate certificates. The client validates the broker certificate, and the broker validates the client certificate. This is the preferred model for serious industrial and IoT deployments.

mTLS handshake:

                            Device certificate
                            │
                            ▼
                            ┌─────────────────┐       validates broker cert       ┌─────────────────┐
                            │ MQTT Client     │ ─────────────────────────────────► │ MQTT Broker     │
                            │ device gw-001   │                                   │ listener 8883   │
                            └─────────────────┘ ◄─────────────────────────────────┘
                            ▲             validates client certificate
                            │
                            └── device identity mapped to:
                            - client_id
                            - tenant
                            - site
                            - allowed topics
                            - device registry record

Recommended listener policy

Public or semi-public network:
                            - expose 8883 only
                            - require TLS
                            - prefer mTLS
                            - disable anonymous access
                            - disable public 1883
                            - restrict by firewall
                            - log all auth failures

                            Private lab:
                            - 1883 allowed only inside trusted network
                            - never expose 1883 to the Internet
                            - use test credentials only
                            - no production devices

MQTT identity should not be improvised. A secure deployment ties the MQTT client ID to a device registry entry, a certificate or token, and a strict topic namespace. A device should not be able to choose another device identity by changing its client ID.

Identity mapping rule

client_id = gw-paris-001

                            Certificate subject:
                            CN = gw-paris-001

                            Device registry:
                            device_id = gw-paris-001
                            tenant = acme
                            site = paris
                            role = gateway
                            status = active

                            Allowed namespace:
                            prod/acme/paris/+/gw-paris-001/#

Secure connection decision flow

Client CONNECT
                            │
                            ▼
                            Validate TLS
                            │
                            ▼
                            Validate client certificate or token
                            │
                            ▼
                            Extract identity
                            │
                            ▼
                            Lookup device registry
                            │
                            ├── not found      -> reject
                            ├── disabled       -> reject
                            ├── expired cert   -> reject
                            └── active         -> continue
                            │
                            ▼
                            Load ACL profile
                            │
                            ▼
                            Accept connection

Identity anti-patterns

MQTT ACLs must be designed around topic namespaces. The default posture should be deny all. Each role receives the minimum publish and subscribe rights required. Devices should publish only under their own namespace and subscribe only to their own command topics.

ACL principles

deny all by default

                            allow device:
                            - publish own telemetry
                            - publish own status
                            - publish own events
                            - subscribe own commands

                            allow ingestion backend:
                            - subscribe telemetry
                            - subscribe status
                            - subscribe events
                            - no command publishing

                            allow command service:
                            - publish command topics
                            - subscribe ack/result topics

                            allow dashboard:
                            - subscribe only authorized tenant/site scope

                            log:
                            - denied publish
                            - denied subscribe
                            - auth failures
                            - unexpected wildcard attempts

ACL example

Device gw-123:

                            allow publish:
                            prod/acme/paris/line-02/gw-123/telemetry/#
                            prod/acme/paris/line-02/gw-123/status/#
                            prod/acme/paris/line-02/gw-123/event/#

                            allow subscribe:
                            prod/acme/paris/line-02/gw-123/command/#

                            deny subscribe:
                            prod/acme/#

                            deny publish:
                            prod/acme/paris/line-02/other-device/#

Role matrix

Dangerous subscriptions

High risk:
                            #

                            High risk:
                            prod/#

                            Still broad:
                            prod/acme/#

                            Safer:
                            prod/acme/paris/+/+/telemetry/temperature

                            Very specific:
                            prod/acme/paris/line-02/gw-123/status/availability

Certificates are powerful, but they create lifecycle responsibilities. A secure MQTT deployment needs a clear process for certificate generation, provisioning, storage, renewal, rotation, revocation and incident response.

Registry fields

Device registry fields:
                            - device_id
                            - tenant_id
                            - site_id
                            - client_id
                            - certificate_fingerprint
                            - certificate_serial
                            - certificate_not_before
                            - certificate_not_after
                            - status: active / disabled / revoked
                            - last_seen_at
                            - last_ip
                            - firmware_version
                            - owner
                            - rotation_due_at

Certificate rotation flow

Rotation workflow:

                            1. Mark device as rotation_due.
                            2. Generate new certificate.
                            3. Provision new certificate securely.
                            4. Allow old and new certificate during grace window.
                            5. Device reconnects with new certificate.
                            6. Confirm new fingerprint in registry.
                            7. Revoke old certificate.
                            8. Remove old certificate from trusted set.
                            9. Audit rotation event.
                            10. Alert if old certificate is still used.

Lost device incident

Lost or compromised device:

                            1. Set registry status = revoked.
                            2. Revoke certificate.
                            3. Remove active session from broker.
                            4. Block client_id and certificate fingerprint.
                            5. Search recent publish activity.
                            6. Search command activity.
                            7. Invalidate pending commands.
                            8. Notify tenant/site owner.
                            9. Rotate related secrets if shared.
                            10. Keep forensic logs.

MQTT does not validate the business payload. A malicious or broken client can send malformed JSON, huge payloads, unexpected types, missing fields, invalid timestamps, command injection strings or data that causes backend errors.

Safe payload contract

Telemetry payload:
                            {
                            "schema_version": "1.0",
                            "message_id": "msg-20260425-001",
                            "device_id": "gw-001",
                            "metric": "temperature",
                            "value": 21.7,
                            "unit": "C",
                            "ts": "2026-04-25T10:00:00Z"
                            }

                            Validation rules:
                            - schema_version required
                            - message_id required for QoS 1
                            - device_id must match authenticated identity
                            - metric must be allowed
                            - value must match metric type
                            - timestamp must be sane
                            - payload size must be limited

Validation pipeline

MQTT PUBLISH
                            │
                            ▼
                            Check topic ACL
                            │
                            ▼
                            Check payload size
                            │
                            ▼
                            Parse payload safely
                            │
                            ▼
                            Validate schema version
                            │
                            ▼
                            Validate identity consistency
                            │
                            ▼
                            Validate metric allowlist
                            │
                            ▼
                            Build idempotency key
                            │
                            ▼
                            Write to buffer or database
                            │
                            ▼
                            Expose failures in admin

Backend failure table

MQTT brokers can be overloaded by connection storms, reconnect loops, message floods, broad subscriptions, oversized payloads or too many retained messages. Rate limiting and quotas protect the broker and the backend.

Flood symptoms

• Broker CPU suddenly high.
• Connection rate spikes.
• Authentication failures increase.
• Message rate jumps unexpectedly.
• Queued messages grow continuously.
• Backend ingestion lag increases.
• Subscribers disconnect repeatedly.

Protection architecture

Internet / WAN
                            │
                            ▼
                            Firewall / Security Group
                            │
                            ├── allow trusted IP ranges
                            ├── block known bad sources
                            └── expose 8883 only
                            │
                            ▼
                            Load Balancer TCP/TLS
                            │
                            ├── connection limits
                            ├── health checks
                            └── TLS policy if terminated here
                            │
                            ▼
                            MQTT Broker
                            │
                            ├── auth
                            ├── ACL
                            ├── quotas
                            ├── max packet size
                            ├── max inflight
                            └── rate limits
                            │
                            ▼
                            Backend buffer
                            │
                            ├── Redis Stream
                            ├── Kafka
                            └── worker queue

Alerting table

MQTT should be placed in a controlled network architecture. The broker should not have unrestricted access to all internal systems. Devices, brokers, backend workers, databases and admin tools should live in separated zones with explicit flows.

Rule of thumb

Devices should never connect directly to:
                            - PostgreSQL
                            - Redis
                            - internal admin APIs
                            - command database
                            - broker admin UI
                            - cloud credentials
                            - internal message queues

                            Devices should only talk to:
                            - MQTT listener
                            - controlled provisioning endpoint if needed

Network diagram

┌───────────────────────────────┐
                            │ Device / Edge Zone             │
                            │ gateways, sensors, PLC bridge  │
                            └───────────────┬───────────────┘
                            │ MQTT/TLS 8883 only
                            ▼
                            ┌───────────────────────────────┐
                            │ MQTT Broker Zone               │
                            │ broker nodes, auth, ACL        │
                            └───────────────┬───────────────┘
                            │ controlled subscribe/publish
                            ▼
                            ┌───────────────────────────────┐
                            │ Backend Zone                   │
                            │ ingestion, command, API        │
                            └───────────────┬───────────────┘
                            │ DB/Kafka access
                            ▼
                            ┌───────────────────────────────┐
                            │ Data Zone                      │
                            │ PostgreSQL, Kafka, S3, Redis   │
                            └───────────────────────────────┘

                            Admin access:
                            VPN + MFA + RBAC + audit logs

Segmentation checklist

[ ] MQTT listener exposed only where needed
                            [ ] 1883 disabled or private-only
                            [ ] 8883 protected by firewall
                            [ ] Broker admin UI not public
                            [ ] Backend databases not reachable from devices
                            [ ] Command service separated from ingestion
                            [ ] Admin access via VPN or private network
                            [ ] Security groups are least-privilege
                            [ ] Logs are centralized
                            [ ] Secrets are not stored on public images

Commands are more dangerous than telemetry. A fake telemetry message can pollute data, but a fake command can reboot a gateway, change a setpoint, stop a machine or trigger unsafe behavior. Command topics require stricter controls.

Secure command payload

{
                            "schema_version": "1.0",
                            "command_id": "cmd-20260425-001",
                            "action": "setpoint",
                            "value": 21.5,
                            "unit": "C",
                            "requested_by": "rules-engine",
                            "reason": "energy-optimization",
                            "expires_at": "2026-04-25T10:01:00Z"
                            }

Command flow with audit

Operator / Rule Engine
                            │
                            ▼
                            Command Service
                            │
                            ├── RBAC check
                            ├── device state check
                            ├── command policy check
                            ├── create command_id
                            ├── write command journal
                            └── publish MQTT command
                            │
                            ▼
                            MQTT Broker
                            │
                            ▼
                            Device
                            │
                            ├── validate command
                            ├── check expiry
                            ├── publish ack
                            ├── execute
                            └── publish result

Command topic ACL

Device gw-123:
                            allow subscribe:
                            prod/acme/paris/line-02/gw-123/command/#

                            allow publish:
                            prod/acme/paris/line-02/gw-123/ack/#
                            prod/acme/paris/line-02/gw-123/result/#

                            deny publish:
                            prod/acme/paris/line-02/gw-123/command/#

                            Command service:
                            allow publish:
                            prod/acme/+/+/+/command/#

                            allow subscribe:
                            prod/acme/+/+/+/ack/#
                            prod/acme/+/+/+/result/#

MQTT broker hardening

[ ] Disable anonymous access
                            [ ] Disable public plaintext 1883
                            [ ] Enable TLS on 8883
                            [ ] Prefer mTLS for devices
                            [ ] Use per-device identity
                            [ ] Bind client_id to certificate or registry
                            [ ] Enforce deny-by-default ACLs
                            [ ] Restrict wildcard subscriptions
                            [ ] Set max payload size
                            [ ] Set connection limits
                            [ ] Set message rate limits
                            [ ] Set max subscriptions per client
                            [ ] Set inflight limits
                            [ ] Log auth failures
                            [ ] Log denied publish/subscribe
                            [ ] Monitor broker metrics
                            [ ] Protect admin UI with VPN/MFA
                            [ ] Version broker configuration
                            [ ] Backup ACL and certificate configuration

Backend hardening

[ ] Validate every payload schema
                            [ ] Enforce device_id identity consistency
                            [ ] Add idempotency keys
                            [ ] Store ingestion failures visibly
                            [ ] Reject unknown metrics
                            [ ] Reject stale or future timestamps
                            [ ] Audit command creation
                            [ ] Require command_id
                            [ ] Require command expiration
                            [ ] Require ack and result for commands
                            [ ] Keep command journal
                            [ ] Separate ingestion service and command service
                            [ ] Do not trust topic alone
                            [ ] Do not trust payload alone

Incident response checklist

Suspected compromised device:

                            1. Disable device in registry.
                            2. Revoke certificate or token.
                            3. Disconnect active broker session.
                            4. Block client_id and fingerprint.
                            5. Search recent telemetry from device.
                            6. Search command activity.
                            7. Invalidate pending commands.
                            8. Check same firmware batch.
                            9. Rotate shared secrets if any.
                            10. Preserve logs for forensics.

                            Suspected credential leak:

                            1. Rotate credential immediately.
                            2. Identify all affected clients.
                            3. Review auth failure logs.
                            4. Review unexpected topic access.
                            5. Tighten ACL if needed.
                            6. Add alert rule for reuse attempt.

Security acceptance test

Paho is the most common Python MQTT client. For a backend ingestion service, it is usually started from a Django management command or a dedicated daemon. The callback receives the MQTT message and quickly pushes it to a buffer.

import json
                            import logging
                            import ssl
                            import time
                            from typing import Any, Dict

                            from paho.mqtt import client as mqtt

                            LOG = logging.getLogger("mqtt_ingestion")

                            BROKER_HOST = "mqtt.example.com"
                            BROKER_PORT = 8883
                            CLIENT_ID = "backend-ingestion-01"

                            TOPICS = [
                            ("prod/acme/+/+/+/telemetry/#", 1),
                            ("prod/acme/+/+/+/status/#", 1),
                            ("prod/acme/+/+/+/event/#", 1),
                            ]


                            def build_record(msg: mqtt.MQTTMessage) -> Dict[str, Any]:
                            payload_raw = msg.payload.decode("utf-8", errors="replace")

                            return {
                            "topic": msg.topic,
                            "qos": msg.qos,
                            "retain": bool(msg.retain),
                            "payload_raw": payload_raw,
                            "received_at_epoch": time.time(),
                            }


                            def enqueue_record(record: Dict[str, Any]) -> None:
                            """
                            Push to Redis Stream, Celery, Kafka or a staging table.
                            This function must stay fast and reliable.
                            """
                            LOG.info("enqueue mqtt topic=%s qos=%s", record["topic"], record["qos"])


                            def on_connect(client, userdata, flags, reason_code, properties=None):
                            LOG.info("mqtt connected reason=%s flags=%s", reason_code, flags)
                            for topic, qos in TOPICS:
                            result, mid = client.subscribe(topic, qos=qos)
                            LOG.info("subscribe topic=%s qos=%s result=%s mid=%s", topic, qos, result, mid)


                            def on_disconnect(client, userdata, reason_code, properties=None):
                            LOG.warning("mqtt disconnected reason=%s", reason_code)


                            def on_message(client, userdata, msg):
                            try:
                            record = build_record(msg)

                            if len(record["payload_raw"]) > 65536:
                            LOG.warning("mqtt payload too large topic=%s", msg.topic)
                            return

                            enqueue_record(record)

                            except Exception:
                            LOG.exception("mqtt callback failed topic=%s", getattr(msg, "topic", "unknown"))


                            def create_client() -> mqtt.Client:
                            client = mqtt.Client(
                            client_id=CLIENT_ID,
                            protocol=mqtt.MQTTv5,
                            callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
                            )

                            client.tls_set(cert_reqs=ssl.CERT_REQUIRED)
                            client.username_pw_set("backend_ingestion", "CHANGE_ME")

                            client.on_connect = on_connect
                            client.on_disconnect = on_disconnect
                            client.on_message = on_message

                            client.reconnect_delay_set(min_delay=1, max_delay=60)
                            return client


                            def main() -> None:
                            logging.basicConfig(level=logging.INFO)

                            client = create_client()
                            client.connect(BROKER_HOST, BROKER_PORT, keepalive=60)
                            client.loop_forever()


                            if __name__ == "__main__":
                            main()

Subscriber production rules

Common failure modes

Failure:
                            Database is slow.

                            Bad design:
                            on_message writes directly to database.

                            Result:
                            MQTT callback blocks.
                            Client cannot consume fast enough.
                            Broker queues grow.
                            Messages are delayed or dropped.

                            Better design:
                            on_message pushes to Redis Stream.
                            Workers write to database in batches.
                            Failures are stored and visible in admin.

A publisher can represent a device, a gateway, an edge agent or a backend command service. For telemetry, the payload should include schema version, message ID, device ID, metric, value, unit and timestamp.

import json
                            import ssl
                            import time
                            import uuid
                            from datetime import datetime, timezone

                            from paho.mqtt import client as mqtt

                            BROKER_HOST = "mqtt.example.com"
                            BROKER_PORT = 8883

                            TENANT = "acme"
                            SITE = "paris"
                            AREA = "line-02"
                            DEVICE_ID = "gw-123"

                            CLIENT_ID = DEVICE_ID
                            USERNAME = "gw_123"
                            PASSWORD = "CHANGE_ME"


                            def utc_now() -> str:
                            return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")


                            def build_payload(sequence: int) -> dict:
                            return {
                            "schema_version": "1.0",
                            "message_id": str(uuid.uuid4()),
                            "device_id": DEVICE_ID,
                            "sequence": sequence,
                            "metric": "temperature",
                            "value": 21.8,
                            "unit": "C",
                            "ts": utc_now(),
                            }


                            def build_topic() -> str:
                            return f"prod/{TENANT}/{SITE}/{AREA}/{DEVICE_ID}/telemetry/temperature"


                            client = mqtt.Client(
                            client_id=CLIENT_ID,
                            protocol=mqtt.MQTTv5,
                            callback_api_version=mqtt.CallbackAPIVersion.VERSION2,
                            )

                            client.tls_set(cert_reqs=ssl.CERT_REQUIRED)
                            client.username_pw_set(USERNAME, PASSWORD)

                            client.connect(BROKER_HOST, BROKER_PORT, keepalive=60)
                            client.loop_start()

                            sequence = 0

                            try:
                            while True:
                            sequence += 1
                            payload = build_payload(sequence)
                            topic = build_topic()

                            result = client.publish(
                            topic,
                            json.dumps(payload, separators=(",", ":")),
                            qos=1,
                            retain=False,
                            )

                            result.wait_for_publish(timeout=5)
                            print(f"published topic={topic} mid={result.mid} rc={result.rc}")

                            time.sleep(10)

                            finally:
                            client.loop_stop()
                            client.disconnect()

Publisher design checklist

Status publisher with retained message

status_topic = "prod/acme/paris/line-02/gw-123/status/availability"

                            online_payload = {
                            "schema_version": "1.0",
                            "device_id": "gw-123",
                            "state": "online",
                            "ts": utc_now(),
                            }

                            client.publish(
                            status_topic,
                            json.dumps(online_payload),
                            qos=1,
                            retain=True,
                            )

In Django, the cleanest approach is to isolate MQTT ingestion in a dedicated app. The MQTT service should be executable as a management command, supervised by systemd, Docker, Kubernetes or your orchestrator.

mqtt_app/
                            __init__.py

                            management/
                            __init__.py
                            commands/
                            __init__.py
                            mqtt_ingest.py
                            mqtt_publish_test.py
                            mqtt_healthcheck.py

                            services/
                            __init__.py
                            mqtt_client.py
                            topic_parser.py
                            payload_validator.py
                            idempotency.py
                            redis_streams.py
                            metrics.py
                            command_router.py

                            models.py
                            admin.py
                            tasks.py
                            apps.py
                            urls.py
                            views.py

Management command pattern

from django.core.management.base import BaseCommand

                            from mqtt_app.services.mqtt_client import run_mqtt_ingestion


                            class Command(BaseCommand):
                            help = "Run MQTT ingestion service."

                            def add_arguments(self, parser):
                            parser.add_argument("--client-id", default="backend-ingestion-01")
                            parser.add_argument("--log-level", default="INFO")

                            def handle(self, *args, **options):
                            client_id = options["client_id"]
                            log_level = options["log_level"]

                            self.stdout.write(f"starting mqtt ingestion client_id={client_id}")
                            run_mqtt_ingestion(client_id=client_id, log_level=log_level)

Service responsibilities

Runtime options

Deployment options:
                            - systemd service running manage.py mqtt_ingest
                            - Docker container running mqtt_ingest
                            - Kubernetes Deployment with one or more subscribers
                            - Celery worker consuming Redis/Kafka buffer
                            - separate command service for outbound MQTT commands

                            Avoid:
                            - running MQTT client inside a web request
                            - starting MQTT loop from AppConfig.ready()
                            - hidden background threads inside Django web workers

Validation must combine topic validation and payload validation. A payload claiming device_id=gw-999 must not be accepted if it was published under gw-123. The backend should reject mismatches and store them as security or quality failures.

Topic parser example

from dataclasses import dataclass


                            @dataclass(frozen=True)
                            class ParsedTopic:
                            env: str
                            tenant: str
                            site: str
                            area: str
                            asset: str
                            stream: str
                            signal: str


                            def parse_topic(topic: str) -> ParsedTopic:
                            parts = topic.split("/")

                            if len(parts) != 7:
                            raise ValueError(f"invalid topic shape: {topic}")

                            env, tenant, site, area, asset, stream, signal = parts

                            allowed_streams = {"telemetry", "status", "event", "command", "ack", "result"}

                            if stream not in allowed_streams:
                            raise ValueError(f"invalid stream: {stream}")

                            return ParsedTopic(
                            env=env,
                            tenant=tenant,
                            site=site,
                            area=area,
                            asset=asset,
                            stream=stream,
                            signal=signal,
                            )

Payload validator example

import json
                            from datetime import datetime, timezone, timedelta
                            from typing import Any, Dict


                            ALLOWED_METRICS = {"temperature", "pressure", "vibration", "rpm"}


                            def parse_json_payload(payload_raw: str) -> Dict[str, Any]:
                            try:
                            payload = json.loads(payload_raw)
                            except json.JSONDecodeError as exc:
                            raise ValueError(f"invalid json: {exc}") from exc

                            if not isinstance(payload, dict):
                            raise ValueError("payload must be a JSON object")

                            return payload


                            def validate_telemetry_payload(parsed_topic, payload: Dict[str, Any]) -> Dict[str, Any]:
                            if payload.get("schema_version") != "1.0":
                            raise ValueError("unsupported schema_version")

                            if payload.get("device_id") != parsed_topic.asset:
                            raise ValueError("device_id does not match topic asset")

                            metric = payload.get("metric")
                            if metric not in ALLOWED_METRICS:
                            raise ValueError(f"metric is not allowed: {metric}")

                            if metric != parsed_topic.signal:
                            raise ValueError("metric does not match topic signal")

                            value = payload.get("value")
                            if not isinstance(value, (int, float)):
                            raise ValueError("value must be numeric")

                            ts_raw = payload.get("ts")
                            if not isinstance(ts_raw, str):
                            raise ValueError("ts is required")

                            ts = datetime.fromisoformat(ts_raw.replace("Z", "+00:00"))
                            now = datetime.now(timezone.utc)

                            if ts > now + timedelta(minutes=5):
                            raise ValueError("timestamp is too far in the future")

                            if ts < now - timedelta(days=30):
                            raise ValueError("timestamp is too old")

                            return payload

Failure categories

Validation failure categories:
                            - invalid_topic
                            - invalid_json
                            - unsupported_schema
                            - identity_mismatch
                            - unknown_metric
                            - invalid_value_type
                            - stale_timestamp
                            - future_timestamp
                            - oversized_payload
                            - security_violation

Celery is useful when MQTT ingestion must trigger asynchronous processing, database writes, alerting, enrichment or WebSocket updates. The MQTT callback can enqueue a task, and workers process messages independently.

tasks.py example

import hashlib
                            import json
                            import logging

                            from celery import shared_task
                            from django.db import transaction

                            from mqtt_app.services.topic_parser import parse_topic
                            from mqtt_app.services.payload_validator import (
                            parse_json_payload,
                            validate_telemetry_payload,
                            )

                            LOG = logging.getLogger("mqtt_tasks")


                            def payload_hash(payload_raw: str) -> str:
                            return hashlib.sha256(payload_raw.encode("utf-8")).hexdigest()


                            @shared_task(
                            bind=True,
                            autoretry_for=(Exception,),
                            retry_backoff=True,
                            retry_kwargs={"max_retries": 5},
                            )
                            def process_mqtt_record(self, record: dict) -> str:
                            topic = record["topic"]
                            payload_raw = record["payload_raw"]

                            try:
                            parsed_topic = parse_topic(topic)
                            payload = parse_json_payload(payload_raw)

                            if parsed_topic.stream == "telemetry":
                            payload = validate_telemetry_payload(parsed_topic, payload)
                            return process_telemetry(parsed_topic, payload, payload_raw)

                            if parsed_topic.stream == "status":
                            return process_status(parsed_topic, payload, payload_raw)

                            if parsed_topic.stream == "event":
                            return process_event(parsed_topic, payload, payload_raw)

                            return "ignored"

                            except ValueError as exc:
                            store_rejected_message(record, str(exc))
                            return "rejected"


                            def process_telemetry(parsed_topic, payload: dict, payload_raw: str) -> str:
                            key = build_idempotency_key(parsed_topic, payload)

                            with transaction.atomic():
                            inbox_created = create_inbox_if_new(
                            idempotency_key=key,
                            topic="/".join([
                            parsed_topic.env,
                            parsed_topic.tenant,
                            parsed_topic.site,
                            parsed_topic.area,
                            parsed_topic.asset,
                            parsed_topic.stream,
                            parsed_topic.signal,
                            ]),
                            payload_hash_value=payload_hash(payload_raw),
                            )

                            if not inbox_created:
                            return "duplicate"

                            write_telemetry(parsed_topic, payload)
                            update_current_state(parsed_topic, payload)

                            return "processed"

Celery topology

MQTT Subscriber
                            │
                            │ delay()
                            ▼
                            Celery Broker
                            │
                            ├── queue: mqtt.telemetry
                            ├── queue: mqtt.status
                            ├── queue: mqtt.event
                            └── queue: mqtt.command_result
                            │
                            ▼
                            Celery Workers
                            │
                            ├── validation
                            ├── idempotency
                            ├── database writes
                            ├── alerting
                            └── websocket fanout

Queue separation

Redis Streams can act as a fast buffer between the MQTT subscriber and the processing workers. This gives you consumer groups, pending entries, replay, lag visibility and better backpressure control than a direct database write inside the callback.

Producer example

import redis

                            redis_client = redis.Redis(
                            host="localhost",
                            port=6379,
                            db=0,
                            decode_responses=True,
                            )

                            STREAM_NAME = "mqtt:ingestion"


                            def enqueue_record_to_stream(record: dict) -> str:
                            message_id = redis_client.xadd(
                            STREAM_NAME,
                            fields={
                            "topic": record["topic"],
                            "qos": str(record["qos"]),
                            "retain": "1" if record["retain"] else "0",
                            "payload_raw": record["payload_raw"],
                            "received_at_epoch": str(record["received_at_epoch"]),
                            },
                            maxlen=100000,
                            approximate=True,
                            )

                            return message_id

Consumer group example

STREAM_NAME = "mqtt:ingestion"
                            GROUP_NAME = "mqtt-workers"
                            CONSUMER_NAME = "worker-01"


                            def ensure_group():
                            try:
                            redis_client.xgroup_create(
                            STREAM_NAME,
                            GROUP_NAME,
                            id="0",
                            mkstream=True,
                            )
                            except redis.exceptions.ResponseError as exc:
                            if "BUSYGROUP" not in str(exc):
                            raise


                            def consume_forever():
                            ensure_group()

                            while True:
                            response = redis_client.xreadgroup(
                            groupname=GROUP_NAME,
                            consumername=CONSUMER_NAME,
                            streams={STREAM_NAME: ">"},
                            count=50,
                            block=5000,
                            )

                            for stream_name, messages in response:
                            for message_id, fields in messages:
                            try:
                            process_record(fields)
                            redis_client.xack(STREAM_NAME, GROUP_NAME, message_id)
                            except Exception:
                            LOG.exception("stream processing failed id=%s", message_id)

Redis Streams architecture

MQTT callback
                            │
                            │ XADD mqtt:ingestion
                            ▼
                            Redis Stream
                            │
                            ├── consumer group: mqtt-workers
                            │       ├── worker-01
                            │       ├── worker-02
                            │       └── worker-03
                            │
                            ├── pending entries list
                            ├── lag metrics
                            └── replay support

What to monitor

MQTT QoS 1 may deliver duplicates. Worker retries may also create duplicates. Idempotency ensures the same message can be received more than once without causing duplicate business records or unsafe repeated actions.

Django models example

from django.db import models


                            class MQTTInboxMessage(models.Model):
                            idempotency_key = models.CharField(max_length=128)
                            topic_hash = models.CharField(max_length=64)
                            payload_hash = models.CharField(max_length=64)

                            client_id = models.CharField(max_length=128, blank=True, default="")
                            topic_preview = models.CharField(max_length=512, blank=True, default="")

                            status = models.CharField(max_length=32, default="pending")
                            duplicate_count = models.IntegerField(default=0)

                            first_seen_at = models.DateTimeField(auto_now_add=True)
                            last_seen_at = models.DateTimeField(auto_now=True)

                            class Meta:
                            constraints = [
                            models.UniqueConstraint(
                            fields=["idempotency_key"],
                            name="uniq_mqtt_inbox_key",
                            )
                            ]


                            class MQTTRejectedMessage(models.Model):
                            topic_preview = models.CharField(max_length=512, blank=True, default="")
                            payload_hash = models.CharField(max_length=64, blank=True, default="")
                            error_code = models.CharField(max_length=64)
                            error_message = models.TextField(blank=True, default="")
                            count = models.IntegerField(default=1)
                            first_seen_at = models.DateTimeField(auto_now_add=True)
                            last_seen_at = models.DateTimeField(auto_now=True)

Idempotency key builder

import hashlib


                            def sha256_text(value: str) -> str:
                            return hashlib.sha256(value.encode("utf-8")).hexdigest()


                            def build_idempotency_key(parsed_topic, payload: dict) -> str:
                            message_id = payload.get("message_id")
                            if message_id:
                            return sha256_text(f"message_id:{message_id}")

                            sequence = payload.get("sequence")
                            if sequence is not None:
                            raw = (
                            f"device_sequence:"
                            f"{parsed_topic.tenant}:"
                            f"{parsed_topic.site}:"
                            f"{parsed_topic.asset}:"
                            f"{sequence}"
                            )
                            return sha256_text(raw)

                            ts = payload.get("ts", "")
                            metric = payload.get("metric", parsed_topic.signal)

                            raw = (
                            f"fallback:"
                            f"{parsed_topic.tenant}:"
                            f"{parsed_topic.site}:"
                            f"{parsed_topic.asset}:"
                            f"{metric}:"
                            f"{ts}"
                            )
                            return sha256_text(raw)

Processing decision

Incoming message
                            │
                            ▼
                            Build idempotency key
                            │
                            ▼
                            Try insert MQTTInboxMessage
                            │
                            ├── insert succeeds
                            │       └── process normally
                            │
                            └── unique conflict
                            ├── increment duplicate_count
                            └── skip business write

MQTT messages should usually be split into several storage targets: hot state for dashboards, historical measurements for analytics, event tables for audit and rejected messages for operational troubleshooting.

Telemetry model example

class MQTTTelemetryPoint(models.Model):
                            tenant = models.CharField(max_length=64)
                            site = models.CharField(max_length=64)
                            area = models.CharField(max_length=64)
                            asset = models.CharField(max_length=128)
                            metric = models.CharField(max_length=64)

                            value_float = models.FloatField(null=True, blank=True)
                            unit = models.CharField(max_length=32, blank=True, default="")

                            event_ts = models.DateTimeField()
                            ingestion_ts = models.DateTimeField(auto_now_add=True)

                            message_key = models.CharField(max_length=128)
                            quality = models.CharField(max_length=32, blank=True, default="good")

Storage architecture

Processing worker
                            │
                            ├── Redis hot state
                            │       └── latest value per asset/metric
                            │
                            ├── Time-series table
                            │       └── historical telemetry
                            │
                            ├── Event table
                            │       └── alarms, errors, lifecycle events
                            │
                            ├── Command journal
                            │       └── command, ack, result, timeout
                            │
                            ├── Rejected messages
                            │       └── validation and security failures
                            │
                            └── Object archive
                            └── optional raw payload retention

Hot state Redis key example

Key:
                            mqtt:state:prod:acme:paris:line-02:gw-123:temperature

                            Value:
                            {
                            "value": 21.8,
                            "unit": "C",
                            "event_ts": "2026-04-25T10:00:00Z",
                            "ingestion_ts": "2026-04-25T10:00:02Z",
                            "quality": "good"
                            }

Retention strategy

Deployment checklist

[ ] MQTT daemon runs outside web workers
                            [ ] Stable backend client_id
                            [ ] TLS enabled
                            [ ] Credentials loaded from environment or secret manager
                            [ ] Topics configured by settings
                            [ ] Reconnect backoff enabled
                            [ ] Callback does not block
                            [ ] Payload size limit enforced
                            [ ] Validation errors stored
                            [ ] Idempotency implemented
                            [ ] Redis/Celery/Kafka buffer monitored
                            [ ] Worker lag monitored
                            [ ] Database write latency monitored
                            [ ] Duplicate count visible
                            [ ] Rejected messages visible in admin
                            [ ] Dead-letter or retry policy defined
                            [ ] Graceful shutdown implemented
                            [ ] systemd/Docker/Kubernetes restart policy set
                            [ ] Healthcheck command available
                            [ ] Logs include topic, client_id, message_id and error code

systemd service example

[Unit]
                            Description=MQTT ingestion service
                            After=network-online.target
                            Wants=network-online.target

                            [Service]
                            User=www-data
                            WorkingDirectory=/srv/ideo-lab
                            Environment=DJANGO_SETTINGS_MODULE=project.settings
                            ExecStart=/srv/venv/bin/python manage.py mqtt_ingest --client-id backend-ingestion-01
                            Restart=always
                            RestartSec=5
                            TimeoutStopSec=30

                            [Install]
                            WantedBy=multi-user.target

Incident diagnosis table

Healthcheck command idea

Healthcheck should verify:
                            - broker reachable
                            - TLS handshake works
                            - backend credentials accepted
                            - test subscription succeeds
                            - Redis or Celery broker reachable
                            - database reachable
                            - last processed message age below threshold
                            - rejected message spike below threshold
                            - worker lag below threshold

MQTT is widely used in home automation and smart building scenarios because devices are small, local networks may be unstable, messages are lightweight, and dashboards need immediate state updates.

Typical architecture

Sensors / Switches / Thermostats
                            │
                            │ Wi-Fi / Zigbee gateway / Ethernet
                            ▼
                            Local MQTT Broker
                            │
                            ├── Home Assistant
                            ├── Django dashboard
                            ├── Rules engine
                            ├── Redis hot state
                            └── Time-series DB

Topic examples

home/floor-1/room-12/temperature
                            home/floor-1/room-12/humidity
                            home/floor-1/room-12/presence
                            home/floor-1/room-12/status/heater
                            home/floor-1/room-12/command/heater
                            home/floor-1/room-12/ack/heater

Heating control flow

Temperature sensor
                            │
                            │ PUBLISH home/floor-1/room-12/temperature
                            ▼
                            MQTT Broker
                            │
                            ├── Dashboard updates value
                            ├── Time-series stores history
                            └── Rules engine checks threshold
                            │
                            ▼
                            Command service publishes:
                            home/floor-1/room-12/command/heater
                            │
                            ▼
                            Heater device receives command
                            │
                            ├── validates command_id
                            ├── applies setpoint
                            └── publishes ack/heater

Payload examples

Temperature:
                            {
                            "schema_version": "1.0",
                            "message_id": "msg-001",
                            "value": 21.7,
                            "unit": "C",
                            "ts": "2026-04-25T10:00:00Z"
                            }

                            Heater command:
                            {
                            "schema_version": "1.0",
                            "command_id": "cmd-001",
                            "action": "set_mode",
                            "mode": "eco",
                            "expires_at": "2026-04-25T10:01:00Z"
                            }

                            Ack:
                            {
                            "command_id": "cmd-001",
                            "status": "accepted",
                            "ts": "2026-04-25T10:00:05Z"
                            }

Industrial MQTT connects machines, PLCs, gateways, historians, dashboards and backend systems. The gateway often translates Modbus, OPC UA, serial protocols or vendor-specific APIs into MQTT topics.

Industrial architecture

PLC / Sensors / Machines
                            │
                            │ OPC UA / Modbus / Serial / Vendor API
                            ▼
                            Edge Gateway
                            │
                            ├── normalize values
                            ├── map asset model
                            ├── buffer during network outage
                            └── publish MQTT
                            │
                            ▼
                            MQTT Broker
                            │
                            ├── Historian
                            ├── SCADA bridge
                            ├── Alerting
                            ├── Maintenance analytics
                            └── Enterprise backend

Topic examples

factory/a1/line-7/press-2/telemetry/pressure
                            factory/a1/line-7/press-2/telemetry/temperature
                            factory/a1/line-7/press-2/telemetry/vibration
                            factory/a1/line-7/press-2/event/alarm
                            factory/a1/line-7/press-2/status/availability
                            factory/a1/line-7/press-2/command/reset
                            factory/a1/line-7/press-2/ack/reset

Industrial semantic layer

Raw machine signal:
                            PLC register 40021 = 723

                            Gateway normalization:
                            {
                            "asset": "press-2",
                            "metric": "pressure",
                            "value": 72.3,
                            "unit": "bar",
                            "quality": "good",
                            "ts": "2026-04-25T10:00:00Z"
                            }

                            MQTT topic:
                            factory/a1/line-7/press-2/telemetry/pressure

Sparkplug-style thinking

Industrial MQTT often needs more than basic topics:

                            - asset model
                            - birth certificate
                            - death certificate
                            - typed metrics
                            - device state
                            - metric quality
                            - node/device hierarchy
                            - consistent payload structure
                            - lifecycle events

                            This is why Sparkplug or OPC UA bridge patterns are common in serious IIoT.

Risk points

Energy systems use MQTT for real-time production, consumption, battery, inverter and load control data. MQTT fits well because measurements are frequent, distributed and often produced by gateways or controllers.

Energy site architecture

Meters / Inverters / Batteries
                            │
                            ▼
                            Energy Gateway
                            │
                            ├── normalize kW, kWh, voltage, current
                            ├── compute local aggregates
                            ├── buffer if WAN down
                            └── publish MQTT
                            │
                            ▼
                            MQTT Broker
                            │
                            ├── live dashboard
                            ├── billing/audit pipeline
                            ├── optimization engine
                            └── load control command service

Energy control loop

Meter publishes consumption
                            │
                            ▼
                            Broker routes telemetry
                            │
                            ├── dashboard updates
                            ├── time-series stores kWh
                            └── optimization engine evaluates threshold
                            │
                            ▼
                            Command service publishes:
                            energy/site-01/controller/command/load-shed
                            │
                            ▼
                            Controller acknowledges:
                            energy/site-01/controller/ack/load-shed
                            │
                            ▼
                            Controller result:
                            energy/site-01/controller/result/load-shed

Payload examples

Power telemetry:
                            {
                            "schema_version": "1.0",
                            "message_id": "msg-pwr-001",
                            "metric": "power",
                            "value": 4.82,
                            "unit": "kW",
                            "ts": "2026-04-25T10:00:00Z"
                            }

                            Load shed command:
                            {
                            "schema_version": "1.0",
                            "command_id": "cmd-ls-001",
                            "action": "load_shed",
                            "target_kw": 2.5,
                            "duration_seconds": 300,
                            "expires_at": "2026-04-25T10:01:00Z"
                            }

Vehicles and fleets are a natural MQTT use case: mobile networks are unstable, data is frequent, payloads are small, and events must be routed to dashboards, alerts, maintenance and analytics.

Topic examples

vehicle/fleet-a/truck-984/telemetry/gps
                            vehicle/fleet-a/truck-984/telemetry/speed
                            vehicle/fleet-a/truck-984/telemetry/battery
                            vehicle/fleet-a/truck-984/telemetry/fuel
                            vehicle/fleet-a/truck-984/event/shock
                            vehicle/fleet-a/truck-984/event/door-open
                            vehicle/fleet-a/truck-984/status/connectivity
                            vehicle/fleet-a/truck-984/command/diagnostic

Fleet architecture

Vehicle Gateway
                            │
                            ├── GPS
                            ├── CAN bus
                            ├── battery / fuel
                            ├── door sensor
                            ├── shock sensor
                            └── cellular modem
                            │
                            │ MQTT/TLS
                            ▼
                            Cloud MQTT Broker
                            │
                            ├── map dashboard
                            ├── alerting
                            ├── maintenance analytics
                            ├── route optimization
                            └── historical storage

Offline buffering pattern

Network available:
                            vehicle publishes live MQTT

                            Network unavailable:
                            gateway stores locally:
                            - sequence_number
                            - event_ts
                            - topic
                            - payload
                            - priority

                            Network restored:
                            gateway replays buffered messages
                            with original event_ts and sequence_number

                            Backend:
                            deduplicates using vehicle_id + sequence_number

GPS payload

{
                            "schema_version": "1.0",
                            "message_id": "gps-00018422",
                            "vehicle_id": "truck-984",
                            "sequence": 18422,
                            "lat": 48.8566,
                            "lon": 2.3522,
                            "speed_kmh": 72.4,
                            "heading": 184,
                            "ts": "2026-04-25T10:00:00Z"
                            }

Logistics systems use MQTT to monitor mobile assets, containers, warehouses, cold rooms, pallets, doors, temperature, humidity and exceptions. The key value is exception-driven monitoring: alert only when something deviates from the expected path or conditions.

Topic examples

logistics/warehouse-01/cold-room-03/telemetry/temperature
                            logistics/warehouse-01/cold-room-03/telemetry/humidity
                            logistics/container/cntr-991/telemetry/location
                            logistics/container/cntr-991/event/door-open
                            logistics/container/cntr-991/event/shock
                            logistics/pallet/pallet-883/status/availability

Cold chain flow

Container sensor
                            │
                            │ PUBLISH temperature
                            ▼
                            MQTT Broker
                            │
                            ├── store time-series
                            ├── update latest state
                            └── threshold rule
                            │
                            ├── normal -> no alert
                            └── out of range
                            │
                            ▼
                            event/cold-chain-violation
                            │
                            ├── notify operator
                            ├── create incident
                            └── store compliance evidence

Cold chain payload

{
                            "schema_version": "1.0",
                            "message_id": "temp-991-0001",
                            "asset_id": "cntr-991",
                            "metric": "temperature",
                            "value": 8.7,
                            "unit": "C",
                            "allowed_min": 2.0,
                            "allowed_max": 8.0,
                            "ts": "2026-04-25T10:00:00Z"
                            }

Compliance event payload

{
                            "schema_version": "1.0",
                            "event_id": "evt-cold-001",
                            "asset_id": "cntr-991",
                            "severity": "high",
                            "code": "TEMPERATURE_OUT_OF_RANGE",
                            "value": 8.7,
                            "unit": "C",
                            "ts": "2026-04-25T10:00:00Z"
                            }

MQTT can be used in healthcare-adjacent environments for non-critical telemetry: room sensors, bed presence, equipment status, environmental monitoring, facility alerts and device availability. Medical use cases require stronger compliance, validation, privacy and safety controls.

Architecture care facility

Room Sensors / Equipment
                            │
                            ▼
                            Local MQTT Broker
                            │
                            ├── nurse dashboard
                            ├── facility alerting
                            ├── equipment state monitor
                            └── audit/event storage
                            │
                            ▼
                            Optional secure cloud sync

Safety and privacy considerations

Alert event payload

{
                            "schema_version": "1.0",
                            "event_id": "evt-call-001",
                            "room_id": "room-204",
                            "source": "call-button",
                            "severity": "high",
                            "code": "ASSISTANCE_REQUESTED",
                            "ts": "2026-04-25T10:00:00Z"
                            }

MQTT is excellent for edge observability: gateways, small servers, remote agents, industrial PCs and distributed sites can publish health, metrics, logs, status and connectivity events to a central backend.

Topic examples

edge/site-01/gateway-01/status/availability
                            edge/site-01/gateway-01/telemetry/cpu
                            edge/site-01/gateway-01/telemetry/memory
                            edge/site-01/gateway-01/telemetry/disk
                            edge/site-01/gateway-01/event/service-down
                            edge/site-01/gateway-01/event/network-loss
                            edge/site-01/gateway-01/command/restart-service

Edge monitoring architecture

Edge Agent
                            │
                            ├── reads local metrics
                            ├── checks services
                            ├── monitors disk
                            ├── publishes heartbeat
                            ├── publishes events
                            └── receives safe commands
                            │
                            ▼
                            MQTT Broker
                            │
                            ├── operations dashboard
                            ├── alert manager
                            ├── time-series storage
                            ├── command service
                            └── incident history

Agent status payload

{
                            "schema_version": "1.0",
                            "agent_id": "gateway-01",
                            "state": "online",
                            "version": "2.4.1",
                            "hostname": "edge-site-01",
                            "uptime_seconds": 88422,
                            "ts": "2026-04-25T10:00:00Z"
                            }

Service-down event

{
                            "schema_version": "1.0",
                            "event_id": "evt-svc-001",
                            "service": "local-ingestion",
                            "state": "down",
                            "severity": "critical",
                            "ts": "2026-04-25T10:00:00Z"
                            }

Pattern 1: MQTT to time-series storage

Devices
                            │
                            ▼
                            MQTT Broker
                            │
                            ▼
                            Ingestion worker
                            │
                            ├── validate
                            ├── deduplicate
                            ├── normalize units
                            └── batch write
                            │
                            ▼
                            TimescaleDB / InfluxDB / ClickHouse

Pattern 2: MQTT to dashboard

Devices
                            │
                            ▼
                            MQTT Broker
                            │
                            ├── Backend subscriber
                            │       └── WebSocket push
                            │
                            └── Retained status
                            └── dashboard initial state

Pattern 3: MQTT to alerting

Telemetry / Event
                            │
                            ▼
                            MQTT Broker
                            │
                            ▼
                            Rules engine
                            │
                            ├── threshold check
                            ├── anomaly check
                            ├── state transition
                            └── alert creation

Pattern 4: Edge broker to cloud broker

Local site
                            │
                            ├── devices
                            ├── local MQTT broker
                            ├── local dashboard
                            └── local rules
                            │
                            │ bridge MQTT/TLS
                            ▼
                            Cloud broker
                            │
                            ├── central dashboard
                            ├── data lake
                            ├── analytics
                            └── command service

Pattern 5: MQTT to Kafka

MQTT Broker
                            │
                            ▼
                            Connector / Worker
                            │
                            ▼
                            Kafka topic
                            │
                            ├── stream processing
                            ├── data lake
                            ├── ML pipeline
                            └── audit replay

Pattern 6: Command and result

Backend command service
                            │
                            ▼
                            MQTT command topic
                            │
                            ▼
                            Device validates command
                            │
                            ├── publish ack
                            └── publish result

MQTT is excellent for

• IoT and distributed devices.
• Telemetry and edge observability.
• Real-time dashboards.
• Machine status and events.
• Fleet and mobile connectivity.
• Lightweight command channels with ack/result.

MQTT is not ideal for

Final design checklist

[ ] Use case really benefits from Pub/Sub
                            [ ] Topic namespace is documented
                            [ ] QoS is chosen per flow
                            [ ] Status uses retained where useful
                            [ ] Offline uses Last Will
                            [ ] Commands have command_id
                            [ ] Commands have ack and result
                            [ ] Backend validates payloads
                            [ ] Backend is idempotent
                            [ ] Storage is designed for history
                            [ ] Broker is monitored
                            [ ] Security ACLs are tested
                            [ ] Failure cases are documented

Sparkplug introduces a vocabulary that is essential for industrial MQTT. The main objects are the application namespace, group, edge node, device, metric and lifecycle message type.

Industrial hierarchy

Sparkplug hierarchy:

                            Application Namespace
                            └── Group ID
                            └── Edge Node ID
                            ├── Node metrics
                            │   ├── CPU
                            │   ├── uptime
                            │   └── network status
                            │
                            └── Device ID
                            ├── temperature
                            ├── pressure
                            ├── vibration
                            ├── running state
                            └── alarm state

Message type families

Birth and death messages are the heart of Sparkplug. A birth message announces that a node or device is online and declares its metrics. A death message announces that it is offline. This allows SCADA systems, historians and dashboards to maintain a reliable live model of the industrial system.

Lifecycle diagram

Edge node starts
                            │
                            ├── CONNECT to MQTT broker
                            │       └── registers NDEATH as Last Will
                            │
                            ├── publish NBIRTH
                            │       └── node metrics are declared
                            │
                            ├── publish DBIRTH for each device
                            │       └── device metrics are declared
                            │
                            ├── publish NDATA / DDATA
                            │       └── regular metric updates
                            │
                            ├── normal shutdown
                            │       └── graceful disconnect
                            │
                            └── unexpected failure
                            └── broker publishes NDEATH

Why birth matters

Without birth:
                            - consumer sees values but not full metric catalog
                            - SCADA may not know what tags exist
                            - historian mapping must be hardcoded
                            - restart creates ambiguity
                            - offline/online is custom

                            With birth:
                            - node announces complete metric set
                            - consumers can rebuild state
                            - SCADA can discover tags
                            - backend can detect restarts
                            - sequence and lifecycle are explicit

Sparkplug defines a topic namespace that carries the namespace version, group, message type, edge node and optionally device ID. This makes routing and interpretation standardized across industrial consumers.

Generic structure

spBv1.0/{group_id}/{message_type}/{edge_node_id}

                            spBv1.0/{group_id}/{message_type}/{edge_node_id}/{device_id}

Examples

Node birth:
                            spBv1.0/factory-a/NBIRTH/edge-01

                            Node data:
                            spBv1.0/factory-a/NDATA/edge-01

                            Node death:
                            spBv1.0/factory-a/NDEATH/edge-01

                            Device birth:
                            spBv1.0/factory-a/DBIRTH/edge-01/press-02

                            Device data:
                            spBv1.0/factory-a/DDATA/edge-01/press-02

                            Device death:
                            spBv1.0/factory-a/DDEATH/edge-01/press-02

                            Device command:
                            spBv1.0/factory-a/DCMD/edge-01/press-02

Topic fields

Subscription patterns

All Sparkplug messages:
                            spBv1.0/#

                            All factory-a messages:
                            spBv1.0/factory-a/#

                            All device data:
                            spBv1.0/+/DDATA/+/+

                            All birth messages:
                            spBv1.0/+/+BIRTH/#

                            One edge node:
                            spBv1.0/factory-a/+/edge-01/#

Sparkplug payloads are designed to represent industrial metrics with structure. A metric is not only a raw value: it can include a name, alias, datatype, timestamp, value, quality and metadata. This is essential for industrial interoperability.

Birth payload idea

DBIRTH for press-02 declares:

                            metrics:
                            - pressure
                            datatype: Float
                            unit: bar
                            alias: 1

                            - temperature
                            datatype: Float
                            unit: C
                            alias: 2

                            - running
                            datatype: Boolean
                            alias: 3

                            - alarm_active
                            datatype: Boolean
                            alias: 4

Data payload idea

DDATA for press-02 sends updates:

                            metrics:
                            - alias: 1
                            value: 72.4
                            timestamp: 2026-04-25T10:00:00Z

                            - alias: 2
                            value: 41.8
                            timestamp: 2026-04-25T10:00:00Z

                            - alias: 3
                            value: true
                            timestamp: 2026-04-25T10:00:00Z

                            Meaning is known because DBIRTH mapped:
                            alias 1 -> pressure
                            alias 2 -> temperature
                            alias 3 -> running

Metric design rules

A Sparkplug architecture usually starts at the edge. The edge node talks to PLCs, sensors and controllers, normalizes data into Sparkplug metrics, publishes lifecycle messages and forwards data to MQTT consumers such as SCADA, historians and dashboards.

Reference architecture

PLC / Sensors / Controllers
                            │
                            │ OPC UA / Modbus / vendor protocol
                            ▼
                            Sparkplug Edge Node / Gateway
                            │
                            ├── reads industrial tags
                            ├── maps tags to metrics
                            ├── publishes NBIRTH
                            ├── publishes DBIRTH
                            ├── publishes NDATA / DDATA
                            ├── registers NDEATH Last Will
                            └── handles NCMD / DCMD
                            │
                            ▼
                            MQTT Broker
                            │
                            ├── SCADA
                            ├── Historian
                            ├── Operations dashboard
                            ├── Alerting engine
                            ├── MES / ERP bridge
                            └── Data lake connector

Consumer roles

Data path

Raw PLC tag
                            │
                            ▼
                            Gateway driver
                            │
                            ▼
                            Metric mapping
                            │
                            ▼
                            Sparkplug payload
                            │
                            ▼
                            MQTT broker
                            │
                            ├── SCADA live display
                            ├── historian time-series
                            └── backend event processing

Sparkplug is valuable because it makes MQTT easier for industrial systems to consume. A SCADA system can discover nodes and devices through birth messages. A historian can map metric names, types and timestamps. Operations dashboards can distinguish offline state from missing data.

SCADA discovery flow

SCADA subscribes:
                            spBv1.0/+/+/+/#

                            Edge node starts:
                            publishes NBIRTH

                            Devices appear:
                            publishes DBIRTH for each device

                            SCADA builds:
                            - node list
                            - device list
                            - metric tags
                            - current state
                            - command capabilities

                            Regular updates:
                            NDATA / DDATA update live values

Historian mapping flow

DBIRTH declares metrics:
                            press-02.pressure
                            press-02.temperature
                            press-02.running
                            press-02.alarm_active

                            Historian creates or maps tags:
                            factory-a.edge-01.press-02.pressure
                            factory-a.edge-01.press-02.temperature

                            DDATA arrives:
                            alias 1 = 72.4
                            alias 2 = 41.8

                            Historian writes:
                            tag, value, event_ts, quality, ingestion_ts

Operations dashboard model

In many factories, MQTT does not directly read the machine. The edge gateway reads industrial protocols such as OPC UA, Modbus, serial protocols or vendor APIs, then publishes a normalized Sparkplug model to MQTT.

OPC UA to Sparkplug mapping

OPC UA source:
                            ns=2;s=Press02.Pressure
                            ns=2;s=Press02.Temperature
                            ns=2;s=Press02.Running
                            ns=2;s=Press02.AlarmActive

                            Gateway mapping:
                            Press02.Pressure     -> metric pressure
                            Press02.Temperature  -> metric temperature
                            Press02.Running      -> metric running
                            Press02.AlarmActive  -> metric alarm_active

                            Sparkplug device:
                            group_id     = factory-a
                            edge_node_id = edge-01
                            device_id    = press-02

Bridge architecture

PLC / Machine
                            │
                            ▼
                            OPC UA Server
                            │
                            ▼
                            Edge Gateway
                            │
                            ├── browse OPC UA nodes
                            ├── map nodes to assets
                            ├── read values
                            ├── preserve quality
                            ├── preserve timestamp
                            ├── build Sparkplug metrics
                            └── publish MQTT
                            │
                            ▼
                            MQTT Broker
                            │
                            ├── SCADA
                            ├── Historian
                            └── Backend

Mapping risks

A Django backend consuming Sparkplug should model lifecycle separately from metric history. Birth and death messages update node/device state. Data messages write time-series points. Commands and command results should be audited.

Recommended Django models

SparkplugGroup
                            - group_id
                            - description
                            - tenant
                            - enabled

                            SparkplugEdgeNode
                            - group
                            - edge_node_id
                            - online_state
                            - last_birth_at
                            - last_death_at
                            - last_sequence
                            - last_seen_at

                            SparkplugDevice
                            - edge_node
                            - device_id
                            - online_state
                            - last_birth_at
                            - last_death_at
                            - last_seen_at

                            SparkplugMetricDefinition
                            - device or node
                            - metric_name
                            - alias
                            - datatype
                            - unit
                            - properties
                            - active

                            SparkplugMetricSample
                            - metric_definition
                            - value
                            - quality
                            - event_ts
                            - ingestion_ts

                            SparkplugCommandJournal
                            - command_id
                            - target
                            - command_name
                            - payload
                            - status
                            - requested_by
                            - created_at
                            - ack_at
                            - result_at

Processing pipeline

MQTT message
                            │
                            ▼
                            Parse Sparkplug topic
                            │
                            ▼
                            Identify message type
                            │
                            ├── NBIRTH
                            │       ├── upsert edge node
                            │       ├── mark online
                            │       └── upsert node metric definitions
                            │
                            ├── NDEATH
                            │       ├── mark edge node offline
                            │       └── mark child devices uncertain/offline
                            │
                            ├── DBIRTH
                            │       ├── upsert device
                            │       ├── mark online
                            │       └── upsert device metric definitions
                            │
                            ├── DDEATH
                            │       └── mark device offline
                            │
                            ├── NDATA / DDATA
                            │       ├── resolve metric alias/name
                            │       ├── write samples
                            │       └── update current state
                            │
                            └── NCMD / DCMD
                            └── audit command flow

Admin views

Sparkplug production checklist

[ ] Group IDs are documented
                            [ ] Edge node IDs are stable
                            [ ] Device IDs are stable
                            [ ] NBIRTH is published on startup
                            [ ] NDEATH is configured as Last Will
                            [ ] DBIRTH is published for devices
                            [ ] Metric definitions are stable
                            [ ] Metric units are declared
                            [ ] Metric datatypes are correct
                            [ ] Metric aliases are tracked
                            [ ] Sequence numbers are monitored
                            [ ] NDATA / DDATA are stored correctly
                            [ ] Death messages update online state
                            [ ] Rebirth handling is implemented
                            [ ] Commands use NCMD / DCMD patterns
                            [ ] Command audit is stored
                            [ ] SCADA subscriptions are scoped
                            [ ] Historian mapping is tested
                            [ ] Backend rejects unknown metrics safely
                            [ ] Dashboard distinguishes offline from stale data

Operational alerts

Anti-patterns

Final architecture rule

Useful URLs

Sparkplug specification:
                            https://sparkplug.eclipse.org/specification/

                            Eclipse Sparkplug project:
                            https://sparkplug.eclipse.org/

                            MQTT official:
                            https://mqtt.org/

                            OPC Foundation:
                            https://opcfoundation.org/

                            Eclipse Tahu:
                            https://github.com/eclipse/tahu

Broker dashboard layout

MQTT Broker Dashboard
                            ├── Broker health
                            │   ├── uptime
                            │   ├── node status
                            │   ├── CPU / RAM
                            │   ├── disk
                            │   └── file descriptors
                            │
                            ├── Connections
                            │   ├── connected clients
                            │   ├── connects/sec
                            │   ├── disconnects/sec
                            │   ├── reconnect loops
                            │   └── top clients by traffic
                            │
                            ├── Traffic
                            │   ├── messages in/sec
                            │   ├── messages out/sec
                            │   ├── bytes in/out
                            │   ├── publish rate by topic prefix
                            │   └── fan-out ratio
                            │
                            ├── Reliability
                            │   ├── dropped messages
                            │   ├── rejected messages
                            │   ├── queued messages
                            │   ├── retained messages
                            │   └── inflight messages
                            │
                            └── Security
                            ├── auth failures
                            ├── ACL denied
                            ├── invalid certificate
                            └── suspicious wildcard attempts

Useful derived metrics

Broker metrics show the platform view. Fleet metrics show the device reality: which clients are online, which firmware versions are unstable, which sites are losing connectivity, and which devices are producing abnormal traffic.

Device state model

Device state:
                            - online
                            - offline
                            - stale
                            - degraded
                            - unknown
                            - disabled
                            - revoked

                            Transitions:
                            online -> stale      if no data for threshold
                            stale  -> offline    if Last Will or long timeout
                            offline -> online    on new retained status or birth
                            online -> degraded   if errors or bad quality increase
                            any -> disabled      operator action
                            any -> revoked       security incident

Fleet dashboard

Fleet / Device Dashboard
                            ├── Global
                            │   ├── total devices
                            │   ├── online devices
                            │   ├── stale devices
                            │   ├── offline devices
                            │   └── disabled/revoked devices
                            │
                            ├── By site
                            │   ├── online percentage
                            │   ├── last message age
                            │   ├── reconnect rate
                            │   └── top failing devices
                            │
                            ├── By firmware
                            │   ├── version distribution
                            │   ├── error rate
                            │   ├── reconnect rate
                            │   └── payload validation failures
                            │
                            └── By topic prefix
                            ├── message rate
                            ├── rejected count
                            ├── payload size
                            └── retained count

Staleness thresholds

MQTT broker health does not guarantee backend health. Your backend can be slow while the broker continues receiving messages. You must monitor queue depth, consumer lag, oldest pending message age, database latency and end-to-end freshness.

End-to-end freshness

event_ts:
                            time when device measured the data

                            received_at:
                            time when backend subscriber received MQTT message

                            stored_at:
                            time when message was stored in database

                            dashboard_seen_at:
                            time when UI displayed updated state

                            Freshness metrics:
                            - received_at - event_ts
                            - stored_at - event_ts
                            - dashboard_seen_at - event_ts
                            - stored_at - received_at

Lag diagnosis diagram

Messages are late
                            │
                            ├── Broker messages in normal?
                            │       ├── no  -> publisher/device/network issue
                            │       └── yes
                            │
                            ├── Subscriber receiving?
                            │       ├── no  -> subscription/ACL/session issue
                            │       └── yes
                            │
                            ├── Buffer growing?
                            │       ├── yes -> workers too slow or DB slow
                            │       └── no
                            │
                            ├── DB write latency high?
                            │       ├── yes -> optimize DB/batch/partition
                            │       └── no
                            │
                            └── Dashboard stale?
                            ├── yes -> WebSocket/cache/API issue
                            └── no  -> incident resolved

Backend SLO examples

MQTT alerts must be actionable. Avoid alerts that only say “broker problem”. Good alerts define the symptom, likely impact, owner, severity, first diagnostic steps and definition of done.

Alert principles

Alert design:
                            - warning: metric outside baseline for a short period
                            - critical: sustained issue with user/business impact
                            - every alert has an owner
                            - every alert links to a runbook
                            - every alert has a definition of done
                            - alerts use rate and trend, not only absolute value
                            - alerts are grouped by site, tenant, broker node or firmware

Example alert definitions

Alert: MQTTConnectedClientsDrop
                            Condition:
                            connected_clients{env="prod"} drops by 30% over 10 minutes

                            Impact:
                            Devices may be unable to publish telemetry.

                            First checks:
                            1. Broker health
                            2. TLS certificate expiry
                            3. Load balancer health
                            4. DNS
                            5. Network/firewall changes
                            6. Recent deployment

                            Definition of done:
                            connected_clients returns to expected baseline
                            and no abnormal reconnect rate remains.


                            Alert: MQTTBackendConsumerLag
                            Condition:
                            oldest_pending_message_age_seconds > freshness_slo_seconds

                            Impact:
                            Dashboard and alerts are delayed.

                            First checks:
                            1. Queue depth
                            2. Worker count
                            3. DB latency
                            4. rejected payload spike
                            5. CPU/RAM workers

                            Definition of done:
                            queue depth stable or decreasing
                            oldest pending age below SLO

Alert routing

One dashboard is not enough. MQTT production needs at least four views: broker health, fleet/device health, backend ingestion, and business/application state.

Broker panels

Broker dashboard panels:
                            - connected clients
                            - messages in/sec
                            - messages out/sec
                            - dropped messages
                            - rejected messages
                            - retained messages count
                            - session queue depth
                            - auth failures
                            - ACL denied
                            - CPU / RAM / network
                            - top topic prefixes by message rate

Ingestion dashboard panels

Backend ingestion dashboard:
                            - Redis stream length
                            - Celery queue depth
                            - Kafka consumer lag
                            - oldest pending message age
                            - processing time p50/p95/p99
                            - DB write latency p95/p99
                            - rejected messages by reason
                            - duplicate messages by device
                            - top noisy devices
                            - latest stored timestamp by flow

Business panels

Incident triage diagram

Incident detected
                            │
                            ▼
                            Is broker reachable?
                            ├── no  -> infra incident
                            └── yes
                            │
                            ▼
                            Are clients connected?
                            ├── no  -> TLS/DNS/network/auth
                            └── yes
                            │
                            ▼
                            Are messages entering broker?
                            ├── no  -> publisher/device issue
                            └── yes
                            │
                            ▼
                            Are messages leaving broker?
                            ├── no  -> subscription/ACL/session issue
                            └── yes
                            │
                            ▼
                            Is backend consuming?
                            ├── no  -> subscriber/queue/worker issue
                            └── yes
                            │
                            ▼
                            Is storage/dashboard fresh?
                            ├── no  -> DB/API/cache/WebSocket issue
                            └── yes -> incident may be scoped or resolved

First 5 minutes checklist

[ ] Check broker health
                            [ ] Check connected clients trend
                            [ ] Check messages in/out trend
                            [ ] Check auth and ACL failures
                            [ ] Check dropped/rejected messages
                            [ ] Check backend lag
                            [ ] Check recent deployments
                            [ ] Check TLS certificate expiry
                            [ ] Check firewall/LB/DNS changes
                            [ ] Check top noisy clients and topic prefixes

Runbook: mass device offline

Symptoms:
                            - connected clients drop sharply
                            - dashboards show offline devices
                            - reconnect attempts may spike

                            Immediate checks:
                            1. Broker process/node status
                            2. Load balancer health
                            3. TLS certificate expiry
                            4. DNS resolution
                            5. Firewall/security group changes
                            6. Recent broker config deployment
                            7. Auth backend availability
                            8. Network provider or VPN status

                            Actions:
                            1. Roll back recent broker/TLS/LB config if suspected.
                            2. Restore previous certificate if rotation failed.
                            3. Move traffic to healthy broker node if cluster.
                            4. Temporarily increase logs for connect failures.
                            5. Communicate impacted sites/tenants.
                            6. Confirm clients reconnect and publish data.

                            Definition of done:
                            - connected clients back to baseline
                            - reconnect rate normal
                            - data freshness below SLO
                            - no auth failure spike remains

Runbook: retained store flood

Symptoms:
                            - retained message count grows unexpectedly
                            - memory or storage increases
                            - new dashboards receive old irrelevant values

                            Immediate checks:
                            1. Top retained topic prefixes
                            2. Recent firmware deployment
                            3. Publish retain flag usage
                            4. Telemetry topics accidentally retained
                            5. ACL allowing retained publish too broadly

                            Actions:
                            1. Stop offending publisher or firmware.
                            2. Disable retain on telemetry flow.
                            3. Clear retained messages for bad prefixes carefully.
                            4. Add alert on retained growth.
                            5. Add validation in publisher library.

                            Definition of done:
                            - retained count stable
                            - bad retained topics removed
                            - publisher fixed
                            - alert added

Runbook: backend consumer lag

Symptoms:
                            - queue depth grows
                            - oldest pending age exceeds SLO
                            - dashboard stale
                            - alarms delayed

                            Immediate checks:
                            1. Worker count and health
                            2. Redis/Celery/Kafka lag
                            3. DB write latency
                            4. rejected payload spike
                            5. top topic prefixes by volume
                            6. recent backend deployment
                            7. database locks or slow queries

                            Actions:
                            1. Scale workers if safe.
                            2. Reduce low-priority ingestion.
                            3. Enable batch writes.
                            4. Pause noisy clients if needed.
                            5. Add temporary consumer group.
                            6. Fix DB bottleneck or rollback deployment.
                            7. Prioritize alarm/event queues.

                            Definition of done:
                            - queue depth decreasing
                            - oldest pending age below SLO
                            - dashboard fresh
                            - alarm flow normal

Runbook: ACL denied spike

Symptoms:
                            - ACL denied rate spikes
                            - devices cannot publish or subscribe
                            - new firmware may fail

                            Immediate checks:
                            1. Which client IDs?
                            2. Which topic prefixes?
                            3. Publish or subscribe denied?
                            4. Firmware version?
                            5. Recent ACL changes?
                            6. Suspicious source IPs?

                            Actions:
                            1. If firmware bug: rollback or hotfix topic convention.
                            2. If ACL config bug: rollback ACL.
                            3. If attack: block source and preserve logs.
                            4. If provisioning issue: fix registry/identity mapping.
                            5. Add regression test for denied/allowed cases.

                            Definition of done:
                            - denied rate back to baseline
                            - affected clients confirmed healthy
                            - root cause documented

MQTT capacity depends on connected clients, publish rate, subscriber fan-out, QoS level, retained messages, persistent sessions, payload size and backend throughput. The broker can often handle large message rates, but the backend database may become the real bottleneck.

Sizing formula thinking

Incoming messages/sec:
                            devices * messages_per_device_per_sec

                            Outgoing messages/sec:
                            incoming_messages * average_subscriber_fanout

                            Incoming bandwidth:
                            incoming_messages * average_payload_size

                            Backend writes/sec:
                            messages_to_store * write_amplification

                            Example:
                            10,000 devices
                            1 message / 10 sec each
                            = 1,000 messages/sec incoming

                            If fanout = 3:
                            = 3,000 messages/sec outgoing

                            If payload = 500 bytes:
                            = 500 KB/sec incoming before overhead

Scaling patterns

Small scale:
                            Devices -> Mosquitto -> Python worker -> PostgreSQL

                            Medium scale:
                            Devices -> MQTT broker -> Redis Stream -> worker pool -> TimescaleDB

                            Large scale:
                            Devices -> MQTT cluster -> Kafka bridge -> stream processors -> data lake

                            Industrial edge:
                            Machines -> local broker -> local dashboard
                            -> bridge -> central broker -> analytics

Load test plan

Load test dimensions:
                            [ ] connected clients
                            [ ] publish rate
                            [ ] payload size
                            [ ] QoS level
                            [ ] retained flag usage
                            [ ] subscriber fan-out
                            [ ] persistent sessions
                            [ ] disconnect/reconnect storm
                            [ ] backend database write rate
                            [ ] command ack latency

                            Measure:
                            [ ] broker CPU/RAM
                            [ ] messages in/out
                            [ ] dropped messages
                            [ ] backend lag
                            [ ] DB latency
                            [ ] dashboard freshness
                            [ ] error rate

Monitoring checklist

[ ] Connected clients monitored
                            [ ] Connect/disconnect rate monitored
                            [ ] Messages in/sec monitored
                            [ ] Messages out/sec monitored
                            [ ] Bytes in/out monitored
                            [ ] Dropped messages monitored
                            [ ] Rejected messages monitored
                            [ ] Auth failures monitored
                            [ ] ACL denied monitored
                            [ ] Session queue depth monitored
                            [ ] Inflight messages monitored
                            [ ] Retained messages monitored
                            [ ] Broker CPU/RAM/disk monitored
                            [ ] File descriptors monitored
                            [ ] Backend queue depth monitored
                            [ ] Backend lag monitored
                            [ ] DB write latency monitored
                            [ ] Rejected payloads visible in admin
                            [ ] Duplicate messages visible in admin
                            [ ] Data freshness SLO monitored
                            [ ] Command ack/result latency monitored

Alert checklist

[ ] Every alert has severity
                            [ ] Every alert has owner
                            [ ] Every alert has runbook
                            [ ] Every alert has impact description
                            [ ] Every alert has definition of done
                            [ ] Alerts use baseline when possible
                            [ ] Alerts are grouped by tenant/site/node
                            [ ] Alerts avoid noisy one-minute spikes
                            [ ] Critical alarms detect business impact
                            [ ] Security alerts preserve forensic context

Operational checklist

[ ] Broker config versioned
                            [ ] ACL config versioned
                            [ ] TLS certificates tracked
                            [ ] Certificate expiry alerts enabled
                            [ ] Rollback procedure documented
                            [ ] Load test performed
                            [ ] Failover test performed
                            [ ] Reconnect storm test performed
                            [ ] Slow subscriber test performed
                            [ ] Retained cleanup procedure documented
                            [ ] Device revocation procedure documented
                            [ ] Command incident procedure documented
                            [ ] Backup and restore tested
                            [ ] Dashboard reviewed with operations
                            [ ] On-call runbooks available
                            [ ] Post-incident review template ready

Definition of a healthy MQTT platform

Minimum run command examples

Smoke subscribe:
                            mosquitto_sub -h broker.example.com -p 8883 \
                            --cafile ca.crt \
                            -u monitor -P password \
                            -t 'prod/acme/+/+/+/status/#' -v

                            Smoke publish:
                            mosquitto_pub -h broker.example.com -p 8883 \
                            --cafile ca.crt \
                            -u test_device -P password \
                            -t 'prod/acme/test/line-01/test-device/status/availability' \
                            -m '{"schema_version":"1.0","state":"online"}' \
                            -q 1 -r

                            Debug local only:
                            mosquitto_sub -h localhost -t '#' -v

Pattern selection

Small project:
                            Devices -> Mosquitto -> Python worker -> PostgreSQL

                            Medium project:
                            Devices -> EMQX/HiveMQ -> Redis Stream -> workers -> TimescaleDB

                            Large analytics:
                            Devices -> MQTT cluster -> Kafka bridge -> stream processors -> data lake

                            Industrial edge:
                            Machines -> local broker -> bridge -> central broker -> historian

                            Cloud managed:
                            Devices -> AWS IoT Core / Azure IoT Hub -> cloud rules -> storage/services

Scaling decision tree

Need local autonomy?
                            ├── yes -> edge broker per site + bridge
                            └── no
                            │
                            ▼
                            Need managed cloud identity?
                            ├── yes -> AWS IoT Core / Azure IoT Hub
                            └── no
                            │
                            ▼
                            Need high connection scale?
                            ├── yes -> EMQX / HiveMQ / VerneMQ cluster
                            └── no
                            │
                            ▼
                            Need simple broker?
                            ├── yes -> Mosquitto
                            └── no -> RabbitMQ plugin if AMQP ecosystem exists

What must scale together

• Broker listener capacity.
• Authentication backend.
• ACL evaluation path.
• Backend subscribers.
• Buffer layer such as Redis or Kafka.
• Database write throughput.
• Dashboard fan-out.
• Monitoring and alerting pipeline.

MQTT clusters are used when one broker instance is not enough for availability, connection count, message rate or operational resilience. The cluster is usually placed behind a TCP/TLS load balancer.

Cluster architecture

Devices
                            │
                            │ MQTT/TLS
                            ▼
                            TCP/TLS Load Balancer
                            │
                            ├── MQTT Broker Node 1
                            ├── MQTT Broker Node 2
                            ├── MQTT Broker Node 3
                            └── MQTT Broker Node N
                            │
                            ├── shared routing
                            ├── session handling
                            ├── retained store
                            ├── auth / ACL
                            └── metrics
                            │
                            ▼
                            Backend subscribers / bridges / rules engine

Load balancer choices

Cluster readiness checklist

[ ] Load balancer health checks
                            [ ] Broker node metrics
                            [ ] Cluster membership metrics
                            [ ] Reconnect behavior tested
                            [ ] Node failure tested
                            [ ] TLS certificate rotation tested
                            [ ] Retained messages behavior tested
                            [ ] Persistent session behavior tested
                            [ ] Shared subscription behavior tested
                            [ ] Auth backend failure tested
                            [ ] ACL changes tested
                            [ ] Backup and restore procedure
                            [ ] Rollback broker config procedure

Shared subscriptions allow multiple backend consumers to share the same subscription group. Instead of every consumer receiving every message, the broker distributes messages across members of the group. This is essential for scaling ingestion workers.

Normal subscription

Subscription:
                            prod/acme/+/+/+/telemetry/#

                            Message:
                            prod/acme/paris/line-02/gw-123/telemetry/temperature

                            Result:
                            worker-1 receives message
                            worker-2 receives message
                            worker-3 receives message

                            Problem:
                            duplicate processing unless consumers are designed for fan-out.

Shared subscription

Subscription:
                            $share/ingestion-workers/prod/acme/+/+/+/telemetry/#

                            Message A -> worker-1
                            Message B -> worker-2
                            Message C -> worker-3
                            Message D -> worker-1

                            Result:
                            load is distributed across the worker group.

Shared subscription architecture

MQTT Broker
                            │
                            └── $share/ingestion-workers/prod/acme/+/+/+/telemetry/#
                            │
                            ├── ingestion-worker-01
                            ├── ingestion-worker-02
                            ├── ingestion-worker-03
                            └── ingestion-worker-04
                            │
                            ▼
                            Redis / Kafka / Database

Design rules

An edge bridge connects a local site broker to a central broker. This is common in factories, buildings, stores, energy sites and remote facilities. The local site can continue operating during WAN outages, then synchronize selected topics when the link returns.

Bridge architecture

Factory local devices
                            │
                            ▼
                            Local MQTT Broker
                            │
                            ├── local dashboard
                            ├── local rules
                            ├── local buffer
                            └── bridge MQTT/TLS/mTLS
                            │
                            ▼
                            Central MQTT Cluster
                            │
                            ├── global dashboard
                            ├── storage
                            ├── alerting
                            ├── analytics
                            └── ERP / CMMS integration

Bridge topic mapping

Local topic:
                            site/paris/line-02/gw-123/telemetry/temperature

                            Central topic:
                            prod/acme/paris/line-02/gw-123/telemetry/temperature

                            Upstream bridge:
                            site/+/+/+/telemetry/#  -> prod/acme/{site}/{area}/{asset}/telemetry/#
                            site/+/+/+/status/#     -> prod/acme/{site}/{area}/{asset}/status/#
                            site/+/+/+/event/#      -> prod/acme/{site}/{area}/{asset}/event/#

                            Downstream bridge:
                            prod/acme/paris/+/+/command/# -> site/paris/{area}/{asset}/command/#

Bridge failure modes

Managed cloud IoT services provide MQTT connectivity plus device identity, policies, routing rules, cloud integrations, monitoring and operational tooling. They reduce broker operations but introduce quotas, cost models, service-specific conventions and cloud lock-in.

AWS IoT Core pattern

Devices
                            │
                            │ MQTT/TLS + device certificate
                            ▼
                            AWS IoT Core
                            │
                            ├── thing registry
                            ├── IoT policies
                            ├── rules engine
                            │       ├── Lambda
                            │       ├── Kinesis
                            │       ├── DynamoDB
                            │       ├── S3
                            │       └── SNS / SQS
                            └── CloudWatch logs and metrics

Azure IoT Hub pattern

Devices
                            │
                            │ MQTT/TLS
                            ▼
                            Azure IoT Hub
                            │
                            ├── device identity
                            ├── device-to-cloud messages
                            ├── cloud-to-device messages
                            ├── Device Twin
                            ├── routing
                            │       ├── Event Hubs
                            │       ├── Storage
                            │       ├── Service Bus
                            │       └── Azure Functions
                            └── Azure Monitor / logs

Managed vs self-hosted

MQTT brokers are excellent for device connectivity, but they are not long-term event logs. For replay, analytics, machine learning, data lakes and high-volume stream processing, MQTT is often connected to Kafka, Redpanda, Pulsar, Kinesis or Event Hubs.

MQTT to streaming architecture

Devices
                            │
                            ▼
                            MQTT Broker
                            │
                            ├── operational subscribers
                            └── streaming connector
                            │
                            ▼
                            Kafka / Redpanda / Pulsar / Kinesis / Event Hubs
                            │
                            ├── stream processing
                            ├── data lake
                            ├── anomaly detection
                            ├── batch analytics
                            └── replay consumers

Connector design

MQTT subscriber connector
                            │
                            ├── subscribes to selected topics
                            ├── validates payload envelope
                            ├── adds metadata
                            │       ├── topic
                            │       ├── tenant
                            │       ├── site
                            │       ├── device_id
                            │       ├── received_at
                            │       └── qos
                            ├── builds partition key
                            └── writes to streaming topic

Partition key choices

Multi-region MQTT is difficult because clients maintain persistent connections, sessions, retained state and QoS flows. The architecture must decide whether regions are active/passive, active/active, site-affine, tenant-affine or edge-first.

Active/passive diagram

Normal state:

                            Devices -> Region A MQTT Cluster -> Backend A -> Storage A
                            │
                            └── replicated data/config -> Region B standby

                            Failover:

                            Devices reconnect -> Region B MQTT Cluster -> Backend B -> Storage B

Multi-region risks

DR checklist

[ ] Broker config replicated
                            [ ] ACLs replicated
                            [ ] Certificates available in DR region
                            [ ] Device registry replicated
                            [ ] Backend consumers deployed
                            [ ] Storage replication tested
                            [ ] DNS or endpoint failover tested
                            [ ] Client reconnect behavior tested
                            [ ] Command authority defined
                            [ ] Idempotency works across regions
                            [ ] Retained state recovery defined
                            [ ] Runbook documented
                            [ ] RTO and RPO defined

MQTT costs come from broker infrastructure, cloud message billing, TLS processing, bandwidth, storage writes, monitoring, retained state, logs, data egress and operational team time. Cloud managed IoT can be very efficient, but high-frequency telemetry can generate large message counts quickly.

Cost drivers

Message volume estimate

Formula:
                            messages_per_month =
                            devices
                            * messages_per_device_per_minute
                            * 60
                            * 24
                            * 30

                            Example:
                            10,000 devices
                            1 message every 10 seconds = 6 messages/minute

                            10,000 * 6 * 60 * 24 * 30
                            = 2,592,000,000 messages/month

                            That is why telemetry frequency matters.

Cost reduction levers

Limits to define upfront

Per client:
                            - max messages/sec
                            - max payload size
                            - max subscriptions
                            - max inflight
                            - max queued messages
                            - session expiry

                            Per tenant:
                            - max devices
                            - max messages/day
                            - max retained messages
                            - max command rate
                            - max storage retention

                            Per platform:
                            - broker nodes
                            - backend worker capacity
                            - database write capacity
                            - dashboard fan-out capacity

Architecture checklist

[ ] Expected connected clients defined
                            [ ] Expected messages/sec defined
                            [ ] Average payload size estimated
                            [ ] Fan-out ratio estimated
                            [ ] QoS policy defined per flow
                            [ ] Offline behavior defined
                            [ ] Retained state policy defined
                            [ ] Persistent session policy defined
                            [ ] Broker choice justified
                            [ ] Cluster or single broker decision justified
                            [ ] Edge bridge strategy defined
                            [ ] Shared subscription groups defined
                            [ ] Backend buffer chosen
                            [ ] Storage throughput tested
                            [ ] Dashboard fan-out tested
                            [ ] Security and ACLs tested
                            [ ] Monitoring and alerting ready
                            [ ] Cost model estimated
                            [ ] Quotas and limits documented

Cloud decision matrix

Production acceptance tests

[ ] Load test with expected client count
                            [ ] Load test with expected message rate
                            [ ] Burst test above expected peak
                            [ ] Reconnect storm test
                            [ ] Slow subscriber test
                            [ ] Broker node failure test
                            [ ] TLS certificate rotation test
                            [ ] ACL negative tests
                            [ ] Retained flood test
                            [ ] Offline edge bridge test
                            [ ] Replay duplicate test
                            [ ] Shared subscription scaling test
                            [ ] Backend database saturation test
                            [ ] Dashboard freshness test
                            [ ] Command ack timeout test
                            [ ] Disaster recovery failover test

Final architecture rule

Useful URLs

AWS IoT Core:
                            https://docs.aws.amazon.com/iot/

                            Azure IoT Hub:
                            https://learn.microsoft.com/azure/iot-hub/

                            EMQX:
                            https://www.emqx.com/

                            HiveMQ:
                            https://www.hivemq.com/

                            Mosquitto:
                            https://mosquitto.org/

                            RabbitMQ MQTT plugin:
                            https://www.rabbitmq.com/docs/mqtt

                            MQTT official:
                            https://mqtt.org/

MQTT and RabbitMQ do not target the same primary world. MQTT was designed for lightweight Pub/Sub communication with devices and unstable networks. RabbitMQ is an enterprise messaging broker built around exchanges, queues, routing keys, acknowledgements, retries and dead-lettering.

When RabbitMQ is better

• Backend job queues.
• Business workflows with retries.
• Dead-letter queues and retry policies.
• Routing to multiple internal queues.
• Celery or worker processing.
• Enterprise integration with AMQP.

Hybrid MQTT to RabbitMQ pattern

Devices
                            │
                            │ MQTT
                            ▼
                            MQTT Broker or RabbitMQ MQTT Plugin
                            │
                            ▼
                            AMQP Exchange
                            │
                            ├── queue.storage
                            │       └── database writer
                            │
                            ├── queue.alerts
                            │       └── alert engine
                            │
                            ├── queue.websocket
                            │       └── live dashboard
                            │
                            └── queue.dead_letter
                            └── failed messages

Decision examples

MQTT and Kafka are frequently complementary. MQTT is a connectivity and Pub/Sub protocol for clients and devices. Kafka is a durable distributed event log for replay, analytics, stream processing and data pipelines.

When Kafka is better

• Long-term event retention.
• Replay from offsets.
• Stream processing.
• Data lake ingestion.
• Multiple analytics consumers.
• High-throughput backend event pipelines.

MQTT to Kafka architecture

Devices / Gateways
                            │
                            │ MQTT/TLS
                            ▼
                            MQTT Broker
                            │
                            ├── operational dashboard
                            ├── alerting subscriber
                            └── Kafka connector
                            │
                            ▼
                            Kafka / Redpanda
                            │
                            ├── stream processing
                            ├── data lake
                            ├── ML pipeline
                            ├── audit replay
                            └── batch analytics

Topic mapping example

MQTT topic:
                            prod/acme/paris/line-02/gw-123/telemetry/temperature

                            Kafka topic options:
                            iot.telemetry
                            iot.telemetry.acme
                            iot.telemetry.temperature

                            Kafka key options:
                            device_id = gw-123
                            site_id = paris
                            hash(topic)

                            Recommended event metadata:
                            {
                            "mqtt_topic": "...",
                            "tenant": "acme",
                            "site": "paris",
                            "device_id": "gw-123",
                            "metric": "temperature",
                            "event_ts": "...",
                            "ingestion_ts": "..."
                            }

WebSocket is a browser and application transport for bidirectional real-time connections. MQTT is a Pub/Sub protocol with topics, QoS, retained messages and Last Will. MQTT can run over WebSocket, but they are not the same abstraction.

When WebSocket is better

• Interactive browser dashboards.
• Chat-like UI.
• Live admin screens.
• Application-specific push from backend to browser.
• When direct MQTT exposure to frontend is not desired.

Recommended dashboard pattern

Devices
                            │
                            │ MQTT
                            ▼
                            MQTT Broker
                            │
                            ▼
                            Backend subscriber
                            │
                            ├── validates data
                            ├── applies user permissions
                            ├── aggregates state
                            └── pushes safe updates
                            │
                            ▼
                            WebSocket Gateway
                            │
                            ▼
                            Browser Dashboard

Direct MQTT over WebSocket

Browser
                            │
                            │ MQTT over WebSocket
                            ▼
                            MQTT Broker

                            Use only if:
                            - broker supports WebSocket
                            - ACLs are extremely strict
                            - browser credentials are short-lived
                            - topic scope is limited per user
                            - sensitive industrial topics are not exposed
                            - rate limits are enforced

HTTP is the standard model for synchronous APIs, CRUD operations, public integrations and admin actions. MQTT is better for persistent, lightweight, topic-based telemetry and device communication. HTTP streaming and SSE can push events to clients, but they do not replace MQTT for device fleets.

When HTTP is better

• Admin APIs.
• CRUD operations.
• Public APIs.
• File uploads and downloads.
• Simple low-frequency device reporting.
• Human-driven requests.

MQTT + HTTP pattern

Device telemetry:
                            Device -> MQTT Broker -> Ingestion backend -> Time-series DB

                            Admin operations:
                            Browser -> HTTP API -> Django backend -> Database

                            Commands:
                            Browser -> HTTP API -> Command service -> MQTT command topic

                            Dashboard:
                            Backend -> WebSocket/SSE -> Browser

Avoid this confusion

NATS is a very fast cloud-native messaging system with subjects, request/reply, Pub/Sub and optional persistence through JetStream. It is excellent for service-to-service communication. MQTT is stronger when the problem is IoT, device connectivity, retained state, Last Will and MQTT-specific broker ecosystem.

When NATS is better

• Fast service-to-service messaging.
• Cloud-native microservices.
• Request/reply between internal services.
• Lightweight internal event bus.
• JetStream persistence for selected flows.

MQTT + NATS hybrid

Devices
                            │
                            │ MQTT
                            ▼
                            MQTT Broker
                            │
                            ▼
                            Ingestion service
                            │
                            ├── validates payload
                            ├── normalizes event
                            └── publishes internal event
                            │
                            ▼
                            NATS
                            │
                            ├── alert service
                            ├── dashboard service
                            ├── enrichment service
                            └── command orchestrator

Decision examples

Pattern 1: MQTT to Kafka for analytics

Devices
                            │
                            ▼
                            MQTT Broker
                            │
                            ▼
                            Kafka Connector
                            │
                            ▼
                            Kafka
                            │
                            ├── stream processing
                            ├── data lake
                            ├── ML features
                            └── replay consumers

Pattern 2: MQTT to RabbitMQ for workers

Devices
                            │
                            ▼
                            MQTT Broker
                            │
                            ▼
                            Ingestion service
                            │
                            ▼
                            RabbitMQ exchange
                            │
                            ├── queue.storage
                            ├── queue.alerting
                            ├── queue.enrichment
                            └── queue.dead_letter

Pattern 3: MQTT to WebSocket for UI

Devices
                            │
                            ▼
                            MQTT Broker
                            │
                            ▼
                            Backend subscriber
                            │
                            ├── checks permissions
                            ├── aggregates state
                            └── pushes safe updates
                            │
                            ▼
                            WebSocket
                            │
                            ▼
                            Browser dashboard

Pattern 4: HTTP API to MQTT command

Operator browser
                            │
                            │ HTTP POST /commands
                            ▼
                            Django API
                            │
                            ├── RBAC check
                            ├── device state check
                            ├── command journal
                            └── publish MQTT command
                            │
                            ▼
                            Device
                            │
                            ├── ack topic
                            └── result topic

Pattern 5: MQTT to NATS internal bus

MQTT Broker
                            │
                            ▼
                            Ingestion service
                            │
                            ├── validate
                            ├── normalize
                            └── publish internal event
                            │
                            ▼
                            NATS
                            │
                            ├── alert service
                            ├── state service
                            ├── notification service
                            └── command service

Architecture smell checklist

[ ] One technology is used for every communication pattern
                            [ ] Devices talk directly to Kafka
                            [ ] Browser subscribes to unrestricted MQTT wildcard
                            [ ] MQTT broker stores business history
                            [ ] Commands are sent without ack/result
                            [ ] RabbitMQ queues are used as time-series storage
                            [ ] REST endpoint receives thousands of telemetry posts per second
                            [ ] WebSocket server implements custom broker logic
                            [ ] No idempotency between MQTT and downstream queue
                            [ ] No clear ownership between MQTT, Kafka and database

Correction map

Need telemetry from devices:
                            MQTT

                            Need durable replay:
                            Kafka / Redpanda / Pulsar

                            Need worker retry and DLQ:
                            RabbitMQ

                            Need live browser UI:
                            WebSocket / SSE

                            Need CRUD API:
                            HTTP REST / GraphQL

                            Need internal fast messaging:
                            NATS

                            Need large file transfer:
                            Object storage + HTTP

Recommended combinations

One-line recommendations

Use MQTT when:
                            devices, telemetry, edge, Pub/Sub, QoS, retained state, Last Will.

                            Use RabbitMQ when:
                            backend queues, retries, DLQ, enterprise routing.

                            Use Kafka when:
                            replay, analytics, durable event log, stream processing.

                            Use WebSocket when:
                            browser needs live updates.

                            Use HTTP when:
                            request/response, CRUD, public API, file transfer.

                            Use NATS when:
                            fast internal cloud-native messaging or request/reply.

Final architecture example

Industrial SaaS platform:

                            Field layer:
                            Devices / PLC gateways
                            │
                            ▼
                            MQTT + Sparkplug

                            Messaging layer:
                            MQTT broker cluster
                            │
                            ├── live operations
                            ├── command service
                            └── connector

                            Streaming layer:
                            Kafka / Redpanda
                            │
                            ├── replay
                            ├── analytics
                            └── data lake

                            Backend layer:
                            RabbitMQ / Celery
                            │
                            ├── enrichment
                            ├── notifications
                            └── reports

                            UI layer:
                            WebSocket
                            │
                            ▼
                            Browser dashboards

                            API layer:
                            HTTP REST / GraphQL
                            │
                            ▼
                            Admin, users, configuration

Final checklist

[ ] Communication pattern identified
                            [ ] Producer type identified
                            [ ] Consumer type identified
                            [ ] Need for replay decided
                            [ ] Need for queue retries decided
                            [ ] Need for browser live UI decided
                            [ ] Need for public API decided
                            [ ] Need for device QoS decided
                            [ ] Need for retained state decided
                            [ ] Need for Last Will decided
                            [ ] Storage responsibility separated
                            [ ] Command responsibility separated
                            [ ] Security boundary defined
                            [ ] Operational ownership defined

Bad topic examples

Bad:
                            data
                            temperature
                            Device 123
                            Factory/Paris/Line 1/Temperature
                            acme/all
                            prod/+/+/+/#
                            commands
                            telemetry_and_commands/gw-123

Correct topic structure

Recommended:
                            prod/{tenant}/{site}/{area}/{asset}/{stream}/{signal}

                            Examples:
                            prod/acme/paris/line-02/gw-123/telemetry/temperature
                            prod/acme/paris/line-02/gw-123/status/availability
                            prod/acme/paris/line-02/gw-123/event/alarm
                            prod/acme/paris/line-02/gw-123/command/reboot
                            prod/acme/paris/line-02/gw-123/ack/reboot
                            prod/acme/paris/line-02/gw-123/result/reboot

Good subscription examples

All telemetry for one tenant:
                            prod/acme/+/+/+/telemetry/#

                            One site:
                            prod/acme/paris/#

                            One device:
                            prod/acme/paris/line-02/gw-123/#

                            All alarms:
                            prod/acme/+/+/+/event/alarm

                            All command results:
                            prod/acme/+/+/+/result/#