Data Sources and Fields

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:

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

Output:

{
  "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:

MQTT Events

You can use EMQX rules to handle event notifications, for example, client online and offline, client subscriptions, etc. In this case, you can use the FROM clause to specify the event topics. The event topic starts with "$events/", such as "$events/client_connected".

TIP

By default, clients are unable to subscribe directly to MQTT event messages. This tutorial will guide you through the process of using rules to subscribe to these messages. To enable direct subscriptions for MQTT event messages with your clients, you must first add the following configuration items to the emqx.conf file. After saving the modifications, restart EMQX to apply the new settings.

sys_topics {
    sys_event_messages {
        client_connected = true,
        client_disconnected = true,
        client_subscribed = true,
        client_unsubscribed = true
    }
}

See the table below for the supported event topic list.

Event Topic List

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:

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

Output:

{
  "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.

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:

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

Output:

{
  "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.

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:

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

Output:

{
  "topic": "t/a",
  "timestamp": 1645003103004,
  "reason": "no_subscribers",
  "qos": 1,
  "node": "emqx@127.0.0.1"
}
``` Field| Explanation  
---|---  
`id`| MQTT message ID  
`reason`| Dropping 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  
`clientid`| Client ID of the publisher  
`username`| Username of the publisher  
`payload`| MQTT payload  
`peerhost`| Client IP Address  
`topic`| MQTT topic  
`qos`| QoS levels  
`flags`| Flags  
`pub_props`| PUBLISH Properties (MQTT 5.0 only)  
`timestamp`| Event trigger time (unit: ms)  
`publish_received_at`| Time when PUBLISH message reaches EMQX (unit: ms)  
`node`| Node where the event is triggered  

### Message Dropped When Delivering Event (“$events/delivery_dropped”)

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

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:

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

Output:

{ "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|   
---|---  
`id`| MQTT message ID  
`reason`| Dropping 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_clientid`| Client ID of the publisher  
`from_username`| Username of the publisher  
`clientid`| Client ID of the subscriber  
`username`| Username of the subscriber  
`payload`| MQTT payload  
`peerhost`| Client IP Address  
`topic`| MQTT topic  
`qos`| Message QoS  
`flags`| Flags  
`pub_props`| PUBLISH Properties (MQTT 5.0 clients only)  
`timestamp`| Event trigger time (unit: ms)  
`publish_received_at`| Time when PUBLISH message reaches EMQX (unit: ms)  
`node`| EMQX 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:

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

Output:

{ "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:

Field| Explanation  
---|---  
`clientid`| Client ID  
`username`| Client username  
`mountpoint`| Mountpoint for bridging messages  
`peername`| IP address and port of terminal  
`sockname`| IP Address and Port listened by EMQX  
`proto_name`| Protocol name  
`proto_ver`| Protocol version  
`keepalive`| MQTT keepalive interval  
`clean_start`| MQTT clean_start  
`expiry_interval`| MQTT session expiration time  
`is_bridge`| Whether the client is acting as a bridge  
`connected_at`| Client connection completion time (unit: ms)  
`conn_props`| CONNECT Properties (MQTT 5.0 clients only)  
`timestamp`| Event trigger time (unit: ms)  
`node`| EMQX node where the event is triggered  

### 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:

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

Output:

{ "username": "u_emqx", "reason": "normal", "node": "emqx@127.0.0.1", "disconnected_at": 1645003578536, "clientid": "c_emqx" } `` Field| Explanation ---|---reason`| Disconnect 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.
clientid| Client ID
username| Client username
peername| IP Address and Port number
sockname| IP Address and Port number listened by EMQX
disconnected_at| Client disconnection completion time (unit: ms)
disconn_props| DISCONNECT Properties (MQTT 5.0 clients only)
timestamp| Event trigger time (unit: ms)
node| EMQX node where the event is triggered

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:

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

Output:

{
  "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:

[^*]: 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 (opens new window)).

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:

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

Output:

{
  "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.

Subscriber Event (“$events/session_subscribed”)

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

Example:

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

Output:

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

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

Unsubcribe Event (“$events/session_unsubscribed”)

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

Example:

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

Output:

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

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

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:

SELECT
  *
FROM
  "$bridges/mqtt:my_mqtt_bridge"

Output:

{
  "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: