A forwarder, or the hub of a pipeline

Mosquitto hands every topic's messages to its subscribers untouched. Nothing gets picked over, nothing gets reshaped. That is a perfectly good broker and for a lot of setups it is all you need.

It stops being enough the moment you want the broker itself to make a decision. Strip the noise out of a sensor reading before anything downstream sees it. Move a reading from the topic the device insists on publishing to, onto the topic your dashboard is subscribed to. Send a notification the second a client drops off the network. None of that is forwarding. All of it is processing the data itself, and that is what EMQX's Rule Engine is for.

At its core the Rule Engine is three stages, source → transformation → action, and it is worth fixing those in your head before any syntax, because every problem you will have later belongs to exactly one of them.

• Data source — where the rule gets its data. It can be MQTT messages, from one topic or several; an MQTT event such as $events/client_connected; or data handed back by Data Integration.
• Transformation — the rule SQL decides which records to take, which fields to keep, and what conditions and functions to apply. FROM names the source, WHERE adds the condition, and SELECT decides the output.
• Action — what happens once a rule matches. EMQX 5.8.9 offers republish, console (print to the console or the log) and Forwarding to Sinks, which sends it to an external system.

flowchart LR
  A["MQTT messages
from one topic or several,
such as t/#"] --> S["The rule SQL
FROM names the source
WHERE adds the condition
SELECT decides the output"]
  B["An MQTT event,
such as
$events/client_connected"] --> S
  C["Data handed back
by Data Integration"] --> S
  S --> D["Republish
to another MQTT topic"]
  S --> E["Console output
to the console or the log"]
  S --> F["Forwarding to a Sink,
out to an external system"]

Where do rules live? EMQX 5.x files them under Data Integration, and you create or edit one from the Dashboard sidebar at Integration → Rules. Note that the Rule Engine handles more than messages: it also handles events, such as a client connecting or disconnecting.

The words, before you meet them in a form field

Eleven terms carry the whole of this part. They come from three different chapters of the source guide and they sound alike on purpose, which is exactly why they get mixed up.

Rule SQL: SELECT, FROM, WHERE

If you have written a database query before, this will look familiar and then behave slightly differently, so read it as its own thing. The basic shape is:

The square brackets around WHERE mean it is optional. Plenty of useful rules never have one.

FROM — naming the source

FROM comes in one of two main shapes.

A topic. For example, this handles every message whose topic matches t/#:

Several topics are separated with commas:

A topic must be wrapped in double or single quotes. That is the single most common thing people leave off, and it is why it has its own line in the summary at the top of this page.

An event. For example, this handles the event of a client connecting successfully:

Event topics start with $events/ and cannot be read with an ordinary subscription. They are not hidden topics you could have subscribed to if only you had known the name; they exist for a rule's FROM clause and nowhere else. If you want that data outside EMQX, a rule has to republish it for you.

WHERE — the extra filter

WHERE narrows what the rule acts on. A field in the condition can be either of two things:

• Metadata, such as clientid or username.
• A field inside the payload, reached with dot syntax such as payload.x.y — as long as the payload is JSON or a Map.

Conditions combine with and and or.

SELECT — the output, and renaming

SELECT picks the fields to output, and it can rename them on the way through:

Read that one twice: there are two different clientid values in play. The bare clientid is the metadata one — who actually published the message. The payload.clientid is whatever the device wrote inside its own JSON, which may be something else entirely. Renaming it to myclientid is how you keep both without a collision.

The output of the SQL is a JSON object. Once the SQL has run, a later action can reference its fields with ${field}. That handoff — SQL produces a named field, action reads it back by name — is the joint that the whole of the next section turns on.

flowchart TD
  A["A message is published
on topic t/1"] --> B{"Does the topic match
the FROM pattern?"}
  B -->|"no"| X["The rule never fires.
Nothing is logged,
nothing is wrong"]
  B -->|"yes"| C{"Is there a WHERE clause,
and does it hold?"}
  C -->|"no, it does not hold"| X
  C -->|"yes, or there is none"| D["SELECT builds the output:
a JSON object of
the fields you named"]
  D --> E["The action reads those fields
back by name, as placeholders"]

Expressions and the escape hatch

Expressions support arithmetic (+, -, *, div, mod), comparison (=, <>, <, > and so on) and logic (and, or), on top of a long list of built-in functions.

There is also a separate FOREACH statement that can emit several results at once — splitting an array of sensor readings in the payload into one message each, for instance. It is genuinely useful and it is not a beginner's tool. Learn it when you get to a payload that needs it, not before.

Two layers of fields, and the functions over them

Which fields a rule can reference depends on which data source it reads. Get this wrong and the symptom is an empty output rather than an error message, which is a lot more confusing than a crash would have been.

For an MQTT message, the metadata gives you these common fields:

An event source such as $events/client_connected adds fields of its own: clientid, username, keepalive, is_bridge and connected_at. And $events/client_disconnected also carries reason, which says why the client dropped — the single most useful field in the whole set when you are chasing a device that keeps falling off. The official documentation lists every event and its fields.

Built-in functions

Functions can be used inside both SELECT and WHERE. The common groups:

They go straight into the clause, and the as rename works on the result:

From t/# to a/1, with a test before it goes live

The fastest way to learn this is the republish rule from the official EMQX tutorial: everything arriving on t/# gets a copy published to a/1. It does nothing useful, which is exactly what you want the first time — every part of the mechanism is visible and nothing important breaks if it goes wrong.

1. Step 1

   Open the rules page

   In the EMQX Dashboard sidebar click Integration → Rules, then Create at the top right. The Create Rule page opens, with an SQL Editor filling most of it and an actions panel down the right-hand side.

2. Step 2

   Fill in the rule details and write the SQL

   Give the rule a name and a note, then type this into the SQL Editor:

   SELECT * FROM "t/#"

   That selects every message whose topic matches t/#. The quotes around the topic are not optional.

3. Step 3

   Test the SQL with Try It Out, before anything is created

   Turn on the Try It Out switch on the Create Rule page. Choose the matching Data Source, then fill in the simulated data — Client ID, Username, Topic, QoS, Payload and so on — and click Run Test.

   You are looking for Test Passed, and for Output Result on the right to show what the SQL actually produced. Read that output rather than glancing at it. It is the only place you will see the shape of the JSON object your action is about to receive, and half the problems in the troubleshooting section further down are visible here first.

4. Step 4

   Add an action

   Click Add Action on the right and choose Republish. Set Target Topic to a/1, QoS to 0 and Retain to false. Then, in the Payload field, type the placeholder:

   ${payload}

   That keeps the original payload intact on the copy. Then click Add to return to the Create Rule page.

5. Step 5

   Create the rule and trigger it

   Back on the Create Rule page, click Create at the bottom to finish. Now prove it: subscribe to a/1, publish a message on topic t/1, and the republished message should arrive on a/1.

That is the whole loop, and it is the loop you will run for every rule you ever write: SQL, test it in place, attach an action, create, trigger it by hand.

Republish, Console Output, and the same thing in a config file

Data that stops at the SQL output is data you cannot use. The action is what turns the result into behavior. In a rule's actions you can configure one or more actions, and the rule runs them in order.

EMQX 5.8.9 has three built-in actions:

• Message Republishing — publishes the rule result to an MQTT topic.
• Console Output — prints the result to the console or the log, mainly for debugging.
• Forwarding to Sinks — hands the result to a Sink, which is the next section.

The data an action uses comes from the SQL result. With placeholder syntax such as ${field} you push a field of the rule output into an action parameter — putting the payload or clientid the SQL selected into the republish topic or payload, for example.

Republish, parameter by parameter

Republish is the everyday “reshape the data and reroute it” action. In the official example, messages matching t/# are republished to a/1.

The important thing about it, and the thing most people assume wrongly: republishing does not intercept the original message. A client subscribed to t/1 still receives it exactly as usual. It simply gets one extra copy on a/1. Nothing is taken away from anybody.

The MQTT 5.0 properties live behind a switch labeled MQTT 5.0 Message Properties on the action form. Turning it on lets you set Payload Format Indicator, Message Expiry Interval, Content Type, Response Topic, Correlation Data and other properties. Leave it off until you have a reason.

The recursion trap

Direct Dispatch deserves its own paragraph, because the failure it prevents is the one genuinely nasty mistake available in this whole chapter. Turning it on delivers the message straight to subscribers, which keeps it from triggering other rules and falling into recursion.

Here is the shape of the accident. Your rule reads FROM "t/#". You set the republish target to something that also matches t/# — t/2, say. The republished message now matches the rule that produced it, so the rule fires again, republishes again, and it does not stop.

flowchart TD
  A["The rule reads FROM t/#
and republishes"] --> C{"Does the target topic
itself match t/#?"}
  C -->|"a/1, outside the pattern"| D["One extra copy on a/1.
Subscribers of t/1 still
get the original"]
  C -->|"t/2, inside the pattern"| E["The republished message
matches the same rule again"]
  E --> F["It fires again,
without end"]
  F --> G["Turn on Direct Dispatch,
or change the target topic"]

Console Output, and where the text comes out

The Console Output action prints the rule result to the console or the log, in the form [rule action] ruleID + Action Data + Envs. Where you read that depends entirely on how EMQX was started:

The same rule, declared in a config file

Rules and actions do not have to be clicked together in the Dashboard. They can also go into emqx.conf, where they take effect before EMQX starts — which is what you want if the rule is part of a deployment rather than something you set up by hand on one machine.

They are defined under the rule_engine namespace with rules.<id>, where the id is a name you choose. This is the official example:

The braces carry the structure, so indenting the nested lines in your own file is for your eyes only and changes nothing — which is why they are shown flush left here. Two details are worth pointing at. The inner quotes around the topic are escaped as \"t/a\", because the whole SQL string is already inside quotes — and the topic still has to be quoted, exactly as in the Dashboard. And ${qos} and ${y} are the same placeholders as in the UI, referencing fields of the SQL output.

Trace that example end to end, because it shows the handoff in miniature. Topic t/a receives {"x":1}. The rule picks out qos and renames the payload's x to y. Then it republishes to t/b with the payload y: 1. The y in the payload template is the y that SELECT invented one line earlier.

Two more config shapes are worth knowing. To name an external integration Sink as a rule's action, put the bridge ID straight into the actions array, such as "mqtt:my_egress_mqtt_bridge". And a console action is written as:

Event rules can be declared in a config file just the same — a client coming online, $events/client_connected, for instance.

Connectors and Sinks

Republish moves data to another MQTT topic, which is still inside EMQX. Getting it out to something that has never heard of MQTT — a phone notification service, an SMS gateway, a cloud API, a relational database — is a separate problem, and EMQX 5.x splits it into two halves.

• A Connector is the low-level connection channel from EMQX to an external data system. It is responsible only for how to connect, never for which data gets processed.
• A Sink decides which external target a rule's output goes to and in what format. It sits in the rule's Action Outputs, where you pick a Connector and fill in its parameters.

That split buys you three concrete things, and the third one is the one you will appreciate at two in the morning:

• Settings stay separate. Connection details — server address, credentials — sit apart from data settings such as rules, topic mapping and payload templates. Changing the connection does not affect the rules.
• Reuse. An external system that several Sinks and Sources talk to needs only one Connector.
• Observable state. The Dashboard shows each Connector's connection state — Connecting, Connected, Disconnected, Inconsistent — which tells you in one glance whether a problem is the connection or the rule.

flowchart LR
  R1["Rule one
its own SQL"] --> K1["Sink: HTTP Server,
method POST"]
  R2["Rule two
different SQL"] --> K2["Sink: HTTP Server,
a different body"]
  K1 --> C["One Connector: my_httpserver
address and credentials only.
State: Connected"]
  K2 --> C
  C -->|"data out"| OUT["The external
HTTP service"]
  OUT -->|"data in"| C
  C --> SRC["A Source, bringing data
from outside into EMQX"]
  SRC --> R3["A rule inside EMQX,
reading what came in"]

What you can actually reach, and on which edition

EMQX 5.8.9 offers many external targets you can use as a Sink, but the Open Source edition can reach only a few of them. The official Connector documentation states that EMQX Open Source supports only two connectors, HTTP and MQTT. Kafka, PostgreSQL, MySQL, AWS and Azure targets are Enterprise edition features.

Read the edition column before you plan anything around a target, not after.

Webhook or HTTP Server

Both are Open Source, both send data over HTTP, and the difference between them is mainly whether a rule (SQL) processes the data first — mainly, and not only, because the two also differ in how much of the outgoing request you get to build.

An HTTP Sink can use POST, PUT and other methods, and you can set the request body and the credentials — the ${var} placeholder works there too. EMQX shows each Sink's connection state and request count, so confirming the data is actually going out takes a glance rather than an investigation.

Hands-on: t/# out to a local HTTP service

This follows the official tutorial's HTTP Server Sink example: send the MQTT messages on t/# to a Flask HTTP service running on the local machine.

1. Step 1

   Create the HTTP Server Connector

   In the Dashboard sidebar go to Integration → Connectors, click Create, and pick the HTTP Server connector. Name it my_httpserver and set the URL to:

   http://localhost:5000

   Click Test Connectivity before you save. Ten seconds here saves you debugging a rule that was never the problem.

2. Step 2

   Create the rule

   Go to Integration → Rules → Create and enter the SQL:

   SELECT * FROM "t/#"
3. Step 3

   Add the HTTP Sink action

   Click + Add Action and set Type to HTTP Server. Pick my_httpserver from the Connector dropdown, set Method to POST, and fill in the name and description.

4. Step 4

   Create the rule

   Back on the Create Rule page, confirm the Sink is listed under Action Outputs, then click Create to finish. If the Sink is not listed there, the action was not saved and nothing further will work.

5. Step 5

   Trigger it with a message

   Use MQTTX to publish to t/1, for example:

   {"msg":"hello HTTP Server"}

   Then look at the Flask server for the POST request it received. That request landing is the proof; the Dashboard's request count on the Sink is the second opinion.

Working backwards from silence

Almost every failure in this part presents the same way: you publish a message and nothing happens at the other end. No error, no red box, just silence. The work is figuring out which of the three stages swallowed it.

flowchart TD
  A["You published,
and nothing arrived
at the other end"] --> B{"Is the rule enabled?"}
  B -->|"no"| B1["Enable it"]
  B -->|"yes"| C{"Does the FROM pattern match
the topic you published to?"}
  C -->|"no: wildcard or letter case"| C1["Fix the topic pattern"]
  C -->|"yes"| D{"Does Try It Out
report Test Passed?"}
  D -->|"no"| D1["The SQL is the problem.
Check the field names"]
  D -->|"yes"| E{"Republish, or a Sink?"}
  E -->|"republish"| F["Check the target topic and QoS,
and that the action was saved"]
  E -->|"sink"| G["Check the Connector reads Connected,
and the URL and Method"]

The rule side

1. 1

   The rule never fires

   Check that the topic pattern in FROM matches the topic being published to — watch the wildcards and the letter case, both of which are exact. Check that the rule is enabled. And check the event name is not misspelled: $events/client_connected, for example, has to be that string precisely.

2. 2

   The SQL output is empty, or the fields are wrong

   Check the official field reference and confirm the field name actually exists on that data source. Remember that metadata and payload are two separate layers of fields — a name that exists on one is not automatically available on the other.

3. 3

   The payload is not JSON but you want a field inside it

   Dot syntax only works on JSON or a Map. For anything else, use a cast function, or run it through a Schema first.

4. 4

   The rule matches but the action does not run

   Use Test the Rule in Try It Out to simulate the matching event or publish, or publish a test message directly. Then read the Action error in that run's log — the rule and the action fail separately and the log tells you which one gave up.

The republish and console side

1. 5

   No republished message arrives

   Four things, in order: that the rule is enabled; that the SQL's FROM pattern matches the topic you publish to; that the action was actually saved; and that the target topic and QoS are right. Simulate it first with Test the Rule rather than guessing.

2. 6

   Nothing shows up in the console

   This is nearly always about where you are looking rather than what EMQX is doing. When EMQX starts in the background the output goes to the journal, which you read with journalctl. When it runs in the Docker foreground, look at the console.

3. 7

   The payload is not in the format you expected

   The republish payload is a string template, so you have to type out the shape you want. That means ${payload} on its own for a straight copy, or text mixed with ${field} for anything else. It will not guess.

4. 8

   The rule triggers itself and recurses

   If the Republish target topic also matches the same rule's FROM, it fires again without end. Avoid that with Direct Dispatch, or change the target topic so it falls outside the pattern.

The connector and sink side

1. 9

   The external HTTP service receives nothing

   Check that the Connector state reads Connected, that the URL and Method are correct, that the Sink has actually been added to the rule, and that the rule is enabled. The Connector state is the fastest of those to read, so start there.

2. 10

   Kafka, PostgreSQL and MySQL connectors are not in the list

   They are not missing and nothing is broken. Open Source supports only the HTTP and MQTT connectors; those targets are Enterprise.

3. 11

   Data stops after you change a Connector

   Updating a Connector that is in use reloads its Sinks and Sources, which can interrupt traffic briefly. Expected behavior, which is why the advice is to make the change off peak.

4. 12

   A Connector will not delete

   A Connector that is in use cannot be deleted. Delete the Sinks and Sources that depend on it first; the Dashboard lists the Sinks attached to it, so you can work out what is holding it.

The ones that come up every time

Where to go from here