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

Installation

Add mqttx to your dependencies:

def deps do
  [
    {:mqttx, "~> 0.6.0"},
    # Optional: Pick a transport
    {:thousand_island, "~> 1.4"},  # or {:ranch, "~> 2.2"}
    # 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, state) do
    IO.puts("Client connected: #{client_id}")
    {: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)
{: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
)

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)

Performance

Designed to scale to 10k-100k+ concurrent devices on a single BEAM node. Each connection is a lightweight Erlang process (~2-5KB), 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

Next Up

Future

License

Apache-2.0