MQTT in Elixir with Tortoise311 and emqtt

Elixir has two solid MQTT clients: Tortoise311, a pure-Elixir client for MQTT 3.1.1 with a handler behaviour and automatic reconnects, and emqtt, the Erlang client from the EMQX team that speaks MQTT 5. Both fit naturally into an OTP supervision tree, so a broker outage becomes a restart, not a crashed application.

This tutorial builds a small telemetry app that subscribes to sensor topics, pattern-matches them in function heads and publishes with QoS 1 — first with Tortoise311, then with emqtt for MQTT 5. Every example targets broker.connectmq.com, so you can watch the traffic in the online MQTT client while you code.

Which library?

Tortoise311 is a maintained fork of the original Tortoise library; the API is the same with the module renamed. If you are unsure whether you need MQTT 5, read MQTT 5 vs 3.1.1 — for plain telemetry and commands, 3.1.1 is fine.

Project setup

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 skips emqtt's optional QUIC dependency, which otherwise needs a C toolchain and CMake. Run mix deps.get, and export the credentials of your broker before starting IEx:

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

A supervised Tortoise311 connection

A Tortoise311 connection is just a child in your supervision tree. It owns the TCP socket, sends keep-alive pings, reconnects and re-subscribes on its own.

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

The example uses Jason for JSON; add {:jason, "~> 1.4"} to your deps or replace it with a plain string. publish_sync/4 blocks until the broker acknowledges the QoS 1 message (PUBACK) and returns :ok; publish/4 returns {:ok, ref} immediately and sends the acknowledgement to the caller as a message later. Client IDs matter more than they look; the client ID generator helps you pick a unique scheme.

Handling messages with pattern matching

Tortoise311 splits every incoming topic into a list of levels, so routing becomes ordinary Elixir pattern matching — no string parsing, no 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

The + wildcard in the subscription becomes the device variable in the function head. Keep a catch-all clause at the end: a wildcard subscription can deliver topics you did not anticipate, and a missing clause raises FunctionClauseError inside the connection process. If you are new to filters, MQTT topics and wildcards explains how + and # match.

Testing it end to end

Start the app with iex -S mix. You should see MQTT connection up and two subscription ... :ok lines. Then publish from another 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

And the other direction, from IEx:

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

Subscribe to testmqtt/elixir/# in the online client to see both messages arrive. To prove the supervision story, kill the connection process with Process.exit(pid, :kill) (find it with Supervisor.which_children(Telemetry.Supervisor)) and watch the supervisor start a new one that re-subscribes by itself.

Watch the Elixir topics in your browser

TLS on port 8883

Switch the transport to Tortoise311.Transport.SSL and give it the system CA store (OTP 25 or newer):

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

With verify: :verify_peer, Erlang's :ssl checks both the certificate chain and the hostname. Never "fix" a handshake error by switching to :verify_none in production; check the hostname and CA instead, as described in testing an MQTT TLS connection.

MQTT 5 with emqtt

emqtt delivers incoming messages to its owner process, so the idiomatic wrapper is a GenServer that starts the client with start_link/1. Because the two processes are linked, a lost connection crashes the wrapper and the supervisor restarts both — "let it crash" used as a reconnect strategy.

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

Add Telemetry.Mqtt5 to the children list. clean_start: false plus a session expiry of one hour means the broker keeps your subscriptions and queued QoS 1 messages while the node restarts — see persistent sessions. The shared test endpoint does not keep offline sessions, so try this part against your own private broker instance. If connect/1 fails, the pattern match raises; the reason code tells you why (:bad_username_or_password, :not_authorized and so on), and the MQTT reason code lookup explains each one.

Fitting MQTT into a larger app

In a Phoenix or Nerves project, treat the MQTT connection as an edge of the system rather than the place where business logic lives:

• One connection per broker per node. Every process that wants to publish calls Telemetry.Mqtt.publish_temp/2 (or a similar module); none of them opens its own socket.
• Fan messages out inside the BEAM. In handle_message/3, broadcast the decoded payload on Phoenix.PubSub or a Registry, so LiveViews, GenServers and pipelines subscribe locally without knowing MQTT exists.
• Decode at the edge. Parse JSON once in the handler and pass structs onwards; reject malformed payloads with a log line instead of crashing.
• Make topics configuration. Keep the topic prefix in config/runtime.exs, so staging and production can share code but never share topics.

This also makes testing easier: unit tests call your modules directly, while one integration test proves that a message published to the real broker reaches the local PubSub.

Common problems

• Connection keeps flapping between up and down — another node uses the same client ID. Check with :inet.gethostname() on each node and see duplicate client ID.
• {:error, :econnrefused} or no up event — wrong port, or a firewall. Port 1883 is plain TCP, 8883 is TLS; mixing them up looks like a timeout.
• Messages arrive at QoS 0 although you published at QoS 1 — the subscription was granted at QoS 0. Inspect the status in subscription/3 or the reason codes returned by :emqtt.subscribe/3.
• FunctionClauseError in handle_message/3 — add a catch-all clause; wildcard filters deliver more than you expect.

Elixir's process model makes MQTT clients unusually robust: one supervised connection per broker, pattern-matched topics, and slow work pushed out of the handler. Start with Tortoise311, reach for emqtt when you need MQTT 5, and keep a browser tab on the topics while you test.

Frequently asked questions