Skip to content

Data Sources and Fields

Rules in EMQX can process data from various data sources, including MQTT Messages, MQTT Events, or Data Bridges.

As discussed in the Rule Engine Syntax section, you can use the FROM clause to specify the data source and the corresponding fields can be referenced in the SELECT and where clauses. This section will introduce the fields for MQTT Messages, MQTT Events, and Data Bridges.

MQTT Message

You can use EMQX rules to handle message publishing, in this case, you need to specify the message topics with the FROM clause.

For example, in the following statement, you will select fields payload.msg (then rename as msg with the AS clause), clientid, username, payload, topic, and qos for any message published to topics following the pattern t/#.

Example:

sql
SELECT
  payload.msg as msg,
  clientid,
  username,
  payload,
  topic,
  qos
FROM
  "t/#"

Output:

json
{
  "username": "u_emqx",
  "topic": "t/a",
  "qos": 1,
  "payload": "{\"msg\":\"hello\"}",
  "msg": "hello",
  "clientid": "c_emqx"
}

Refer to the table below for fields that can be selected from the received MQTT messages:

FieldExplanation
idMQTT message ID
clientidClient ID of the publisher
usernameUsername of the publisher
payloadMQTT payload
peerhostClient IP Address
topicMQTT topic
qosQoS level
flagsFlags
headersInternal data related to the message processing
pub_propsPUBLISH Properties (MQTT 5.0 clients only)
timestampTimestamp (unit: ms)
publish_received_atTime when PUBLISH message reaches EMQX (unit: ms)
nodeNode where the event is triggered
client_attrsClient attributes

MQTT Events

You can use EMQX rules to extract data from event topics to get event notifications, for example, client online and offline, client subscriptions, etc. The event topic starts with "$events/", such as "$events/client_connected", which can be specified in the FROM clause of the rule.

TIP

By default, clients are unable to subscribe directly to MQTT event messages. This section describes using rules to subscribe to these messages. You can also obtain the data from MQTT event messages by subscribing to system topics.

See the table below for the supported event topic list.

Event Topic List

Event topic nameExplanation
$events/message_deliveredMessage delivery
$events/message_ackedMessage acknowledged
$events/message_droppedMessage dropped when routing
$events/delivery_droppedMessage dropped when delivering
$events/client_connectedConnection complete
$events/client_disconnectedDisconnect
$events/client_connackConnection acknowledged
$events/client_check_authz_completeAuthorization check complete
$events/client_check_authn_completeAuthentication check complete
$events/session_subscribedSubscribe
$events/session_unsubscribedUnsubscribe

Message Delivery Event ("$events/message_delivered")

This event topic can be used to trigger a rule when a message is delivered to a client.

For example, to extract data from the "$events/message_delivered" event topic that includes the following data fields, ID and username of the publisher, message topic, message QoS, EMQX node with the event triggered, and the time when the event was triggered, you can use the statement below:

Example:

sql
SELECT
  from_clientid,
  from_username,
  topic,
  qos,
  node,
  timestamp
FROM
  "$events/message_delivered"

Output:

json
{
  "topic": "t/a",
  "timestamp": 1645002753259,
  "qos": 1,
  "node": "emqx@127.0.0.1",
  "from_username": "u_emqx_1",
  "from_clientid": "c_emqx_1"
}

Below are detailed explanations of each field.

CodeExplanation
idMQTT message ID
from_clientidClient ID of the publisher
from_usernameUsername of the publisher
clientidClient ID of the subscriber
usernameUsername of the subscriber
payloadMQTT payload
peerhostClient IP address
topicMQTT topic
qosQoS level
flagsFlags
pub_propsPUBLISH Properties (MQTT 5.0 clients only)
timestampEvent trigger time (unit: ms)
publish_received_atTime when PUBLISH message reaches EMQX
(unit: ms)
nodeEMQX node where the event triggered

Message Acknowledged Event ("$events/message_acked")

This event topic can be used to trigger a rule when the message delivery is acknowledged.

TIP

Only availabe for QOS 1 and QOS 2 messages.

For example, to extract data from the "$events/message_acked" event topic that includes the following data fields, ID and username of the publisher, message topic, message QoS, EMQX node where the event triggered, and the time when the event was triggered, you can use the statement below:

Example:

sql
SELECT
  from_clientid,
  from_username,
  topic,
  qos,
  node,
  timestamp
FROM
  "$events/message_acked"

Output:

json
{
  "topic": "t/a",
  "timestamp": 1645002965664,
  "qos": 1,
  "node": "emqx@127.0.0.1",
  "from_username": "u_emqx_1",
  "from_clientid": "c_emqx_1"
}

Below are detailed explanations of each field.

CodeExplanation
idMQTT message ID
from_clientidClient ID of the publisher
from_usernameUsername of the publisher
clientidClient ID of the subscriber
usernameUsername of the subscriber
payloadMQTT payload
peerhostClient IPAddress
topicMQTT topic
qosQoS levels
flagsFlags
pub_propsPUBLISH Properties (MQTT 5.0 only)
puback_propsPUBACK Properties (MQTT 5.0 only)
timestampEvent trigger time (in milliseconds)
publish_received_atTime when PUBLISH message reaches EMQX (unit: ms)
nodeEMQX node where the event triggered

Message Dropped When Routing Event ("$events/message_dropped")

This event topic can be used to trigger a rule when a message is dropped during routing.

For example, to extract data from the "$events/message_dropped" event topic that includes the following data fields: drop reason, message topic, message QoS, EMQX node where the event triggered, and the time when the event was triggered, you can use the statement below:

Example:

sql
SELECT
  reason,
  topic,
  qos,
  node,
  timestamp
FROM
  "$events/message_dropped"

Output:

json
{
  "topic": "t/a",
  "timestamp": 1645003103004,
  "reason": "no_subscribers",
  "qos": 1,
  "node": "emqx@127.0.0.1"
}
FieldExplanation
idMQTT message ID
reasonDropping reasons:

no_subscribers: No clients subscribe to the topic

receive_maximum_exceeded: awaiting_rel queue is full

packet_identifier_inuse: Receive a QoS 2 message with unreleased packet ID
clientidClient ID of the publisher
usernameUsername of the publisher
payloadMQTT payload
peerhostClient IP Address
topicMQTT topic
qosQoS levels
flagsFlags
pub_propsPUBLISH Properties (MQTT 5.0 only)
timestampEvent trigger time (unit: ms)
publish_received_atTime when PUBLISH message reaches EMQX (unit: ms)
nodeNode where the event is triggered

Message Dropped When Delivering Event ("$events/delivery_dropped")

This event topic can trigger a rule when a message is dropped during delivery.

For example, to extract data from the "$events/delivery_dropped" event topic that includes the following data fields: publisher ID and username, drop reason, message topic and QoS, you can use the statement below:

Example:

sql
SELECT
  from_clientid,
  from_username,
  reason,
  topic,
  qos
FROM "$events/delivery_dropped"

Output:

json
{
  "topic": "t/a",
  "reason": "queue_full",
  "qos": 1,
  "from_username": "u_emqx_1",
  "from_clientid": "c_emqx_1"
}

Below are detailed explanations of each field.

Explanation
idMQTT message ID
reasonDropping reasons:

queue_full: Message (QoS>0) queue is full.

no_local: Clients are not allowed to receive messages published by themselves.

expired: Message or the Session expired.

qos0_msg: Message (QoS 0) queue is full.
from_clientidClient ID of the publisher
from_usernameUsername of the publisher
clientidClient ID of the subscriber
usernameUsername of the subscriber
payloadMQTT payload
peerhostClient IP Address
topicMQTT topic
qosMessage QoS
flagsFlags
pub_propsPUBLISH Properties (MQTT 5.0 clients only)
timestampEvent trigger time (unit: ms)
publish_received_atTime when PUBLISH message reaches EMQX (unit: ms)
nodeEMQX node where the event is triggered

Connection Complete Event ("$events/client_connected")

This event topic can be used to trigger a rule when a client is connected successfully.

For example, to extract data from the "$events/client_connected" event topic that includes the following data fields: client ID and username, keepalive interval, and whether the connected MQTT client is acting as a bridge, you can use the statement below:

Example:

sql
SELECT
  clientid,
  username,
  keepalive,
  is_bridge
