MqttX

Fast, pure Elixir MQTT 5.0 — client, server, and codec in one package.

• 🚀 High-performance packet codec
• 🖥️ Transport-agnostic server/broker
• 📡 Modern client with automatic reconnection
• 🔌 Pluggable transports (ThousandIsland, Ranch)
• 📦 Optional payload codecs (JSON, Protobuf)

AI coding assistants: see AGENTS.md for the mental model, idiomatic patterns, and a list of mistakes commonly made when integrating MqttX.

Name note: MqttX (this Elixir library) is not affiliated with MQTTX, EMQX's desktop MQTT client tool. The hex package name mqttx is stable — you can depend on it.

Installation

Requires Elixir 1.18+ / OTP 27+ (the JSON payload codec uses the native JSON module; CI covers Elixir 1.18-1.20 on OTP 27-29).

Add mqttx to your dependencies:

def deps do
  [
    {:mqttx, "~> 0.12.0"},
    # Optional: Pick a transport
    {:thousand_island, "~> 1.4"},  # or {:ranch, "~> 2.2"}
    # Optional: WebSocket transport
    {:bandit, "~> 1.6"},
    {:websock_adapter, "~> 0.5 or ~> 0.6"},
    # Optional: Payload codecs
    {:protox, "~> 2.0"}
  ]
end

Quick Start

MQTT Server

Create a handler module:

defmodule MyApp.MqttHandler do
  use MqttX.Server

  @impl true
  def init(_opts) do
    %{subscriptions: %{}}
  end

  @impl true
  def handle_connect(client_id, credentials, connect_info, state) do
    # credentials:  %{username: String.t(), password: String.t()}
    # connect_info: %{protocol_version: 3 | 4 | 5, keep_alive: non_neg_integer(),
    #                 session_present: boolean(), subscriptions: [map()], ...}
    IO.puts("[MQTT] CONNECT #{client_id} v#{connect_info.protocol_version} keepalive=#{connect_info.keep_alive}")
    {:ok, state}
  end

  @impl true
  def handle_publish(topic, payload, opts, state) do
    IO.puts("Received on #{inspect(topic)}: #{payload}")
    {:ok, state}
  end

  @impl true
  def handle_subscribe(topics, state) do
    qos_list = Enum.map(topics, fn t -> t.qos end)
    {:ok, qos_list, state}
  end

  @impl true
  def handle_disconnect(reason, _state) do
    IO.puts("Client disconnected: #{inspect(reason)}")
    :ok
  end
end

Start the server:

{:ok, _pid} = MqttX.Server.start_link(
  MyApp.MqttHandler,
  [],
  transport: MqttX.Transport.ThousandIsland,
  port: 1883
)

MQTT Client

# Connect with TCP (default). connect/1 is asynchronous — `await_connect: true`
# blocks until the session is live so the calls below work inline; long-lived
# clients should instead act on the handler's :connected event.
{:ok, client} = MqttX.Client.connect(
  host: "localhost",
  port: 1883,
  client_id: "my_client",
  await_connect: true,
  username: "user",        # optional
  password: "secret"       # optional
)

# Subscribe (returns {:ok, granted_qos_list})
{:ok, [1]} = MqttX.Client.subscribe(client, "sensors/#", qos: 1)

# Publish
:ok = MqttX.Client.publish(client, "sensors/temp", "25.5")

# Disconnect
:ok = MqttX.Client.disconnect(client)

TLS/SSL Connection

Certificates are verified by default since v0.11.0 — verify_peer against the OS trust store, with SNI and HTTPS-style hostname checking:

{:ok, client} = MqttX.Client.connect(
  host: "broker.example.com",
  port: 8883,                    # default SSL port
  client_id: "secure_client",
  transport: :ssl
)

Options in :ssl_opts are merged over that baseline — supply a private CA with ssl_opts: [cacertfile: "/etc/ssl/private-ca.pem"], or, for a development broker with a self-signed certificate, opt out explicitly with ssl_opts: [verify: :verify_none] (logs a warning on every connect).

Behind an HTTP proxy

