Publier sur MQTT depuis AWS Lambda et des fonctions serverless

Vous pouvez publier vers un broker MQTT depuis AWS Lambda (ou n'importe quelle fonction serverless) avec une bibliothèque cliente MQTT ordinaire — la difficulté, c'est le cycle de vie. Une fonction vit quelques centaines de millisecondes, est gelée entre les invocations et peut tourner en dizaines de copies parallèles. Connectez-vous à la demande, réutilisez le client entre invocations à chaud, attendez l'acquittement du publish avant de rendre la main et donnez à chaque environnement d'exécution son propre client ID. Si vous ne faites que publier, un pont HTTP peut être encore plus simple.

Ce tutoriel présente un handler Python prêt pour la production (paho-mqtt 2.x), la même idée en Node.js, l'alternative HTTP et la façon de tester l'ensemble.

Le cycle de vie serverless face à MQTT

MQTT suppose des connexions TCP longue durée avec des pings de keep-alive. Lambda suppose des exécutions courtes et sans état. Le décalage apparaît à quatre endroits :

Le gel est le point subtil. Si un environnement reste inactif plus d'environ 1,5× l'intervalle de keep-alive, le broker ferme la connexion. Au dégel, le client se croit toujours connecté jusqu'à sa prochaine écriture. Un client mis en cache naïvement échouera donc, ou restera bloqué, à la première invocation après une période calme.

Stratégie 1 : une connexion par invocation

Le schéma correct le plus simple : se connecter, publier, attendre le PUBACK, se déconnecter. Rien ne survit entre deux appels, donc rien ne peut devenir obsolète.

• Avantages : aucun bug de connexion obsolète, aucune collision de client ID avec un ID aléatoire, facile à comprendre.
• Inconvénients : chaque invocation paie le handshake de connexion — nettement plus avec TLS et depuis une région éloignée — et une rafale d'invocations devient une rafale de nouvelles connexions sur le broker.

C'est un bon choix pour les déclencheurs à faible volume : un job nocturne, une notification d'upload S3, un webhook occasionnel.

Stratégie 2 : réutiliser la connexion entre invocations à chaud

Pour tout ce qui est fréquent, créez le client dans la portée globale afin que les invocations à chaud évitent le handshake, et méfiez-vous de la connexion en cache après une période d'inactivité.

import json, os, threading, time, uuid
import paho.mqtt.client as mqtt

HOST = os.environ.get("MQTT_HOST", "broker.connectmq.com")
PORT = int(os.environ.get("MQTT_PORT", "8883"))
KEEPALIVE = 60
MAX_IDLE = KEEPALIVE            # après cette pause, supposer que le broker nous a coupés

_client = None
_connected = threading.Event()
_last_used = 0.0

def _on_connect(client, userdata, flags, reason_code, properties):
    if reason_code.is_failure:
        print(f"connect refused: {reason_code}")
    else:
        _connected.set()

def _on_disconnect(client, userdata, flags, reason_code, properties):
    _connected.clear()

def _new_client():
    c = mqtt.Client(
        mqtt.CallbackAPIVersion.VERSION2,
        client_id=f"lambda-{uuid.uuid4().hex[:12]}",   # unique par environnement
        protocol=mqtt.MQTTv5,
    )
    c.username_pw_set(os.environ["MQTT_USER"], os.environ["MQTT_PASS"])
    c.tls_set()                                         # bundle de CA du système
    c.on_connect, c.on_disconnect = _on_connect, _on_disconnect
    c.connect_async(HOST, PORT, keepalive=KEEPALIVE)
    c.loop_start()
    return c

def _get_client(timeout=5.0):
    global _client
    if _client is not None and time.time() - _last_used > MAX_IDLE:
        _client.disconnect()                            # obsolète après un gel
        _client.loop_stop()
        _connected.clear()
        _client = None
    if _client is None:
        _client = _new_client()
    if not _connected.wait(timeout):
        raise TimeoutError("MQTT connect timed out")
    return _client

def handler(event, context):
    global _last_used
    client = _get_client()
    payload = json.dumps({"source": "lambda", "event": event, "ts": time.time()})
    info = client.publish("testmqtt/lambda/events", payload, qos=1)
    info.wait_for_publish(timeout=5)
    if not info.is_published():
        raise RuntimeError("PUBACK not received")       # laisser Lambda réessayer
    _last_used = time.time()
    return {"statusCode": 200, "body": json.dumps({"mid": info.mid})}

Le rôle de chaque élément :

• connect_async + loop_start — la boucle réseau tourne dans un thread d'arrière-plan qui gère CONNACK, PUBACK et les reconnexions pendant l'exécution de la fonction.
• MAX_IDLE — si l'environnement est resté inactif plus longtemps que le keep-alive, le code jette le client et se reconnecte au lieu d'écrire dans un socket mort.
• wait_for_publish — ne rendez jamais la main avant l'acquittement QoS 1. Une fois le handler terminé, l'environnement peut être gelé avec le message encore dans le buffer du client.
• Lever une exception en cas d'échec — pour les déclencheurs asynchrones, Lambda relance les invocations en échec, exactement ce qu'il faut pour un publish perdu.
• Client ID aléatoire — deux environnements concurrents avec le même ID se déconnecteraient mutuellement en boucle (client ID dupliqué).

Empaquetez avec pip install paho-mqtt -t package/ puis zippez le dossier avec le handler, ou utilisez une Lambda layer. Fixez le timeout de la fonction nettement au-dessus du timeout de connexion (10–15 s est un point de départ raisonnable) et stockez les identifiants dans des variables d'environnement alimentées par Secrets Manager ou Parameter Store.

Le même schéma en Node.js

Avec MQTT.js 5, le client en cache est une variable de module. La bibliothèque se reconnecte seule, mais après un gel il vaut mieux vérifier la connexion avant de publier.

import mqtt from 'mqtt';
import { randomUUID } from 'node:crypto';

let client;
let lastUsed = 0;

async function getClient() {
  if (client && Date.now() - lastUsed > 60_000) { await client.endAsync(true); client = undefined; }
  if (!client) {
    client = await mqtt.connectAsync('mqtts://broker.connectmq.com:8883', {
      clientId: `lambda-${randomUUID().slice(0, 12)}`,
      username: process.env.MQTT_USER,
      password: process.env.MQTT_PASS,
      keepalive: 60,
      connectTimeout: 5000,
    });
  }
  return client;
}

export const handler = async (event) => {
  const c = await getClient();
  await c.publishAsync('testmqtt/lambda/events', JSON.stringify({ source: 'lambda', event }), { qos: 1 });
  lastUsed = Date.now();
  return { statusCode: 200 };
};

publishAsync en QoS 1 ne se résout qu'après le PUBACK : rendre la main après le await est donc sûr.

Stratégie 3 : publier via HTTP

Si vos fonctions ne font que publier et n'ont pas besoin de sémantique MQTT côté client, une requête HTTP s'accorde naturellement avec le serverless : pas d'état de connexion, pas de client IDs, pas de problème de gel.

• API HTTP du broker. Certains brokers exposent un endpoint REST qui publie un message à votre place — par exemple l'API HTTP d'EMQX ou l'endpoint HTTPS du data plane d'AWS IoT Core. Consultez la documentation de votre broker pour l'authentification et le format du payload.
• Votre propre pont. Un petit service permanent qui maintient une connexion MQTT et accepte des requêtes POST /publish authentifiées. Il centralise les identifiants et transforme mille connexions Lambda éphémères en une seule connexion persistante.

Le compromis : vous perdez la QoS de bout en bout — le HTTP 200 indique que le pont a accepté le message, pas qu'un subscriber l'a reçu — et vous ajoutez un composant à exploiter. Pour une comparaison plus large des deux protocoles, voir MQTT vs HTTP.

Et l'abonnement ?

Une fonction Lambda ne peut pas être un subscriber MQTT longue durée : elle ne tourne que pendant une invocation. Inversez le sens — laissez un composant permanent s'abonner et invoquer la fonction :

• Une règle ou une intégration du broker qui transfère les messages correspondants vers un endpoint HTTP (API Gateway ou une function URL Lambda). Voir MQTT vers webhook pour le schéma général.
• Un petit service abonné qui pousse les messages dans SQS, qui déclenche ensuite Lambda par lots.

Erreurs courantes

• Task timed out after 3.00 seconds — le timeout par défaut de Lambda est plus court qu'une connexion TLS à froid depuis une autre région. Augmentez-le et vérifiez que le port du broker est joignable.
• Le connect se bloque dans un VPC — les fonctions dans un VPC ont besoin d'une NAT gateway pour sortir vers Internet. Voir timeout de connexion MQTT.
• Not authorized / reason code 0x87 — identifiants erronés ou ACL qui n'autorise pas le topic ; la référence des reason codes donne la signification exacte.
• Des messages manquent après le retour de la fonction — vous avez rendu la main avant le PUBACK. Attendez toujours la fin du publish.

Tester

1. Ouvrez le client MQTT en ligne, connectez-vous à broker.connectmq.com et abonnez-vous à testmqtt/lambda/#.
2. Invoquez la fonction depuis la console ou avec aws lambda invoke --function-name mqtt-publisher --payload '{"hello":"world"}' --cli-binary-format raw-in-base64-out out.json.
3. Le message doit apparaître dans le navigateur en moins d'une seconde. Invoquez de nouveau aussitôt : l'invocation à chaud doit être nettement plus rapide dans la métrique de durée de la fonction.
4. Attendez quelques minutes, invoquez encore et vérifiez que le chemin « connexion obsolète » se reconnecte proprement au lieu d'expirer.

Voir arriver les messages de votre Lambda

L'endpoint de test partagé est limité en débit et ne conserve pas de sessions, ce qui suffit pour cette vérification ; pour des tests de charge avec de nombreuses invocations parallèles, pointez la fonction vers votre propre instance de broker privée.

Serverless et MQTT font bon ménage dès lors que vous respectez le cycle de vie : connexion à la demande, méfiance envers les connexions après un gel, attente de l'acquittement et identité propre à chaque environnement. Si vous avez seulement besoin de publier en masse sans attendre de réponse, un pont HTTP supprime complètement la gestion des connexions.

Questions fréquentes