Server Guide

MqttX.Server is a behaviour for building MQTT brokers. You implement handler callbacks; MqttX handles the protocol, transport, and routing.

Handler Callbacks

defmodule MyApp.MqttHandler do
  use MqttX.Server

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

  @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_unsubscribe(_topics, state), do: {:ok, state}

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

  @impl true
  def handle_info(message, state) do
    # Handle custom messages (e.g., Phoenix.PubSub broadcasts)
    {:ok, state}
  end
end

Callback Summary

Use handle_info/2 with {:publish, topic, payload, state} to push messages to connected clients from external events (e.g., PubSub).

Any callback that returns {:disconnect, reason_code, state} or {:disconnect, reason_code, properties, state} will send an MQTT 5.0 DISCONNECT packet to the client and close the connection.

Transport Adapters

ThousandIsland (Recommended)

# mix.exs: {:thousand_island, "~> 1.4"}

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

Ranch

# mix.exs: {:ranch, "~> 2.2"}

{:ok, _pid} = MqttX.Server.start_link(
  MyApp.MqttHandler,
  [],
  transport: MqttX.Transport.Ranch,
  port: 1883,
  # :ranch_tcp (default) or :ranch_ssl. Named `:transport` before v0.11.0,
  # which collided with the adapter selector above.
  ranch_transport: :ranch_tcp
)

WebSocket

# mix.exs: {:bandit, "~> 1.6"}, {:websock_adapter, "~> 0.5 or ~> 0.6"}

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

All transports support TCP and TLS. See MqttX.Transport for implementing custom adapters.

Topic Routing

The built-in router uses a trie data structure for efficient topic matching — O(L+K) where L is the topic depth and K is matching subscriptions. It supports MQTT wildcard subscriptions (+ single-level, # multi-level):

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)

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

Shared Subscriptions

Distribute messages across a group of subscribers with round-robin load balancing:

router = Router.subscribe(router, "$share/workers/jobs/#", worker1, qos: 1)
router = Router.subscribe(router, "$share/workers/jobs/#", worker2, qos: 1)

# Messages to "jobs/process" alternate between worker1 and worker2
{matches, router} = Router.match_and_advance(router, "jobs/process")

Rate Limiting

MqttX supports per-client connection and message rate limiting. Configure it via the :rate_limit option:

MqttX.Server.start_link(MyApp.MqttHandler, [],
  transport: MqttX.Transport.ThousandIsland,
  port: 1883,
  rate_limit: [
    max_connections: 100,    # new connections per second
    max_messages: 1000,      # messages per client per second
    interval: 1000           # window size in ms (default)
  ]
)

When a client exceeds the message rate limit:

• QoS 0: Messages are silently dropped (per MQTT spec)
• QoS 1+: PUBACK is sent with reason code 0x96 (message_rate_too_high)

When the connection rate limit is exceeded, new connections are immediately closed.

Rate limiting uses ETS with atomic update_counter operations, making it lock-free and safe for concurrent access from multiple transport handler processes.

Retained Messages

The server automatically stores retained messages in ETS and delivers them to new subscribers. Publish with an empty payload to clear a retained message.

Keepalive Timeout

The server enforces MQTT keepalive as defined in the spec: if no packet is received from a client within 1.5x the keep_alive interval (set in the CONNECT packet), the server disconnects the client and publishes any will message.

The keepalive timer resets on every received packet (not just PINGREQ). Clients that set keep_alive: 0 are exempt from keepalive enforcement — but not from the idle ceiling below.

When a keepalive timeout fires, your handle_disconnect/2 callback receives :keepalive_timeout as the reason.

Idle ceiling

Because MQTT keepalive does not apply when a client connects with keep_alive: 0, the server additionally closes any socket idle for longer than :max_idle_timeout (default 15 minutes), regardless of the negotiated Keep Alive. The timer resets on any inbound data, so it never interferes with a client that is honoring its keepalive. handle_disconnect/2 receives :idle_timeout.

transport_opts: %{max_idle_timeout: 900_000}   # :infinity disables

A separate :connect_timeout (default 10 s) closes sockets that never complete CONNECT.

Will Messages

Will messages from the CONNECT packet are published automatically when a client disconnects without sending DISCONNECT.

Will Delay Interval (MQTT 5.0)

MQTT 5.0 clients can set will_delay_interval in the will properties to delay publication of the will message:

• will_delay_interval: 0 (or MQTT 3.1.1) — will published immediately (default)
• will_delay_interval: N — will published N seconds after ungraceful disconnect

This allows a grace period for clients to reconnect before their "last will" is broadcast.

Session State

The server operates in clean-session mode for every protocol version: session_present is always false in CONNACK, and session state (subscriptions, queued messages, in-flight QoS 1/2) is not persisted across reconnections. A client that connects with clean_session: false therefore gains nothing against this broker, though it works as expected against EMQX, Mosquitto, and other brokers. If your application needs session resumption, implement it at the handler level using handle_connect/3 and a session store.

