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. Also rendered on hexdocs.

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.11.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()}
    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",
  clean_session: false,          # maintain session across reconnects
  session_store: MqttX.Session.ETSStore  # built-in ETS store
)

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.

Publishing from a server callback (broadcast / fan-out)

To bridge from your application (Phoenix.PubSub, a GenServer, an Oban worker, …) to a connected MQTT client, send a message to the 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"})

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-350457794457489",
  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

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

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>>)

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_handling), server keepalive override
• 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
    qos2_retry_interval: 5000,      # QoS 2 retry timer (ms)
    qos2_max_retries: 3             # QoS 2 max retries before drop
  }],
  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}

Use handle_connect/4 to log protocol version or make version-specific decisions. Existing handle_connect/3 handlers continue to work unchanged.

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
• Empty-buffer fast path: Skips binary concatenation when the TCP buffer is empty (common case)
• 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-optimized retained delivery: O(1) lookup for exact topic subscriptions

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.

These are estimates from architectural analysis and the codec benchmarks below — not end-to-end load tests. 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:

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:

MqttX.Server

Callbacks:

MqttX.Packet.Codec

MqttX.Server.Router

MqttX.Topic

Roadmap

License

Apache-2.0