Realtime MQTT Broker

Category: Backend & APIs

This page is generated from the Air Pipe marketplace. Browse it live to install into your organization.

A real MQTT broker — pub/sub, wildcards, retained messages, QoS 0/1/2, sessions — the alternative to EMQX / HiveMQ / AWS IoT Core, with one thing they don't have: every interface-shaped message can run your Air Pipe pipeline (validate, enrich, store, alert). One vendor for transport + logic + auth + billing, carried over secure WebSocket with no broker to operate.

Great for IoT telemetry, device commands, presence/status, and any pub/sub where you also want to process the message.

How Air Pipe compares (MQTT)

Capability	EMQX / HiveMQ	AWS IoT Core	Air Pipe
Pub/sub + wildcards (`+`, `#`)	✅	✅	✅
Retained messages	✅	✅	✅
QoS 0 / 1 / 2	✅	0/1 only	✅
Persistent sessions, LWT, keep-alive	✅	✅	✅
Per-topic subscribe authorization	✅	✅ (policies)	✅ `subscribe_authorizer`
Auth required (no anonymous foot-gun)	optional	✅	✅
Run your logic on the message	❌ (needs rules engine + Lambda)	❌ (Rules → Lambda/SQS)	🏆 built-in pipeline
One vendor: broker + compute + auth + billing	❌	partial (AWS sprawl)	🏆
Per-tenant safety on shared infra	❌ (you build it)	✅ (account quotas)	🏆 built-in floors
No broker/cluster to operate	❌	managed	🏆

Parity on every broker fundamental; ahead on "broker + your backend logic", one bill, and safe multi-tenancy.

What's included

File	Purpose
`broker.yml`	`telemetry` (validate + process each reading), `commands` (retained, authorized device commands), `authorize-sub` (per-topic subscribe ACL).

Endpoint

Transport	Connect URL	Auth
MQTT over WebSocket	`wss://your-airpipe-host/mqtt` (sub-protocol `mqtt`)	username `<ORG_UUID>` · password `<API_KEY>`

Topic format for pipeline triggers is <environment>/<interface>/<trigger> — here production/telemetry/reading and production/commands/set. Plain pub/sub between clients works on any topic; interface-shaped topics additionally run a pipeline.

Broker features (no config needed)

Wildcards — subscribe production/telemetry/+ (one level) or production/# (all).
Retained — publish with retain=true; a new/reconnecting subscriber gets the last value immediately (great for device state + last command).
QoS 0/1/2 — pick per publish/subscribe; QoS 1/2 give at-least-once / exactly-once.
Sessions/LWT — clean or persistent sessions, keep-alive, last-will.

Quick start

Node (MQTT.js) — device publishes telemetry, dashboard subscribes with a wildcard:

const mqtt = require("mqtt");
const c = mqtt.connect("wss://your-airpipe-host/mqtt", { username: "<ORG_UUID>", password: "<API_KEY>" });

c.on("connect", () => {
  // dashboard: raw readings on .../reading AND the pipeline's ENRICHED output on .../processed
  c.subscribe("production/telemetry/+");
  // device: publish a reading — the broker delivers it AND the `telemetry` pipeline runs,
  // validates + stamps it, and republishes the enriched reading to production/telemetry/processed
  c.publish("production/telemetry/reading",
    JSON.stringify({ device_id: "sensor-01", temp_c: 21.4, humidity: 48 }), { qos: 1 });
});
c.on("message", (topic, msg) => console.log(topic, JSON.parse(msg.toString())));
// → production/telemetry/reading   {device_id, temp_c, humidity}                (raw, from the device)
// → production/telemetry/processed {device_id, temp_c, humidity, reading_id, received_at}  (pipeline output)

Retained device command (backend publishes, devices get it live + on reconnect):

c.publish("production/commands/set",
  JSON.stringify({ device_id: "sensor-01", command: "reboot" }),
  { qos: 1, retain: true });   // retained → a device that reconnects still receives it

Python (paho-mqtt, WebSocket transport):

import json, paho.mqtt.client as mqtt
c = mqtt.Client(transport="websockets")
c.username_pw_set("<ORG_UUID>", "<API_KEY>")
c.ws_set_options(path="/mqtt"); c.tls_set()          # wss://
c.connect("your-airpipe-host", 443); c.loop_start()
c.subscribe("production/telemetry/#")
c.publish("production/telemetry/reading",
          json.dumps({"device_id": "sensor-01", "temp_c": 21.4}), qos=1)

MQTT publish is fire-and-forget — the pipeline runs server-side with no reply to the publisher. To see a pipeline's result, persist it (a database: insert) or forward it (http:), then read it back.

Customisation

Store telemetry: add a Store action (database: main + INSERT) + a global.databases.main block pointing at a|ap_var::DATABASE_URL| (Postgres/MySQL/Mongo/MSSQL/SQLite) — see rest-api-starter.
Alert on threshold: add an action that checks temp_c and calls an http: webhook (Slack/PagerDuty) when exceeded.
Tie subscribe ACL to device identity: authorize-sub gets the requested topic as a|body::channel|; bind it to a JWT claim or a device registry lookup instead of a fixed topic.
Pure ingest starter: if you only need "device → pipeline" (no pub/sub between clients), see the mqtt-telemetry-ingest pack.

Notes

