MQTT em Elixir com Tortoise311 e emqtt

Elixir tem dois clientes MQTT sólidos: Tortoise311, um cliente em Elixir puro para MQTT 3.1.1 com um behaviour de handler e reconexão automática, e emqtt, o cliente Erlang do time do EMQX que fala MQTT 5. Os dois se encaixam naturalmente em uma árvore de supervisão OTP, então uma queda do broker vira um restart, não uma aplicação quebrada.

Este tutorial monta um pequeno app telemetry que assina topics de sensores, faz pattern matching deles nas cabeças de função e publica com QoS 1 — primeiro com Tortoise311, depois com emqtt para MQTT 5. Todos os exemplos usam broker.connectmq.com, então você pode acompanhar o tráfego no cliente MQTT online enquanto programa.

Qual biblioteca?

Tortoise311 é um fork mantido da biblioteca Tortoise original; a API é a mesma, só o módulo foi renomeado. Se você não sabe se precisa de MQTT 5, leia MQTT 5 vs 3.1.1 — para telemetria e comandos simples, 3.1.1 basta.

Configurando o projeto

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 pula a dependência opcional de QUIC do emqtt, que de outra forma exige um toolchain C e CMake. Rode mix deps.get e exporte as credenciais do seu broker antes de abrir o IEx:

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

Uma conexão Tortoise311 supervisionada

Uma conexão Tortoise311 é só mais um filho na sua árvore de supervisão. Ela é dona do socket TCP, envia os pings de keep alive, reconecta e assina de novo sozinha.

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

O exemplo usa Jason para JSON; adicione {:jason, "~> 1.4"} às dependências ou troque por uma string simples. publish_sync/4 bloqueia até o broker confirmar a mensagem QoS 1 (PUBACK) e retorna :ok; publish/4 retorna {:ok, ref} na hora e envia a confirmação ao chamador como mensagem depois. Client IDs importam mais do que parece; o gerador de client ID ajuda a escolher um esquema único.

Tratando mensagens com pattern matching

O Tortoise311 divide cada topic recebido em uma lista de níveis, então o roteamento vira pattern matching comum de Elixir — sem parsing de string, sem 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

O wildcard + da assinatura vira a variável device na cabeça da função. Mantenha uma cláusula genérica no final: uma assinatura com wildcard pode entregar topics que você não previu, e uma cláusula faltando gera FunctionClauseError dentro do processo de conexão. Se filtros são novidade para você, topics e wildcards no MQTT explica como + e # funcionam.

Testando de ponta a ponta

Inicie o app com iex -S mix. Você deve ver MQTT connection up e duas linhas subscription ... :ok. Depois publique de outro 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

E no sentido contrário, a partir do IEx:

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

Assine testmqtt/elixir/# no cliente online para ver as duas mensagens chegarem. Para provar a supervisão, mate o processo de conexão com Process.exit(pid, :kill) (encontre-o com Supervisor.which_children(Telemetry.Supervisor)) e veja o supervisor iniciar um novo, que assina de novo sozinho.

Acompanhe os topics do Elixir no navegador

TLS na porta 8883

Troque o transporte para Tortoise311.Transport.SSL e passe o repositório de CAs do sistema (OTP 25 ou mais recente):

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

Com verify: :verify_peer, o :ssl do Erlang verifica tanto a cadeia de certificados quanto o hostname. Nunca "resolva" um erro de handshake trocando para :verify_none em produção; confira o hostname e a CA, como descrito em testar uma conexão MQTT com TLS.

MQTT 5 com emqtt

O emqtt entrega as mensagens recebidas ao processo dono, então o wrapper idiomático é um GenServer que inicia o cliente com start_link/1. Como os dois processos estão linkados, uma conexão perdida derruba o wrapper e o supervisor reinicia ambos — "let it crash" usado como estratégia de reconexão.

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

Adicione Telemetry.Mqtt5 à lista de filhos. clean_start: false mais um session expiry de uma hora significa que o broker guarda suas assinaturas e as mensagens QoS 1 enfileiradas enquanto o nó reinicia — veja sessões persistentes. O endpoint de teste compartilhado não mantém sessões offline, então teste esta parte na sua própria instância privada de broker. Se connect/1 falhar, o pattern match lança uma exceção; o reason code diz o motivo (:bad_username_or_password, :not_authorized etc.), e a consulta de reason codes MQTT explica cada um.

Encaixando MQTT em um app maior

Em um projeto Phoenix ou Nerves, trate a conexão MQTT como uma borda do sistema, e não como o lugar onde mora a regra de negócio:

• Uma conexão por broker por nó. Todo processo que quer publicar chama Telemetry.Mqtt.publish_temp/2 (ou um módulo parecido); nenhum abre seu próprio socket.
• Distribua as mensagens dentro da BEAM. Em handle_message/3, faça broadcast do payload decodificado em Phoenix.PubSub ou em um Registry, para que LiveViews, GenServers e pipelines assinem localmente sem saber que o MQTT existe.
• Decodifique na borda. Faça o parse do JSON uma vez no handler e passe structs adiante; rejeite payloads malformados com uma linha de log em vez de derrubar o processo.
• Transforme topics em configuração. Guarde o prefixo de topic em config/runtime.exs, para que staging e produção compartilhem código, mas nunca topics.

Isso também facilita os testes: testes unitários chamam seus módulos diretamente, enquanto um único teste de integração prova que uma mensagem publicada no broker real chega ao PubSub local.

Problemas comuns

• A conexão fica alternando entre up e down — outro nó usa o mesmo client ID. Confira com :inet.gethostname() em cada nó e veja client ID duplicado.
• {:error, :econnrefused} ou nenhum evento up — porta errada ou firewall. A porta 1883 é TCP puro, a 8883 é TLS; confundir as duas parece um timeout.
• As mensagens chegam em QoS 0 embora você tenha publicado em QoS 1 — a assinatura foi concedida em QoS 0. Verifique o status em subscription/3 ou os reason codes retornados por :emqtt.subscribe/3.
• FunctionClauseError em handle_message/3 — adicione uma cláusula genérica; filtros com wildcard entregam mais do que você espera.

O modelo de processos do Elixir torna os clientes MQTT especialmente robustos: uma conexão supervisionada por broker, topics com pattern matching e o trabalho lento fora do handler. Comece com Tortoise311, use emqtt quando precisar de MQTT 5 e deixe uma aba do navegador nos topics enquanto testa.

Perguntas frequentes