MQTT in Elixir mit Tortoise311 und emqtt

Elixir hat zwei solide MQTT-Clients: Tortoise311, einen reinen Elixir-Client für MQTT 3.1.1 mit Handler-Behaviour und automatischen Reconnects, und emqtt, den Erlang-Client des EMQX-Teams, der MQTT 5 spricht. Beide passen nahtlos in einen OTP-Supervision-Tree – ein Broker-Ausfall wird so zu einem Neustart statt zu einer abgestürzten Anwendung.

In diesem Tutorial baust du eine kleine telemetry-App, die Sensor-Topics abonniert, sie per Pattern Matching in Funktionsköpfen auseinanderhält und mit QoS 1 publiziert – zuerst mit Tortoise311, dann mit emqtt für MQTT 5. Alle Beispiele nutzen broker.connectmq.com, du kannst den Traffic also beim Programmieren im Online-MQTT-Client mitverfolgen.

Welche Bibliothek?

Tortoise311 ist ein gepflegter Fork der ursprünglichen Tortoise-Bibliothek; die API ist gleich, nur das Modul heißt anders. Wenn du nicht sicher bist, ob du MQTT 5 brauchst, lies MQTT 5 vs. 3.1.1 – für einfache Telemetrie und Befehle reicht 3.1.1.

Projekt anlegen

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 überspringt die optionale QUIC-Abhängigkeit von emqtt, die sonst eine C-Toolchain und CMake braucht. Führe mix deps.get aus und exportiere die Zugangsdaten deines Brokers, bevor du IEx startest:

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

Eine überwachte Tortoise311-Verbindung

Eine Tortoise311-Verbindung ist einfach ein Kind in deinem Supervision-Tree. Sie besitzt den TCP-Socket, schickt Keep-Alive-Pings, verbindet sich neu und abonniert selbstständig erneut.

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

Das Beispiel nutzt Jason für JSON; füge {:jason, "~> 1.4"} zu deinen Dependencies hinzu oder ersetze es durch einen einfachen String. publish_sync/4 blockiert, bis der Broker die QoS-1-Nachricht bestätigt (PUBACK), und gibt :ok zurück; publish/4 liefert sofort {:ok, ref} und schickt die Bestätigung später als Nachricht an den Aufrufer. Client-IDs sind wichtiger, als sie aussehen; der Client-ID-Generator hilft dir bei einem eindeutigen Schema.

Nachrichten per Pattern Matching verarbeiten

Tortoise311 zerlegt jedes eingehende Topic in eine Liste von Ebenen. Routing wird damit zu ganz normalem Elixir-Pattern-Matching – ohne String-Parsing, ohne 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

Das +-Wildcard der Subscription wird zur Variable device im Funktionskopf. Behalte am Ende eine Catch-all-Klausel: Ein Wildcard-Abo kann Topics liefern, mit denen du nicht gerechnet hast, und eine fehlende Klausel wirft FunctionClauseError im Verbindungsprozess. Wenn Filter neu für dich sind, erklärt MQTT-Topics und Wildcards, wie + und # matchen.

End-to-End testen

Starte die App mit iex -S mix. Du solltest MQTT connection up und zwei Zeilen subscription ... :ok sehen. Dann publizierst du aus einem anderen 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

Und in die andere Richtung, aus IEx:

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

Abonniere testmqtt/elixir/# im Online-Client, um beide Nachrichten ankommen zu sehen. Um die Supervision zu beweisen, beende den Verbindungsprozess mit Process.exit(pid, :kill) (zu finden über Supervisor.which_children(Telemetry.Supervisor)) und beobachte, wie der Supervisor einen neuen startet, der sich von selbst wieder anmeldet.

Elixir-Topics im Browser beobachten

TLS auf Port 8883

Stell den Transport auf Tortoise311.Transport.SSL um und übergib den CA-Store des Systems (OTP 25 oder neuer):

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