Targets MQTT-over-WebSocket clients (MQTT.js, paho with transport="websockets", HiveMQ web client, most modern SDKs). Native raw-TCP MQTT for embedded devices is a separate endpoint.
Safe on shared infra: per-org limits bound subscriptions, publish rate, retained messages and topic count, so one org can't exhaust a node. Higher plans lift the ceilings.
Each message that triggers a pipeline counts as one request against your plan; plain pub/sub delivery between clients does not.
Only production and staging environments are accepted in the topic.

Configuration

broker.yml

name: RealtimeMqtt
description: >
  A real MQTT broker — pub/sub, wildcards, retained messages and QoS 0/1/2 — the alternative to
  EMQX / HiveMQ / AWS IoT Core, PLUS every interface-shaped message can run your Air Pipe pipeline
  (validate, enrich, store, alert). One vendor for transport + logic + auth + billing, on shared
  infra that stays safe under per-org limits. Carried over secure WebSocket, no broker to operate.

docs: true

# ── How it fits together ────────────────────────────────────────────────────────
# Connect an MQTT-over-WebSocket client to  wss://your-airpipe-host/mqtt  (sub-protocol `mqtt`):
#     username = <ORG_UUID>      password = <API_KEY>
#
# It is a full broker, so plain pub/sub works between any clients (no config needed):
#     • wildcards   — subscribe `production/telemetry/+`  or  `production/#`
#     • retained    — publish with retain=true; new subscribers get the last value immediately
#     • QoS 0/1/2   — at-most-once / at-least-once / exactly-once, per the MQTT spec
#     • sessions    — clean/persistent sessions, keep-alive, LWT
#
# ON TOP of that, a PUBLISH to an interface-shaped topic `<env>/<interface>/<trigger>` ALSO runs
# that interface's pipeline in-process (the published JSON is `a|body|`). That is the differentiator:
# a broker AND your compute on the same message, no separate rules engine / Lambda / queue.

interfaces:

  # ── Telemetry ingest + processing ───────────────────────────────────────────────
  # Devices PUBLISH readings to `production/telemetry/reading`. The broker delivers them to any
  # subscriber (dashboards on `production/telemetry/+`, etc.) AND this pipeline runs: validate the
  # reading, stamp it, and flag anything over a threshold. Swap the flag for a DB insert / webhook.
  telemetry:
    mqtt: "reading"                # realtime — no `output` needed
    # output: http                 # optional: also accept a reading over HTTP (dual interface)
    summary: Validate + process device telemetry (MQTT)
    description: Each reading published to production/telemetry/reading is validated, stamped and flagged.
    tags: [mqtt, iot, telemetry, pubsub, processing]
    request_example:
      device_id: sensor-01
      temp_c: 21.4
      humidity: 48
    actions:
      - name: Validate
        input: a|body|
        hide_data_on_success: true
        assert:
          error_message: "device_id and temp_c are required"
          tests:
            - value: device_id
              is_not_null: true
              is_not_empty: true
            - value: temp_c
              is_not_null: true
      - name: Stamp
        run_when_succeeded:
          actions: [Validate]
        input: a|Validate|
        post_transforms:
          - add_attribute:
              reading_id: a|uuid|
              received_at: a|timestamp:datetimeutctz|
      # Fan the ENRICHED reading back onto the broker (pipeline → topic) so dashboards can consume
      # the processed/stamped stream on `production/telemetry/processed` — something a plain broker
      # can't do, and a plain rules engine needs a separate service for. Publishing to a different
      # trigger (`processed` ≠ `reading`) so it does not re-trigger this pipeline.
      - name: Republish
        run_when_succeeded:
          actions: [Stamp]
        mqtt_publish:
          topics: production/telemetry/processed
          data: a|Stamp|
          qos: 1
      # To persist every reading: add a `Store` action (database: main + INSERT) and a
      # `global.databases.main` block pointing at a|ap_var::DATABASE_URL| — see rest-api-starter.

  # ── Device commands (retained + gated subscribe) ────────────────────────────────
  # Your backend PUBLISHES a command to `production/commands/set` with retain=true; every device
  # subscribed to it gets the command immediately, and a device that reconnects gets the LAST
  # command from the retained value (no missed commands). `subscribe_authorizer` gates who may
  # subscribe. This pipeline also runs on each command publish, so you can log/audit it.
  commands:
    mqtt: "set"                    # realtime — no `output` needed
    subscribe_authorizer: authorize-sub
    # output: http                 # optional: also accept a command over HTTP (dual interface)
    summary: Deliver device commands (retained, authorized)
    description: Backend publishes a retained command; authorized devices receive it live + on reconnect.
    tags: [mqtt, iot, commands, retained]
    request_example:
      device_id: sensor-01
      command: reboot
    actions:
      - name: Audit
        input: a|body|
        post_transforms:
          - add_attribute:
              command_id: a|uuid|
              issued_at: a|timestamp:datetimeutctz|

  # Subscribe ACL — runs on every device SUBSCRIBE with the requested topic as a|body::channel|.
  # Allowed iff it returns 2xx; a denial drops the SUBSCRIBE (no subscription established). Here we
  # only allow the commands topic; bind it to a device identity/claim for real deployments.
  authorize-sub:                   # internal authorizer — run on SUBSCRIBE, not an HTTP route
    actions:
      - name: Check
        input: a|body|
        assert:
          http_code_on_error: 403
          error_message: "topic not allowed"
          tests:
            - value: channel
              is_equal_to: production/commands/set

Realtime MQTT Broker

How Air Pipe compares (MQTT)​

What's included​

Endpoint​

Broker features (no config needed)​

Quick start​

Customisation​

Notes​

Configuration​

broker.yml​