MQTT en Elixir avec Tortoise311 et emqtt

Elixir dispose de deux clients MQTT solides : Tortoise311, un client 100 % Elixir pour MQTT 3.1.1 avec un behaviour de handler et des reconnexions automatiques, et emqtt, le client Erlang de l'équipe EMQX qui parle MQTT 5. Les deux s'intègrent naturellement dans un arbre de supervision OTP : une panne du broker devient un redémarrage, pas une application plantée.

Ce tutoriel construit une petite app telemetry qui s'abonne à des topics de capteurs, les distingue par pattern matching dans les têtes de fonction et publie en QoS 1 — d'abord avec Tortoise311, puis avec emqtt pour MQTT 5. Tous les exemples ciblent broker.connectmq.com, vous pouvez donc suivre le trafic dans le client MQTT en ligne pendant que vous codez.

Quelle bibliothèque ?

Tortoise311 est un fork maintenu de la bibliothèque Tortoise d'origine ; l'API est identique, seul le module a été renommé. Si vous ne savez pas si vous avez besoin de MQTT 5, lisez MQTT 5 vs 3.1.1 — pour de la télémétrie et des commandes simples, 3.1.1 suffit.

Préparer le projet

mix new telemetry --sup
cd telemetry

defp deps do
  [
    {:tortoise311, "~> 0.12"},
    # Only needed for the MQTT 5 section. Pin a release tag in real projects.
    {:emqtt, github: "emqx/emqtt", system_env: [{"BUILD_WITHOUT_QUIC", "1"}]}
  ]
end

BUILD_WITHOUT_QUIC=1 ignore la dépendance QUIC optionnelle d'emqtt, qui exige sinon une toolchain C et CMake. Lancez mix deps.get, puis exportez les identifiants de votre broker avant de démarrer IEx :

export MQTT_USER="username"
export MQTT_PASS="password"

Une connexion Tortoise311 supervisée

Une connexion Tortoise311 n'est qu'un enfant de plus dans votre arbre de supervision. Elle possède le socket TCP, envoie les pings de keep alive, se reconnecte et se réabonne toute seule.

defmodule Telemetry.Application do
  use Application

  @impl true
  def start(_type, _args) do
    children = [
      {Tortoise311.Connection,
       client_id: Telemetry.Mqtt.client_id(),
       server: {Tortoise311.Transport.Tcp, host: ~c"broker.connectmq.com", port: 1883},
       user_name: System.fetch_env!("MQTT_USER"),
       password: System.fetch_env!("MQTT_PASS"),
       keep_alive: 30,
       handler: {Telemetry.MqttHandler, []},
       subscriptions: [{"testmqtt/elixir/+/temp", 1}, {"testmqtt/elixir/+/cmd", 1}]}
    ]

    Supervisor.start_link(children, strategy: :one_for_one, name: Telemetry.Supervisor)
  end
end

defmodule Telemetry.Mqtt do
  # One client ID per node: two nodes sharing an ID kick each other off the broker.
  def client_id do
    {:ok, host} = :inet.gethostname()
    "testmqtt-elixir-#{host}"
  end

  def publish_temp(device, celsius) do
    payload = Jason.encode!(%{device: device, celsius: celsius})
    Tortoise311.publish_sync(client_id(), "testmqtt/elixir/#{device}/temp", payload,
      qos: 1, timeout: 5_000)
  end
end

L'exemple utilise Jason pour le JSON ; ajoutez {:jason, "~> 1.4"} à vos dépendances ou remplacez-le par une simple chaîne. publish_sync/4 bloque jusqu'à ce que le broker accuse réception du message QoS 1 (PUBACK) et renvoie :ok ; publish/4 renvoie immédiatement {:ok, ref} et envoie l'accusé de réception plus tard à l'appelant sous forme de message. Les client IDs comptent plus qu'il n'y paraît ; le générateur de client ID vous aide à choisir un schéma unique.

Traiter les messages par pattern matching