Where direct outbound to 1883/8883 is blocked, tunnel through an HTTP CONNECT proxy — works for every transport, and TLS is still negotiated with the broker through the tunnel:

{:ok, client} = MqttX.Client.connect(
  host: "broker.example.com",
  port: 8883,
  client_id: "behind_proxy",
  transport: :ssl,
  proxy: [host: "proxy.corp", port: 3128, auth: {"user", "pass"}]
)

Session Persistence

# Enable session persistence for QoS 1/2 message reliability
{:ok, client} = MqttX.Client.connect(
  host: "localhost",
  client_id: "persistent_client",
  protocol_version: 5,
  clean_session: false,          # maintain session across reconnects
  connect_properties: %{session_expiry_interval: 3600},  # v5 default is 0 = ends at disconnect
  session_store: MqttX.Session.ETSStore  # built-in ETS store
)

The stored record carries expires_at from the negotiated Session Expiry Interval; an expired record is discarded on the next connect and the ETS store sweeps expired records periodically.

Packet Codec (Standalone)

# Encode a packet
packet = %{
  type: :publish,
  topic: "test/topic",
  payload: "hello",
  qos: 0,
  retain: false
}
{:ok, binary} = MqttX.Packet.Codec.encode(4, packet)

# Decode a packet
{:ok, {decoded, rest}} = MqttX.Packet.Codec.decode(4, binary)

Why MQTT, and why MqttX?

MQTT is the right protocol for constrained and cellular deployments, and this library exists because the alternatives in the ecosystem each leave a gap. The reasoning — protocol-overhead comparisons against HTTP and WebSocket, real payload measurements, monthly cellular data budgets, and an honest comparison with the other Elixir/Erlang MQTT libraries — lives in Why MQTT for IoT.

Common Patterns

Receiving messages on the client

Provide a handler module that implements handle_mqtt_event/3. The client calls it on connect, disconnect, for every incoming PUBLISH, and when the broker rejects one of your QoS 1/2 publishes:

defmodule MyApp.MqttClientHandler do
  def handle_mqtt_event(:message, {topic, payload, _packet}, state) do
    IO.puts("Got #{payload} on #{Enum.join(topic, "/")}")
    state
  end

  # Catch-all so other events (:connected, :disconnected, :publish_error)
  # don't raise
  def handle_mqtt_event(_event, _data, state), do: state
end

{:ok, client} = MqttX.Client.connect(
  host: "broker.example.com",
  client_id: "subscriber",
  handler: MyApp.MqttClientHandler,
  handler_state: %{},
  await_connect: true
)

{:ok, _granted} = MqttX.Client.subscribe(client, "sensors/#", qos: 1)

topic arrives as a list of segments (["sensors", "room1", "temp"]), not the original string. The full event list, the payload/packet shapes, and the rules for calling back into the client from a handler are in the Client Guide.

Module-based client (use MqttX)

For a client that owns its callbacks, connection, and supervision in one module:

defmodule MyApp.Sensors do
  use MqttX

  @impl true
  def handle_message(topic, payload, _packet, state) do
    # Safe to publish from inside a callback — callbacks run in this module's
    # own process, not inside the connection
    publish("ack/" <> Enum.join(topic, "/"), payload, qos: 1)
    {:ok, state}
  end
end

# In your supervision tree:
children = [{MyApp.Sensors, host: "broker.example.com", client_id: "sensors-1"}]

Every callback has a default, so implement only what you need. The full callback list and the injected helpers are documented in MqttX.SimpleClient and the Client Guide.

Fan-out between clients (MqttX.Server.deliver/3)

The broker does its own routing when you ask it to. Authorize the publish in handle_publish/4, then hand it to deliver/3: it matches every session's subscriptions on the listener, applies each subscription's options, sends to connected clients and queues QoS 1/2 for persistent sessions that are away:

defmodule MyApp.MqttHandler do
  use MqttX.Server

  def init(_), do: %{client_id: nil}

  def handle_connect(client_id, _creds, _info, state),
    do: {:ok, %{state | client_id: client_id}}

  def handle_publish(topic, payload, opts, state) do
    if MyApp.ACL.may_publish?(state.client_id, topic) do
      MqttX.Server.deliver(topic, payload, Map.put(opts, :publisher, state.client_id))
      {:ok, state}
    else
      {:error, :unauthorized, state}
    end
  end

  def handle_subscribe(topics, state), do: {:ok, Enum.map(topics, & &1.qos), state}
  def handle_disconnect(_reason, _state), do: :ok
end

# From anywhere in your app, by the listener's :name (or port):
MqttX.Server.start_link(MyApp.MqttHandler, [], name: MyApp.Broker, port: 1883)
MqttX.Server.deliver(MyApp.Broker, "device-123/cmd", "reboot", %{qos: 1})

Pushing to one connection from your app

To bridge from your application (Phoenix.PubSub, a GenServer, an Oban worker, …) to one connected MQTT client, send a message to its connection process and return a {:publish, ...} tuple from handle_info/2:

defmodule MyApp.MqttHandler do
  use MqttX.Server

  def init(_), do: %{}

  def handle_connect(client_id, _creds, _info, state) do
    Phoenix.PubSub.subscribe(MyApp.PubSub, "client:#{client_id}")
    {:ok, state}
  end

  def handle_publish(_topic, _payload, _opts, state), do: {:ok, state}
  def handle_subscribe(topics, state), do: {:ok, Enum.map(topics, & &1.qos), state}
  def handle_disconnect(_reason, _state), do: :ok

  def handle_info({:downlink, topic, payload}, state) do
    {:publish, topic, payload, %{qos: 1, retain: false}, state}
  end
end

Then anywhere in your app:

Phoenix.PubSub.broadcast(MyApp.PubSub, "client:device-123",
  {:downlink, "device-123/cmd", "reboot"})

When the topic matches one of the client's subscriptions, the delivery honors that subscription's options: QoS is capped at the granted QoS, retain is forwarded only for Retain-As-Published subscriptions, and subscription identifiers are attached. Pass publisher: originating_client_id in the opts to honor no_local. A publish matching no subscription is sent exactly as given.

MQTT 5.0 persistent sessions (resume after disconnect)

In MQTT 5.0 the client tells the broker how long to keep its session via :session_expiry_interval and resumes by reconnecting with the same client_id and clean_session: false:

{:ok, client} = MqttX.Client.connect(
  host: "broker.example.com",
  client_id: "device-imei-490154203237518",
  protocol_version: 5,
  clean_session: false,
  connect_properties: %{session_expiry_interval: 3600},
  session_store: MqttX.Session.ETSStore
)

A spec-compliant broker queues QoS 1/2 messages while the client is offline (up to 1 hour in this example) and replays them on reconnect. Note that MqttX's own broker does not implement offline queueing — see MqttX.Server if you are running MqttX as the broker.

Common Pitfalls

• connect/1 does not wait for the broker. It returns as soon as the client process starts, so a subscribe/publish issued immediately after gets {:error, :not_connected}. Subscribe once the session is live, or pass await_connect: true. The async default lets a client start before its broker is reachable and retry with backoff.
• clean_session: false does nothing without :session_store. The flag tells the broker to keep state — but for the client to resume QoS 1/2 in-flight on reconnect, you must also pass a :session_store module.
• Server's :server_keep_alive overrides the client. When set in transport_opts, MQTT 5.0 clients use the broker's value regardless of what they sent in CONNECT. Useful for surviving cloud-proxy idle timeouts.
• :max_packet_size is enforced both ways. A client that sends a packet larger than the broker's limit gets DISCONNECT 0x95; the reverse is also true if the client advertises one in CONNECT properties.
• Topic wildcards in PUBLISH are illegal. + and # are subscribe-only — use MqttX.Topic.validate_publish/1 on untrusted input.
• $SYS/... topics need explicit subscription. Per MQTT §4.7.2, # and +/... subscribers do not receive $-prefixed topics; subscribe to $SYS/# directly.

Transport Adapters

MqttX supports pluggable transports:

ThousandIsland (Recommended)

