An MQTT connector based on emqtt for Broadway.

MQTT is a lightweight publish/subscribe protocol widely used in IoT, industrial automation, and telemetry. This library connects a Broadway pipeline to an MQTT broker, using the MQTT protocol itself for backpressure and message reliability rather than an in-process buffer.

def deps do
  [
    {:off_broadway_emqtt, "~> 0.3.0"}
  ]
end

By default, :emqtt compiles the Quic transport library. To build without it:

BUILD_WITHOUT_QUIC=1 mix deps.compile

defmodule MyBroadway do
  use Broadway

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {OffBroadway.EMQTT.Producer,
          topics: [
            {"sensors/status",      :at_most_once},    # QoS 0
            {"sensors/temperature", :at_least_once},   # QoS 1
            {"sensors/humidity",    :exactly_once}     # QoS 2
          ],
          max_inflight: 100,
          config: [
            host: "mqtt.example.com",
            port: 1883,
            username: "my-user",
            password: "my-password",
            clientid: "my-pipeline"
          ]
        },
        concurrency: 1
      ],
      processors: [default: [concurrency: 10]],
      batchers: [default: [batch_size: 100, batch_timeout: 500, concurrency: 5]]
    )
  end

  @impl true
  def handle_message(_, message, _context) do
    IO.inspect(message.data, label: "payload")
    IO.inspect(message.metadata.topic, label: "topic")
    message
  end

  @impl true
  def handle_batch(_, messages, _, _) do
    messages
  end
end

Each MQTT message is delivered as a Broadway message where data is the raw payload binary and metadata contains the remaining fields from the MQTT packet (topic, qos, packet_id, etc.).

Use QoS 1 (:at_least_once) or QoS 2 (:exactly_once) for reliable delivery. Messages are only acknowledged to the broker after Broadway successfully processes them. If Broadway fails to process a message, the broker redelivers it (with QoS 1/2).

QoS 0 (:at_most_once) provides no delivery guarantee. The broker fires and forgets.

The max_inflight option limits how many unACKed QoS 1/2 messages the broker will send before waiting for acknowledgements. This is the primary backpressure mechanism: the broker stops delivering new messages once the window is full.

For MQTT v5 (config: [proto_ver: :v5]), Receive-Maximum is set automatically in the CONNECT properties so the broker enforces the limit server-side.

clean_start defaults to false. When a producer restarts (after a crash or deployment), the broker recognises the clientid and redelivers any QoS 1/2 messages that were not acknowledged before the restart. No messages are lost between restarts.

If you want a fresh session on every connect (discarding unACKed messages), set config: [clean_start: true] explicitly.

By default, if the MQTT connection is lost the producer process stops and Broadway's supervisor restarts it. The fresh producer creates a new emqtt connection and re-subscribes to all topics. With clean_start: false (the default), the broker redelivers any unACKed QoS 1/2 messages.

Alternatively, you can enable emqtt's built-in reconnect by passing reconnect in config:

config: [
  host: "mqtt.example.com",
  clientid: "my-pipeline",
  reconnect: :infinity,
  reconnect_timeout: 5,
  clean_start: false
]

To distribute messages across multiple producer instances, set concurrency > 1 and provide a shared_group name. Without shared_group, every producer instance receives every message, causing duplicates.

producer: [
  module: {OffBroadway.EMQTT.Producer,
    topics: [{"work/queue", :at_least_once}],
    shared_group: "my-pipeline",
    config: [host: "mqtt.example.com", clientid: "worker"]
  },
  concurrency: 3
]

This creates three MQTT connections (worker_0, worker_1, worker_2) all in the shared group my-pipeline. The broker distributes messages across them using MQTT shared subscriptions ($share/my-pipeline/work/queue).

max_inflight applies per connection. With concurrency: 3 and max_inflight: 100, up to 300 unACKed messages can be in-flight across the pipeline at once.

Implement OffBroadway.EMQTT.MessageHandler to customise how MQTT messages are converted to Broadway messages. The default handler places the raw payload binary in data and the rest of the MQTT packet fields in metadata.

defmodule MyApp.JsonHandler do
  @behaviour OffBroadway.EMQTT.MessageHandler

  @impl true
  def handle_message(message, ack_ref, _opts) do
    {payload, metadata} = Map.pop(message, :payload)

    %Broadway.Message{
      data: Jason.decode!(payload),
      metadata: metadata,
      acknowledger: {OffBroadway.EMQTT.Acknowledger, ack_ref, %{}}
    }
  end
end

producer: [
  module: {OffBroadway.EMQTT.Producer,
    topics: [{"events/#", :at_least_once}],
    message_handler: MyApp.JsonHandler,
    config: [host: "mqtt.example.com"]
  },
  concurrency: 1
]

See the Producer moduledoc for common connection.down reason shapes (auth failures, TLS errors, transport errors).

See CHANGELOG.md for upgrade instructions and release history.