Session Expiry (MQTT 5.0)

MQTT 5.0 clients can set session_expiry_interval in the CONNECT properties. After the client disconnects, the server waits the specified interval then calls your handle_session_expired/2 callback.

The timer is supervised and cancelled if the same client reconnects within the window, so expiry never fires for a session that resumed in time. A client may also revise the interval in its DISCONNECT packet (§3.14.2.2.2) — commonly session_expiry_interval: 0 meaning "drop my session now".

@impl true
def handle_session_expired(client_id, state) do
  # Clean up stored subscriptions, queued messages, etc.
  MyApp.SessionStore.delete(client_id)
  :ok
end

Server-Initiated Disconnect

Kick a client from the server with an MQTT 5.0 reason code:

# From outside the handler (e.g., admin action)
MqttX.Server.disconnect(transport_pid, 0x98)
MqttX.Server.disconnect(transport_pid, 0x89, %{reason_string: "Session taken over"})

Or return {:disconnect, reason_code, state} from any handler callback:

@impl true
def handle_publish(topic, _payload, _opts, state) do
  if forbidden?(topic) do
    {:disconnect, 0x98, state}  # Use assigned client identifier
  else
    {:ok, state}
  end
end

DISCONNECT packets are only sent for MQTT 5.0 connections. For MQTT 3.1.1, the connection is simply closed.

MQTT 5.0 Protocol Features

The server automatically handles these MQTT 5.0 features when clients connect with protocol version 5:

Topic Aliases

Clients can use topic aliases to reduce bandwidth by replacing repeated topic strings with short integer aliases. The server:

• Advertises topic_alias_maximum: 100 in CONNACK (configurable via transport_opts)
• Resolves incoming topic aliases: first publish with alias + topic stores the mapping, subsequent publishes with alias only use the stored topic
• No application code needed — alias resolution is transparent to your handler callbacks

Flow Control (Receive Maximum)

The server enforces receive_maximum for incoming QoS 1 and QoS 2 messages:

• Advertises receive_maximum in CONNACK (default: 65535, configurable via transport_opts)
• Tracks in-flight QoS 2 messages (between PUBREC and PUBCOMP)
• Rejects excess QoS 2 publishes with PUBREC reason code 0x93 (Receive Maximum exceeded)

Maximum Packet Size

Incoming packets are capped at 1 MiB by default (since v0.11.0) and the limit is enforced on the declared remaining length before the body is buffered, so an unauthenticated peer cannot force a large allocation. Override or disable it with :infinity:

# NOTE: protocol options live in `transport_opts` inside the *handler options*
# (2nd argument), not the transport options (3rd argument).
MqttX.Server.start_link(
  MyApp.MqttHandler,
  [transport_opts: %{max_packet_size: 1_048_576}],
  transport: MqttX.Transport.ThousandIsland,
  port: 1883
)

• Server sends DISCONNECT with reason code 0x95 (Packet too large) for oversized incoming packets
• Retained storage is likewise bounded by :max_retained_messages (default 100_000 distinct topics)
• Outgoing publishes exceeding the client's advertised maximum_packet_size are silently dropped
• Server advertises its maximum_packet_size in CONNACK when configured

QoS 2 Retransmission

The server automatically retries stale QoS 2 handshake messages:

• Re-sends PUBREC if no PUBREL received within the retry interval (default: 5 seconds)
• Re-sends PUBLISH with dup: true for outgoing QoS 2 awaiting PUBREC
• Re-sends PUBREL for outgoing QoS 2 awaiting PUBCOMP
• Drops entries after max retries (default: 3)
• Handles DUP incoming PUBLISH by re-sending PUBREC without re-storing

Shared Subscription Availability

The server advertises shared_subscription_available: 1 in CONNACK. See Shared Subscriptions under Topic Routing for usage.

CONNACK Properties

For MQTT 5.0 connections, the server automatically includes these properties in CONNACK:

Publish Properties

The opts map passed to handle_publish/4 includes a :properties key containing any MQTT 5.0 publish properties sent by the client (e.g., user_properties, content_type, correlation_data, response_topic, payload_format_indicator, message_expiry_interval). These properties are also forwarded when the server sends outgoing PUBLISH messages via handle_info/2.

Subscription Options

The server supports MQTT 5.0 subscription options:

• no_local: When set to true, messages published by a client are not delivered back to that client's own matching subscriptions. Requires passing the publisher identity to Router.match/3.
• retain_handling: Controls retained message delivery on subscribe:
  • 0 — Send retained messages (default)
  • 1 — Send retained messages if the subscription does not already exist
  • 2 — Do not send retained messages on subscribe

← Back to the documentation index