MqttX.Server.start_link(
  MyHandler,
  [],
  transport: MqttX.Transport.ThousandIsland,
  port: 1883
)

Ranch

MqttX.Server.start_link(
  MyHandler,
  [],
  transport: MqttX.Transport.Ranch,
  port: 1883,
  # :ranch_tcp (default) or :ranch_ssl — note this option was named
  # `:transport` before v0.11.0, which collided with the adapter selector above
  ranch_transport: :ranch_tcp
)

WebSocket

# ws://
MqttX.Server.start_link(
  MyHandler,
  [],
  transport: MqttX.Transport.WebSocket,
  port: 8083
)

# wss://
MqttX.Server.start_link(
  MyHandler,
  [],
  transport: MqttX.Transport.WebSocket,
  port: 8084,
  scheme: :https,
  transport_options: [certfile: "/etc/mqttx/cert.pem", keyfile: "/etc/mqttx/key.pem"]
)

Payload Codecs

Built-in payload codecs for message encoding/decoding:

JSON (Erlang/OTP 27+)

Uses the built-in Erlang JSON module:

{:ok, json} = MqttX.Payload.JSON.encode(%{temp: 25.5})
{:ok, data} = MqttX.Payload.JSON.decode(json)

Protobuf

{:ok, binary} = MqttX.Payload.Protobuf.encode(my_proto_struct)
{:ok, struct} = MqttX.Payload.Protobuf.decode(binary, MyProto.Message)

Raw (Pass-through)

{:ok, binary} = MqttX.Payload.Raw.encode(<<1, 2, 3>>)
{:ok, binary} = MqttX.Payload.Raw.decode(<<1, 2, 3>>)

Publishing with a codec

Pass the term and the codec; the payload is encoded and the codec's MQTT 5.0 payload_format_indicator / content_type go out with it:

MqttX.Client.publish(client, "sensors/room1", %{temp: 21.5},
  qos: 1,
  codec: MqttX.Payload.JSON
)

See the Codec guide.

Topic Routing

The server includes a topic router with wildcard support:

alias MqttX.Server.Router

router = Router.new()
router = Router.subscribe(router, "sensors/+/temp", client_ref, qos: 1)
router = Router.subscribe(router, "alerts/#", client_ref, qos: 0)

# Find matching subscriptions
matches = Router.match(router, "sensors/room1/temp")
# => [{client_ref, %{qos: 1}}]

Protocol Support

• MQTT 3.1 (protocol version 3)
• MQTT 3.1.1 (protocol version 4)
• MQTT 5.0 (protocol version 5)

All 15 packet types are supported:

• CONNECT, CONNACK
• PUBLISH, PUBACK, PUBREC, PUBREL, PUBCOMP
• SUBSCRIBE, SUBACK
• UNSUBSCRIBE, UNSUBACK
• PINGREQ, PINGRESP
• DISCONNECT
• AUTH (MQTT 5.0)

Compliance

Fully compliant with MQTT 3.1, 3.1.1, and 5.0 specifications:

• Server: CONNACK capability properties, protocol ordering enforcement, topic alias validation, MQTT 5.0 property forwarding, subscription options (no_local, retain_as_published, retain_handling, subscription identifiers), server keepalive override, enhanced AUTH (exchange before CONNACK, re-authentication)
• Client: server_keep_alive override, assigned_client_identifier, maximum_packet_size enforcement, server_reference handling, enhanced AUTH (multi-step), flow control (receive_maximum)

Validated against Mosquitto (104 automated protocol tests across TCP and WebSocket) and EMQX Cloud (49 interop tests covering all QoS levels, properties, session persistence, and subscription options).

MQTT 5.0 Server Features

Server CONNACK properties (sent to MQTT 5.0 clients):

transport_opts configuration:

MqttX.Server.start_link(
  MyHandler,
  [transport_opts: %{
    server_keep_alive: 30,           # override client keepalive (v5)
    topic_alias_maximum: 100,        # max topic aliases
    receive_maximum: 65535,          # max inflight QoS>0
    max_packet_size: 256_000,        # reject oversized packets
    max_outbound_queue: 10_000,      # publishes queued per session (window + offline)
    max_session_expiry_interval: 86_400  # longest a session outlives its connection
  }],
  transport: MqttX.Transport.ThousandIsland,
  port: 1883
)