Tortoise311 découpe chaque topic entrant en une liste de niveaux : le routage devient du pattern matching Elixir ordinaire — pas de parsing de chaînes, pas de regex.

defmodule Telemetry.MqttHandler do
  use Tortoise311.Handler
  require Logger

  @impl true
  def init(args), do: {:ok, args}

  @impl true
  def connection(status, state) do
    # status is :up, :down or :terminating
    Logger.info("MQTT connection #{status}")
    {:ok, state}
  end

  @impl true
  def subscription(status, topic_filter, state) do
    Logger.info("subscription #{topic_filter}: #{inspect(status)}")
    {:ok, state}
  end

  @impl true
  def handle_message(["testmqtt", "elixir", device, "temp"], payload, state) do
    Logger.info("#{device} reports #{payload}")
    {:ok, state}
  end

  def handle_message(["testmqtt", "elixir", device, "cmd"], "reboot", state) do
    Logger.warning("reboot requested for #{device}")
    {:ok, state}
  end

  def handle_message(topic, _payload, state) do
    Logger.debug("ignored #{Enum.join(topic, "/")}")
    {:ok, state}
  end

  @impl true
  def terminate(_reason, _state), do: :ok
end

Le wildcard + de l'abonnement devient la variable device dans la tête de fonction. Gardez une clause fourre-tout à la fin : un abonnement avec wildcard peut livrer des topics imprévus, et une clause manquante lève FunctionClauseError dans le processus de connexion. Si les filtres sont nouveaux pour vous, topics et wildcards MQTT explique comment + et # fonctionnent.

Tester de bout en bout

Démarrez l'app avec iex -S mix. Vous devriez voir MQTT connection up et deux lignes subscription ... :ok. Publiez ensuite depuis un autre terminal :

mosquitto_pub -h broker.connectmq.com -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t testmqtt/elixir/sensor-1/temp -m '{"celsius":21.5}' -q 1
mosquitto_pub -h broker.connectmq.com -u "$MQTT_USER" -P "$MQTT_PASS" \
  -t testmqtt/elixir/sensor-1/cmd -m reboot -q 1

Et dans l'autre sens, depuis IEx :

iex> Telemetry.Mqtt.publish_temp("sensor-2", 22.1)
:ok

Abonnez-vous à testmqtt/elixir/# dans le client en ligne pour voir arriver les deux messages. Pour vérifier la supervision, tuez le processus de connexion avec Process.exit(pid, :kill) (trouvez-le avec Supervisor.which_children(Telemetry.Supervisor)) et regardez le superviseur en démarrer un nouveau qui se réabonne tout seul.

Suivre les topics Elixir dans le navigateur

TLS sur le port 8883

Passez le transport à Tortoise311.Transport.SSL et fournissez-lui le magasin de CA du système (OTP 25 ou plus récent) :

server:
  {Tortoise311.Transport.SSL,
   host: ~c"broker.connectmq.com",
   port: 8883,
   verify: :verify_peer,
   cacerts: :public_key.cacerts_get()}

Avec verify: :verify_peer, le module :ssl d'Erlang vérifie à la fois la chaîne de certificats et le nom d'hôte. Ne « corrigez » jamais une erreur de handshake en passant à :verify_none en production ; vérifiez plutôt le nom d'hôte et la CA, comme expliqué dans tester une connexion MQTT TLS.

MQTT 5 avec emqtt

emqtt livre les messages entrants à son processus propriétaire ; le wrapper idiomatique est donc un GenServer qui démarre le client avec start_link/1. Comme les deux processus sont liés, une connexion perdue fait planter le wrapper et le superviseur redémarre les deux — le « let it crash » utilisé comme stratégie de reconnexion.