FROM
  "$events/client_connected"

Output:

json
{
  "username": "u_emqx",
  "keepalive": 60,
  "is_bridge": false,
  "clientid": "c_emqx"
}

Refer to the table below for fields that can be selected from the received MQTT messages:

FieldExplanation
clientidClient ID
usernameClient username
mountpointMountpoint for bridging messages
peernameIP address and port of terminal
socknameIP Address and Port listened by EMQX
proto_nameProtocol name
proto_verProtocol version
keepaliveMQTT keepalive interval
clean_startMQTT clean_start
expiry_intervalMQTT session expiration time
is_bridgeWhether the client is acting as a bridge
connected_atClient connection completion time (unit: ms)
conn_propsCONNECT Properties (MQTT 5.0 clients only)
timestampEvent trigger time (unit: ms)
nodeEMQX node where the event is triggered
client_attrsClient attributes

Disconnect Event ("$events/client_disconnected")

This event topic can be used to trigger a rule when a client is disconnected.

For example, you can use the statement below to extract data from the "$events/client_disconnected" event topic that includes the following data fields: client ID, username, disconnect reason, disconnect time, and EMQX node where the event is triggered.

Example:

sql
SELECT
  clientid,
  username,
  reason,
  disconnected_at,
  node
FROM
  "$events/client_disconnected"

Output:

json
{
  "username": "u_emqx",
  "reason": "normal",
  "node": "emqx@127.0.0.1",
  "disconnected_at": 1645003578536,
  "clientid": "c_emqx"
}
FieldExplanation
reasonDisconnect reasons

normal: The client is intentionally disconnected

kicked: EMQX has forcibly removed the client through REST API

keepalive_timeout: The specified keepalive time period expired.

not_authorized: Authorization failed.

tcp_closed: The peer has closed network connection.

discarded: Another client ( with clean_start set to true) connected with the same ClientID, causing the previous connection to be dropped.

takenover: Another client ( with clean_start set to false) connected with the same ClientID, taking over the previous connection..

internal_error: An error has occurred due to an improperly formatted message or other unknown issues.
clientidClient ID
usernameClient username
peernameIP Address and Port number
socknameIP Address and Port number listened by EMQX
disconnected_atClient disconnection completion time (unit: ms)
disconn_propsDISCONNECT Properties (MQTT 5.0 clients only)
timestampEvent trigger time (unit: ms)
nodeEMQX node where the event is triggered
client_attrsClient attributes

Connection Acknowlege Event ("$events/client_connack")

This event topic can be used to trigger a rule when the EMQX sends a CONNACK packet to the client.

Example:

sql
SELECT
  clientid,
  username,
  reason_code,
  node
FROM
  "$events/client_connack"

Output:

json
{
  "username": "u_emqx",
  "reason_code": "success",
  "node": "emqx@127.0.0.1",
  "connected_at": 1645003578536,
  "clientid": "c_emqx"
}

Refer to the table below for fields that can be extracted:

FieldExplanation
reason_codeReason code*
clientidClient ID of the publisher
usernameUsername of the publisher
peernameIP and port
socknameIP and port listened by EMQX
proto_nameProtocol name
proto_verProtocol version
keepaliveMQTT keepalive interval
clean_startMQTT clean_start
expiry_intervalMQTT session expiration time
conn_propsCONNECT Properties (MQTT 5.0 clients only)
timestampEvent trigger time (unit: ms)
nodeEMQX node where the alarm is triggered.

[^*]: The MQTT v5.0 protocol renames the return code to a reason code, adding a reason code to indicate more types of errors (Reason code and ACK - MQTT 5.0 new features).

Here is the reason code for MQTT v3.1.1 and MQTT v5.0.

Authorization Check Complete Event ("$events/client_check_authz_complete")

This event topic can be used to trigger a rule when the authorization check for the client is complete.

Example:

sql
SELECT
  clientid,
  username,
  topic,
  action,
  result,
  authz_source,
  node
FROM
  "$events/client_check_authz_complete"

Output:

json
{
  "username": "u_emqx",
  "topic": "t/a",
  "action": "publish",
  "result": "allow",
  "authz_source": "cache",
  "node": "emqx@127.0.0.1",
  "clientid": "c_emqx"
}

