MQTT en Elixir con Tortoise311 y emqtt

Elixir tiene dos clientes MQTT sólidos: Tortoise311, un cliente en Elixir puro para MQTT 3.1.1 con un behaviour de handler y reconexiones automáticas, y emqtt, el cliente Erlang del equipo de EMQX que habla MQTT 5. Ambos encajan de forma natural en un árbol de supervisión OTP, así que una caída del broker se convierte en un reinicio, no en una aplicación rota.

Este tutorial construye una pequeña app telemetry que se suscribe a topics de sensores, los distingue con pattern matching en las cabeceras de función y publica con QoS 1: primero con Tortoise311 y luego con emqtt para MQTT 5. Todos los ejemplos apuntan a broker.connectmq.com, así que puedes ver el tráfico en el cliente MQTT online mientras programas.

¿Qué librería elegir?

Tortoise311 es un fork mantenido de la librería Tortoise original; la API es la misma con el módulo renombrado. Si no tienes claro si necesitas MQTT 5, lee MQTT 5 vs 3.1.1: para telemetría y comandos sencillos, 3.1.1 es suficiente.

Preparar el proyecto

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 omite la dependencia opcional de QUIC de emqtt, que de lo contrario necesita un toolchain de C y CMake. Ejecuta mix deps.get y exporta las credenciales de tu broker antes de arrancar IEx:

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

Una conexión Tortoise311 supervisada

Una conexión Tortoise311 es simplemente un hijo más de tu árbol de supervisión. Es dueña del socket TCP, envía los pings de keep alive, se reconecta y se vuelve a suscribir por sí sola.

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

El ejemplo usa Jason para JSON; añade {:jason, "~> 1.4"} a tus dependencias o sustitúyelo por un string normal. publish_sync/4 se bloquea hasta que el broker confirma el mensaje QoS 1 (PUBACK) y devuelve :ok; publish/4 devuelve {:ok, ref} al instante y envía la confirmación al llamador como mensaje más tarde. Los client IDs importan más de lo que parece; el generador de client IDs te ayuda a elegir un esquema único.

Gestionar mensajes con pattern matching

Tortoise311 divide cada topic entrante en una lista de niveles, así que el enrutado se convierte en pattern matching normal de Elixir: sin parsear strings ni usar 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

El comodín + de la suscripción se convierte en la variable device de la cabecera de la función. Deja siempre una cláusula comodín al final: una suscripción con wildcard puede entregar topics que no esperabas, y si falta la cláusula se lanza FunctionClauseError dentro del proceso de conexión. Si los filtros son nuevos para ti, topics y wildcards en MQTT explica cómo funcionan + y #.

Probarlo de extremo a extremo

Arranca la app con iex -S mix. Deberías ver MQTT connection up y dos líneas subscription ... :ok. Después publica desde otra 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

Y en la otra dirección, desde IEx:

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

Suscríbete a testmqtt/elixir/# en el cliente online para ver llegar ambos mensajes. Para comprobar la supervisión, mata el proceso de conexión con Process.exit(pid, :kill) (lo encuentras con Supervisor.which_children(Telemetry.Supervisor)) y observa cómo el supervisor arranca uno nuevo que se vuelve a suscribir solo.

Mira los topics de Elixir en tu navegador

TLS en el puerto 8883

Cambia el transporte a Tortoise311.Transport.SSL y pásale el almacén de CAs del sistema (OTP 25 o superior):

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

Con verify: :verify_peer, el módulo :ssl de Erlang comprueba tanto la cadena de certificados como el hostname. Nunca "arregles" un error de handshake cambiando a :verify_none en producción; revisa el hostname y la CA, como se explica en probar una conexión MQTT con TLS.

MQTT 5 con emqtt

emqtt entrega los mensajes entrantes a su proceso propietario, así que el envoltorio idiomático es un GenServer que arranca el cliente con start_link/1. Como ambos procesos están enlazados, una conexión perdida tumba el envoltorio y el supervisor reinicia los dos: "let it crash" usado como estrategia de reconexión.

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

Añade Telemetry.Mqtt5 a la lista de hijos. clean_start: false junto con un session expiry de una hora significa que el broker conserva tus suscripciones y los mensajes QoS 1 en cola mientras el nodo se reinicia; consulta sesiones persistentes. El endpoint de pruebas compartido no guarda sesiones offline, así que prueba esta parte contra tu propia instancia privada de broker. Si connect/1 falla, el pattern match lanza una excepción; el reason code te dice por qué (:bad_username_or_password, :not_authorized, etc.) y la consulta de reason codes MQTT explica cada uno.

Integrar MQTT en una app más grande

En un proyecto Phoenix o Nerves, trata la conexión MQTT como un borde del sistema y no como el lugar donde vive la lógica de negocio:

• Una conexión por broker y por nodo. Todo proceso que quiera publicar llama a Telemetry.Mqtt.publish_temp/2 (o a un módulo similar); ninguno abre su propio socket.
• Reparte los mensajes dentro de la BEAM. En handle_message/3, difunde el payload decodificado por Phoenix.PubSub o un Registry, para que LiveViews, GenServers y pipelines se suscriban localmente sin saber que MQTT existe.
• Decodifica en el borde. Parsea el JSON una sola vez en el handler y pasa structs hacia dentro; rechaza los payloads mal formados con una línea de log en lugar de romper el proceso.
• Haz que los topics sean configuración. Guarda el prefijo de topic en config/runtime.exs, para que staging y producción compartan código pero nunca topics.

Esto también facilita las pruebas: los tests unitarios llaman directamente a tus módulos, mientras que un único test de integración demuestra que un mensaje publicado en el broker real llega al PubSub local.

Problemas comunes

• La conexión oscila entre up y down — otro nodo usa el mismo client ID. Compruébalo con :inet.gethostname() en cada nodo y consulta client ID duplicado.
• {:error, :econnrefused} o ningún evento up — puerto equivocado o un firewall. El puerto 1883 es TCP plano y el 8883 es TLS; confundirlos parece un timeout.
• Los mensajes llegan con QoS 0 aunque publicaste con QoS 1 — la suscripción se concedió con QoS 0. Revisa el estado en subscription/3 o los reason codes que devuelve :emqtt.subscribe/3.
• FunctionClauseError en handle_message/3 — añade una cláusula comodín; los filtros con wildcard entregan más de lo que esperas.

El modelo de procesos de Elixir hace que los clientes MQTT sean especialmente robustos: una conexión supervisada por broker, topics resueltos con pattern matching y el trabajo lento fuera del handler. Empieza con Tortoise311, recurre a emqtt cuando necesites MQTT 5 y deja una pestaña del navegador abierta con los topics mientras pruebas.

Preguntas frecuentes