# 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`](https://hexdocs.pm/mqttx/agents.html) > 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](https://mqttx.app), 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: ```elixir 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: ```elixir 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: ```elixir {:ok, _pid} = MqttX.Server.start_link( MyApp.MqttHandler, [], transport: MqttX.Transport.ThousandIsland, port: 1883 ) ``` ### MQTT Client ```elixir # 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: ```elixir {: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: ```elixir {: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 ```elixir # 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) ```elixir # 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](https://hexdocs.pm/mqttx/why-mqtt-for-iot.html). ## 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: ```elixir 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](https://hexdocs.pm/mqttx/client.html#receiving-messages)**. ### Module-based client (`use MqttX`) For a client that owns its callbacks, connection, and supervision in one module: ```elixir 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](https://hexdocs.pm/mqttx/client.html#module-based-clients-use-mqttx)**. ### 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`: ```elixir 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: ```elixir 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`: ```elixir {: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) ```elixir MqttX.Server.start_link( MyHandler, [], transport: MqttX.Transport.ThousandIsland, port: 1883 ) ``` ### Ranch ```elixir 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 ```elixir 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: ```elixir {:ok, json} = MqttX.Payload.JSON.encode(%{temp: 25.5}) {:ok, data} = MqttX.Payload.JSON.decode(json) ``` ### Protobuf ```elixir {:ok, binary} = MqttX.Payload.Protobuf.encode(my_proto_struct) {:ok, struct} = MqttX.Payload.Protobuf.decode(binary, MyProto.Message) ``` ### Raw (Pass-through) ```elixir {: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: ```elixir 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): | Property | Default | Configurable | |----------|---------|--------------| | `shared_subscription_available` | `1` | No | | `topic_alias_maximum` | `100` | Yes (`transport_opts`) | | `receive_maximum` | `65535` | Yes (`transport_opts`) | | `retain_available` | `1` | No | | `wildcard_subscription_available` | `1` | No | | `subscription_identifier_available` | `0` | No | | `server_keep_alive` | Not sent | Yes (`transport_opts`) | | `maximum_packet_size` | Not sent | Yes (`transport_opts`) | **`transport_opts` configuration:** ```elixir 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: ```elixir # 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](https://hexdocs.pm/mqttx/performance.html#capacity-planning) method, which reserves headroom for the runtime, ETS, and reconnect storms: | Instance | Idle-ish devices (~1 msg/min) | Chatty devices (1 msg/sec) | Binding constraint | |----------|------------------------------|----------------------------|--------------------| | 1 vCPU / 2 GB | ~50,000 | ~15,000 | RAM / CPU | | 4 vCPU / 16 GB | ~400,000 | ~60,000 | RAM, fds, kernel memory | | 16 vCPU / 128 GB | ~1,000,000 | ~160,000 | ETS contention, accept rate | 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](https://hexdocs.pm/mqttx/performance.html) 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: | Operation | MqttX | mqtt_packet_map | Result | |-----------|-------|-----------------|--------| | PUBLISH encode | 5.05M ips | 1.72M ips | **2.9x faster** | | SUBSCRIBE encode | 3.42M ips | 0.82M ips | **4.2x faster** | | PUBLISH decode | 2.36M ips | 2.25M ips | ~same | The performance guide also covers [VM tuning](https://hexdocs.pm/mqttx/performance.html#vm-tuning) (`+P`/`+Q` limits, which you *must* raise past 65K connections), OS tuning, and multi-node deployment. ## Guides [Getting Started](https://hexdocs.pm/mqttx/getting-started.html) · [Why MQTT for IoT](https://hexdocs.pm/mqttx/why-mqtt-for-iot.html) · [Client](https://hexdocs.pm/mqttx/client.html) · [Server / Broker](https://hexdocs.pm/mqttx/server.html) · [Packet Codec](https://hexdocs.pm/mqttx/codec.html) · [Telemetry](https://hexdocs.pm/mqttx/telemetry.html) · [Performance & Scaling](https://hexdocs.pm/mqttx/performance.html) ## API Reference ### MqttX.Client | Function | Description | |----------|-------------| | `connect(opts)` | Connect to an MQTT broker | | `connect_supervised(opts)` | Connect under `MqttX.Client.Supervisor` with crash recovery | | `list()` | List all registered client connections | | `whereis(client_id)` | Look up a connection by client_id | | `publish(client, topic, payload, opts \\ [])` | Publish a message. Options: `:qos` (0-2), `:retain` (boolean), `:properties` (MQTT 5.0) | | `subscribe(client, topics, opts \\ [])` | Subscribe to topics. Options: `:qos` (0-2), `:no_local`, `:retain_as_published`, `:retain_handling`, `:properties` (MQTT 5.0) | | `unsubscribe(client, topics)` | Unsubscribe from topics | | `disconnect(client)` | Disconnect from the broker | | `connected?(client)` | Check if client is connected | **Connect Options:** | Option | Description | Default | |--------|-------------|---------| | `:host` | Broker hostname | *required* | | `:port` | Broker port | `1883` / `8883` / `8083` / `8084` | | `:client_id` | Client identifier | *required* | | `:username` | Authentication username | `nil` | | `:password` | Authentication password | `nil` | | `:clean_session` | Start fresh session | `true` | | `:keepalive` | Keep-alive interval (seconds) | `60` | | `:await_connect` | Block until the first CONNACK resolves (see [Common Pitfalls](#common-pitfalls)) | `false` | | `:protocol_version` | MQTT protocol level: `3`, `4` (3.1.1) or `5` | `5` | | `:transport` | `:tcp`, `:ssl`, `:ws`, or `:wss` | `:tcp` | | `:ssl_opts` | SSL options, merged **over** the secure baseline (see [TLS/SSL](https://hexdocs.pm/mqttx/client.html#tls-ssl)) | `[]` | | `:ws_path` | WebSocket path for `:ws` or `:wss` | `"/mqtt"` | | `:proxy` | HTTP CONNECT proxy, e.g. `[host: "proxy.corp", port: 3128, auth: {"u", "p"}]` | `nil` | | `:retry_interval` | QoS retry interval (ms) | `5000` | | `:max_inflight` | Max pending QoS 1/2 messages | `100` | | `:max_packet_size` | Reject inbound packets declaring more than this (`:infinity` disables) | `1 MiB` | | `:will_topic` / `:will_payload` / `:will_qos` / `:will_retain` / `:will_properties` | Last Will & Testament | `nil` / `""` / `0` / `false` / `%{}` | | `:connect_properties` | MQTT 5.0 CONNECT properties (e.g. `%{session_expiry_interval: 3600}`) | `%{}` | | `:session_store` | Session store module | `nil` | | `:handler` | Callback module for messages | `nil` | | `:handler_state` | Initial handler state | `nil` | ### MqttX.Server | Function | Description | |----------|-------------| | `start_link(handler, handler_opts, opts)` | Start an MQTT server. Options: `:transport`, `:port`, `:rate_limit`, `:ip`. Protocol options go in `handler_opts` under `:transport_opts` | **Callbacks:** | Callback | Description | |----------|-------------| | `init(opts)` | Initialize handler state | | `handle_connect(client_id, credentials, state)` | Handle client connection. Return `{:ok, state}` or `{:error, reason_code, state}` | | `handle_connect(client_id, credentials, connect_info, state)` | *(optional)* Same as above with connection metadata (`protocol_version`, `keep_alive`). Takes precedence over 3-arity when defined | | `handle_publish(topic, payload, opts, state)` | Handle incoming PUBLISH. Return `{:ok, state}` | | `handle_subscribe(topics, state)` | Handle SUBSCRIBE. Return `{:ok, granted_qos_list, state}` | | `handle_unsubscribe(topics, state)` | Handle UNSUBSCRIBE. Return `{:ok, state}` | | `handle_disconnect(reason, state)` | Handle client disconnection. Return `:ok` | | `handle_session_expired(client_id, state)` | *(optional)* MQTT 5.0 session expiry elapsed after disconnect. Return `:ok` | | `handle_info(message, state)` | Handle custom messages. Return `{:ok, state}`, `{:publish, topic, payload, state}`, `{:publish, topic, payload, opts, state}`, `{:disconnect, reason_code, state}`, `{:disconnect, reason_code, properties, state}`, or `{:stop, reason, state}` | ### MqttX.Packet.Codec | Function | Description | |----------|-------------| | `encode(version, packet)` | Encode a packet to binary. Returns `{:ok, binary}` | | `decode(version, binary)` | Decode a packet from binary. Returns `{:ok, {packet, rest}}` or `{:error, reason}` | | `encode_iodata(version, packet)` | Encode to iodata (more efficient). Returns `{:ok, iodata}` | ### MqttX.Server.Router | Function | Description | |----------|-------------| | `new()` | Create a new empty router | | `subscribe(router, filter, client, opts)` | Add a subscription. Options: `:qos` | | `unsubscribe(router, filter, client)` | Remove a subscription | | `unsubscribe_all(router, client)` | Remove all subscriptions for a client | | `match(router, topic)` | Find matching subscriptions. Returns `[{client, opts}]` | ### MqttX.Topic | Function | Description | |----------|-------------| | `validate(topic)` | Validate and normalize a topic. Returns `{:ok, normalized}` or `{:error, :invalid_topic}` | | `validate_publish(topic)` | Validate topic for publishing (no wildcards) | | `matches?(filter, topic)` | Check if a filter matches a topic | | `normalize(topic)` | Normalize topic to list format | | `flatten(normalized)` | Convert normalized topic back to binary string | | `wildcard?(topic)` | Check if topic contains wildcards | ## Roadmap | Feature | Status | Description | |---------|--------|-------------| | **Full MQTT 5.0 Compliance** | Done | Complete server and client compliance — all CONNACK properties, enhanced AUTH, flow control, server redirect | | **WebSocket Transport** | Done | MQTT over WebSocket via Bandit (`ws://` and `wss://`) | | **Broker Validation** | Done | 104 Mosquitto tests (TCP + WebSocket) + 49 EMQX Cloud interop tests | | **Clustering** | Planned | Distributed router across Erlang nodes via `pg` | | **Session Persistence (Server)** | Planned | Server-side session persistence (currently client-only) | | **MQTT 5.0 Enhanced Auth** | Partial | AUTH exchange and re-authentication implemented; no built-in SCRAM/external providers | | **Telemetry Docs** | Done | See the [Telemetry guide](https://hexdocs.pm/mqttx/telemetry.html) | | **Property-based Tests** | Done | StreamData round-trips + decode/encode fuzzing of the codec | | **End-to-end Load Tests** | Planned | Benchee-based throughput validation under realistic workloads | ## License Apache-2.0