Refer to the table below for fields that can be extracted.

FieldExplanation
clientidClient ID
usernameUsername
peerhostClient IP Address
topicMQTT topic
actionPublish or subscribe action
resultAccess control check result
authz_source The authorization source
timestampTimestamp (unit: ms)
nodeEMQX node where the event is triggered.
client_attrsClient attributes

Authentication Check Complete Event ("$events/client_check_authn_complete")

This event topic can be used to trigger a rule when the authentication check for the client is complete.

Example:

sql
SELECT
  clientid,
  username,
  reason_code,
  is_superuser,
  is_anonymous
FROM
  "$events/client_check_authn_complete"

Output:

json
{
  "clientid": "c_emqx",
  "username": "u_emqx",
  "reason_code": "success",
  "is_superuser": true,
  "is_anonymous": false
}

Refer to the table below for fields that can be extracted.

FieldExplanation
clientidClient ID
usernameUsername
peernameClient IP Address
reason_codeAuthentication result
is_superuserWhether this client is a super user
is_anonymousWhether this client is a anonymous user
client_attrsClient attributes

Subscriber Event ("$events/session_subscribed")

This event topic can be used to trigger a rule when the client subscribes successfully.

Example:

sql
SELECT
  clientid,
  username,
  topic,
  qos
FROM
  "$events/session_subscribed"

Output:

json
{
  "username": "u_emqx",
  "topic": "t/a",
  "qos": 1,
  "clientid": "c_emqx"
}

Refer to the table below for fields that can be extracted.

FieldExplanation
clientidClient ID
usernameClient username
peerhostClient IP Address
topicMQTT topic
qosQoS levels
sub_propsSUBSCRIBE Properties (MQTT 5.0 client only)
timestampEvent trigger time (unit: ms)
nodeEMQX node where the event is triggered
client_attrsClient attributes

Unsubcribe Event ("$events/session_unsubscribed")

The rule is triggered when the terminal subscription is cancelled successfully.

Example:

sql
SELECT
  clientid,
  username,
  topic,
  qos
FROM
  "$events/session_unsubscribed"

Output:

json
{
  "username": "u_emqx",
  "topic": "t/a",
  "qos": 1,
  "clientid": "c_emqx"
}

Refer to the table below for fields that can be extracted.

FieldExplanation
clientidClient ID
usernameClient username
peerhostClient IP Address
topicMQTT topic
qosQoS levels
unsub_propsUNSUBSCRIBE Properties (MQTT 5.0 clients only)
timestampEvent trigger time (unit: ms)
nodeEMQX node where the event is triggered
client_attrsClient attributes

Data Bridges

Rules use topics prefixed by $bridges/ to present messages or events triggered by a data bridge. The format is:

$bridges/<type>:<name>

Where

  • <type>:<name> is the bridge Id,
  • <type> is the bridge type,
  • <name> is the bridge name.

For example, the MQTT Bridge events can be referred to in the format of "$bridges/mqtt:*. To set a rule for all messages sent by the MQTT data bridge named my_mqtt_bridge, you can use the statement below:

Example:

sql
SELECT
  *
FROM
  "$bridges/mqtt:my_mqtt_bridge"

Output:

json
{
  "id": "0005E27C1D24E44FF440000017520000",
  "server": "broker.emqx.io:1883",
  "payload": "hello",
  "topic": "t/a",
  "qos": 1,
  "dup": false,
  "retain": false,
  "pub_props": {
    "Message-Expiry-Interval": 30,
    "Payload-Format-Indicator": 0,
    "User-Property": {
      "foo": "bar"
    },
    "User-Property-Pairs": [
      {
        "key": "foo"
      },
      {
        "value": "bar"
      }
    ]
  },
  "message_received_at": 1645002753259,
}

For each field in the returned output:

FieldExplanation
idMQTT message ID
serverServer name of the remote MQTT broker, such as "broker.emqx.io:1883"
payloadMQTT payload
topicMQTT topic
qosMQTT QoS
dupMQTT DUP flag
retainMQTT Retain flag
pub_propsPUBLISH Properties (MQTT 5.0 clients only)
message_received_atTimestamp when the message is received (unit: ms)