Mit verify: :verify_peer prüft Erlangs :ssl sowohl die Zertifikatskette als auch den Hostnamen. „Repariere“ einen Handshake-Fehler in Produktion niemals mit :verify_none; prüfe stattdessen Hostname und CA, wie in MQTT-TLS-Verbindung testen beschrieben.

MQTT 5 mit emqtt

emqtt liefert eingehende Nachrichten an seinen Owner-Prozess. Der idiomatische Wrapper ist deshalb ein GenServer, der den Client mit start_link/1 startet. Da beide Prozesse verlinkt sind, reißt eine verlorene Verbindung den Wrapper mit, und der Supervisor startet beide neu – „let it crash“ als Reconnect-Strategie.

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

Füge Telemetry.Mqtt5 zur Kinderliste hinzu. clean_start: false plus eine Session Expiry von einer Stunde bedeutet, dass der Broker deine Subscriptions und wartende QoS-1-Nachrichten aufbewahrt, während die Node neu startet – siehe Persistent Sessions. Der geteilte Test-Endpoint speichert keine Offline-Sessions; teste diesen Teil daher gegen deine eigene private Broker-Instanz. Schlägt connect/1 fehl, wirft das Pattern Match; der Reason Code verrät den Grund (:bad_username_or_password, :not_authorized usw.), und das MQTT-Reason-Code-Nachschlagewerk erklärt jeden einzelnen.

MQTT in eine größere App einbauen

In einem Phoenix- oder Nerves-Projekt behandelst du die MQTT-Verbindung als Rand des Systems, nicht als Ort für Geschäftslogik:

• Eine Verbindung pro Broker und Node. Jeder Prozess, der publizieren will, ruft Telemetry.Mqtt.publish_temp/2 (oder ein ähnliches Modul) auf; keiner öffnet einen eigenen Socket.
• Nachrichten innerhalb der BEAM verteilen. Broadcaste den dekodierten Payload in handle_message/3 über Phoenix.PubSub oder eine Registry, damit LiveViews, GenServer und Pipelines lokal abonnieren, ohne von MQTT zu wissen.
• Am Rand dekodieren. Parse JSON einmal im Handler und reiche Structs weiter; verwirf fehlerhafte Payloads mit einer Logzeile, statt abzustürzen.
• Topics als Konfiguration. Lege das Topic-Präfix in config/runtime.exs ab, damit Staging und Produktion Code teilen, aber niemals Topics.

Das vereinfacht auch das Testen: Unit-Tests rufen deine Module direkt auf, und ein einziger Integrationstest beweist, dass eine an den echten Broker publizierte Nachricht im lokalen PubSub ankommt.

Häufige Probleme

• Die Verbindung springt ständig zwischen up und down – eine andere Node nutzt dieselbe Client-ID. Prüfe das mit :inet.gethostname() auf jeder Node und lies doppelte Client-ID.
• {:error, :econnrefused} oder kein up-Event – falscher Port oder eine Firewall. Port 1883 ist reines TCP, 8883 ist TLS; eine Verwechslung sieht wie ein Timeout aus.
• Nachrichten kommen mit QoS 0 an, obwohl du mit QoS 1 publiziert hast – die Subscription wurde mit QoS 0 gewährt. Prüfe den Status in subscription/3 oder die Reason Codes, die :emqtt.subscribe/3 zurückgibt.
• FunctionClauseError in handle_message/3 – füge eine Catch-all-Klausel hinzu; Wildcard-Filter liefern mehr, als du erwartest.

Das Prozessmodell von Elixir macht MQTT-Clients ungewöhnlich robust: eine überwachte Verbindung pro Broker, Topics per Pattern Matching und langsame Arbeit außerhalb des Handlers. Starte mit Tortoise311, greif zu emqtt, wenn du MQTT 5 brauchst, und lass beim Testen einen Browser-Tab mit den Topics offen.

Häufig gestellte Fragen