handle_connect callback:

The optional 4-arity handle_connect/4 receives connection metadata separately from credentials:

# credentials (both arities):
%{username: "device_imei", password: "secret"}

# connect_info (4-arity only):
%{protocol_version: 5, keep_alive: 50, authentication_method: nil,
  session_present: true, subscriptions: [%{topic: "sensors/#", qos: 1, ...}]}

Use handle_connect/4 to log the protocol version, make version-specific decisions, or see which session the client is resuming (session_present and the subscriptions coming back with it). Existing handle_connect/3 handlers continue to work unchanged.

Sessions: the broker keeps Session State per listener, in ETS. A client connecting with Clean Start 0 and a Session Expiry Interval (or CleanSession 0 on 3.1.1) gets session_present: true on reconnect, its subscriptions back, unacknowledged QoS 1/2 messages re-sent, and whatever deliver/3 queued while it was away. Sessions are kept for at most max_session_expiry_interval (default one day) and live in memory on one node.

Performance

Architected to scale from tens of thousands to roughly a million concurrent devices on a single BEAM node, depending on hardware and workload. Each connection is a lightweight Erlang process (~20KB of BEAM state plus ~4-8KB of kernel socket buffers), and the hot paths are optimized for high message throughput:

• Trie-based topic router: O(L+K) matching where L = topic depth, K = matching subscriptions — independent of total subscription count
• iodata encoding: Socket sends use iodata directly, avoiding binary copies on every packet
• Zero-copy binary references: Decoder returns sub-binaries for payload and topic
• Declared-length reassembly: A packet spanning many TCP reads is collected without copying and assembled once — linear, where per-read concatenation was quadratic
• Cached callback dispatch: function_exported? computed once at connection init, not per message
• Direct inflight counter: O(1) flow control check instead of scanning pending_acks
• ETS match-spec retained delivery: Exact filters are one lookup; wildcard filters traverse only the key range their literal prefix bounds

Capacity depends on hardware, so these figures are anchored to instance sizes rather than given as a single ceiling. Devices are the practical targets from the capacity planning method, which reserves headroom for the runtime, ETS, and reconnect storms:

Beyond roughly 500K connections per node the limit stops being RAM and becomes kernel socket memory, file descriptors, and contention on shared ETS tables — none of which improve with more cores — so horizontal scaling usually beats a larger instance. Message-rate figures assume small QoS 0 payloads, plaintext TCP, and a handler doing negligible work.

The device counts are estimates from architectural analysis — nobody has load-tested a million connections here. Per-message cost, on the other hand, is measured end to end (below). See the Performance & Scaling guide for the sizing formula, per-vCPU throughput, and the caveats behind each number.

Codec benchmarks vs mqtt_packet_map — measured, on an Apple M4 Pro:

End-to-end, MqttX client ↔ MqttX broker over loopback — measured on the same machine with mix run bench/e2e.exs (one publisher; fan-out is one PUBLISH forwarded to 50 subscribed connections):

WebSocket figures, the receive-buffer and retained-store measurements, and how to run the suites yourself are in Measured Throughput; mix test --include load runs a many-client completeness check.

The performance guide also covers VM tuning (+P/+Q limits, which you must raise past 65K connections), OS tuning, and multi-node deployment.

Guides

Getting Started · Why MQTT for IoT · Client · Server / Broker · Packet Codec · Telemetry · Performance & Scaling

API Reference

MqttX.Client

Connect Options:

Unacknowledged QoS 1/2 publishes are held until the broker acknowledges them and re-sent only when a session is resumed after a reconnect (MQTT §4.4) — never on a live connection. :retry_interval is deprecated and ignored since 0.12.0.

MqttX.Server

Callbacks:

MqttX.Packet.Codec

MqttX.Server.Router

MqttX.Topic

Roadmap

License

Apache-2.0