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)

MQTT for Cellular IoT

For IoT devices on cellular networks (LTE-M, NB-IoT), every byte matters. Data transmission costs money, drains batteries, and increases latency. MQTT combined with Protobuf dramatically outperforms WebSocket with JSON:

Protocol Overhead Comparison

Real-World Payload Example

Sending a sensor reading {temperature: 25.5, humidity: 60, battery: 85}:

Monthly Data Usage (1 device, 1 msg/min)

For fleets of thousands of devices, this translates to significant cost savings on cellular data plans and extended battery life from reduced radio-on time.

MQTT vs WebSocket (Same JSON Payload)

Even when using JSON for both protocols, MQTT still provides significant overhead savings:

Key insight: MQTT's binary protocol has lower framing overhead than WebSocket's text-based frames. For high-frequency IoT messages, this adds up significantly.

Why MqttX?

Existing Elixir/Erlang MQTT libraries have limitations:

• mqtt_packet_map: Erlang-only codec, no server/client, slower encoding
• Tortoise/Tortoise311: Client-only, complex supervision, dated architecture
• emqtt: Erlang-focused, heavy dependencies

MqttX provides a unified, pure Elixir solution with:

• 2.9-4.2x faster encoding than mqtt_packet_map for common packets
• Modern GenServer-based client with exponential backoff reconnection
• Transport-agnostic server that works with ThousandIsland or Ranch
• Clean, composable API designed for IoT and real-time applications
• Zero external dependencies for the core codec

The codec has been tested for interoperability with:

• Zephyr RTOS MQTT client (Nordic nRF9160, nRF52)
• Eclipse Paho clients (C, Python, JavaScript)
• Mosquitto broker
• Standard MQTT test suites

Connecting Nordic Thingy91 / nRF9160 (Zephyr MQTT)

Key Zephyr MQTT settings for MqttX compatibility:

CONFIG_MQTT_KEEPALIVE=30        # Must be < cloud proxy idle timeout (e.g. Fly.io 60s)
CONFIG_MQTT_LIB_TLS=y           # TLS required for production
CONFIG_MQTT_CLEAN_SESSION=1     # Or use MQTT 5.0 session_expiry

Important notes:

• Zephyr's MQTT library supports MQTT 3.1.1 and 5.0
• For MQTT 5.0: server_keep_alive in CONNACK overrides the client's CONFIG_MQTT_KEEPALIVE — set it server-side for fleet control
• For cellular (LTE-M/NB-IoT): use keepalive ≤ 30s to survive cloud proxy idle timeouts (Fly.io, AWS IoT, Azure)
• Protobuf payloads recommended for cellular bandwidth savings

Cloud Deployment with TLS Proxy

When deploying behind a TLS-terminating proxy (Fly.io, AWS NLB, Azure Front Door), ensure:

• Client keepalive < proxy idle timeout (usually 60s)
• Use server_keep_alive transport opt to enforce this server-side for all clients
• Fly.io: internal_port 8883, TLS terminated by Fly proxy

Installation

Add mqttx to your dependencies:

def deps do
  [
    {:mqttx, "~> 0.9.0"},
    # Optional: Pick a transport
    {:thousand_island, "~> 1.4"},  # or {:ranch, "~> 2.2"}
    # Optional: WebSocket transport
    {:bandit, "~> 1.6"},
    {:websock_adapter, "~> 0.5"},
    # 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_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
  }],
  transport: MqttX.Transport.ThousandIsland,
  port: 1883
)

MQTT Client

# Connect with TCP (default)
{:ok, client} = MqttX.Client.connect(
  host: "localhost",
  port: 1883,
  client_id: "my_client",
  username: "user",        # optional
  password: "secret"       # optional
)

# Subscribe
:ok = 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

# Connect with TLS
{:ok, client} = MqttX.Client.connect(
  host: "broker.example.com",
  port: 8883,                    # default SSL port
  client_id: "secure_client",
  transport: :ssl,
  ssl_opts: [
    verify: :verify_peer,
    cacerts: :public_key.cacerts_get(),
    server_name_indication: ~c"broker.example.com"
  ]
)

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)

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
)

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 hundreds of thousands of concurrent devices on a single BEAM node, depending on hardware and workload. Each connection is a lightweight Erlang process (~20KB total with connection state and socket), 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

Codec benchmarks vs mqtt_packet_map (Apple M4 Pro):

See the Performance & Scaling guide for VM tuning, OS tuning, and deployment recommendations.

API Reference

MqttX.Client

Connect Options:

MqttX.Server

Callbacks:

MqttX.Packet.Codec

MqttX.Server.Router

MqttX.Topic

Roadmap

License

Apache-2.0