defmodule Telemetry.Mqtt5 do
  use GenServer
  require Logger

  def start_link(opts), do: GenServer.start_link(__MODULE__, opts, name: __MODULE__)
  def publish(topic, payload), do: GenServer.call(__MODULE__, {:publish, topic, payload})

  @impl true
  def init(_opts) do
    {:ok, conn} =
      :emqtt.start_link(
        host: ~c"broker.connectmq.com",
        port: 1883,
        clientid: Telemetry.Mqtt.client_id() <> "-v5",
        username: System.fetch_env!("MQTT_USER"),
        password: System.fetch_env!("MQTT_PASS"),
        proto_ver: :v5,
        clean_start: false,
        properties: %{"Session-Expiry-Interval": 3600}
      )

    {:ok, _props} = :emqtt.connect(conn)
    {:ok, _props, _reason_codes} = :emqtt.subscribe(conn, "testmqtt/elixir/v5/#", qos: 1)
    {:ok, %{conn: conn}}
  end

  @impl true
  def handle_call({:publish, topic, payload}, _from, %{conn: conn} = state) do
    {:reply, :emqtt.publish(conn, topic, payload, 1), state}
  end

  @impl true
  def handle_info({:publish, %{topic: topic, payload: payload}}, state) do
    Logger.info("v5 #{topic}: #{payload}")
    {:noreply, state}
  end

  def handle_info({:disconnected, reason_code, _props}, state) do
    {:stop, {:disconnected, reason_code}, state}
  end
end

Ajoutez Telemetry.Mqtt5 à la liste des enfants. clean_start: false associé à un session expiry d'une heure signifie que le broker conserve vos abonnements et les messages QoS 1 en attente pendant que le nœud redémarre — voir sessions persistantes. Le endpoint de test partagé ne conserve pas les sessions hors ligne : testez donc cette partie sur votre propre instance de broker privée. Si connect/1 échoue, le pattern match lève une erreur ; le reason code vous dit pourquoi (:bad_username_or_password, :not_authorized, etc.) et la recherche de reason codes MQTT les explique un par un.

Intégrer MQTT dans une app plus large

Dans un projet Phoenix ou Nerves, traitez la connexion MQTT comme une frontière du système, pas comme l'endroit où vit la logique métier :

• Une connexion par broker et par nœud. Tout processus qui veut publier appelle Telemetry.Mqtt.publish_temp/2 (ou un module similaire) ; aucun n'ouvre son propre socket.
• Diffusez les messages à l'intérieur de la BEAM. Dans handle_message/3, diffusez le payload décodé via Phoenix.PubSub ou un Registry, pour que LiveViews, GenServers et pipelines s'abonnent localement sans savoir que MQTT existe.
• Décodez à la frontière. Parsez le JSON une seule fois dans le handler et transmettez des structs ; rejetez les payloads mal formés avec une ligne de log au lieu de planter.
• Faites des topics une configuration. Placez le préfixe de topic dans config/runtime.exs, pour que staging et production partagent le code mais jamais les topics.

Cela facilite aussi les tests : les tests unitaires appellent vos modules directement, tandis qu'un seul test d'intégration prouve qu'un message publié sur le vrai broker arrive bien dans le PubSub local.

Problèmes courants

• La connexion oscille entre up et down — un autre nœud utilise le même client ID. Vérifiez avec :inet.gethostname() sur chaque nœud et consultez client ID en double.
• {:error, :econnrefused} ou aucun événement up — mauvais port ou pare-feu. Le port 1883 est du TCP simple, le 8883 du TLS ; les confondre ressemble à un timeout.
• Les messages arrivent en QoS 0 alors que vous publiez en QoS 1 — l'abonnement a été accordé en QoS 0. Inspectez le statut dans subscription/3 ou les reason codes renvoyés par :emqtt.subscribe/3.
• FunctionClauseError dans handle_message/3 — ajoutez une clause fourre-tout ; les filtres avec wildcard livrent plus que prévu.

Le modèle de processus d'Elixir rend les clients MQTT particulièrement robustes : une connexion supervisée par broker, des topics en pattern matching et le travail lent sorti du handler. Commencez avec Tortoise311, passez à emqtt quand vous avez besoin de MQTT 5, et gardez un onglet de navigateur ouvert sur les topics pendant vos tests.

Questions fréquentes