SQL 数据源和字段

规则的 SQL 语句可以处理的数据源有：MQTT 消息、MQTT 事件，或者是数据桥接。

SQL 语句使用 FROM 来指定数据源，在 SELECT 和 WHERE 子句中可以引用相应的字段。 数据源类型不同，可以使用的字段也不同。

数据桥接

规则使用 $bridges/ 开头的主题来表示数据桥接的消息或事件。 格式为：$bridges/<type>:<name>。

其中 <type>:<name> 部分是数据桥接的 ID，<type> 是数据桥接的类型，<name> 是数据桥接的名字。 比如 $bridges/mqtt:my_mqtt_bridge。

MQTT 桥接事件 (“$bridges/mqtt:*“)

当该 MQTT 桥接从远程 MQTT Broker 接收到消息时触发规则

示例

SELECT
  *
FROM
  "$bridges/mqtt:my_mqtt_bridge"

输出:

{
  "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,
}

MQTT 消息

规则的 SQL 语句可以处理消息发布。 在一个规则语句中，用户可以用 FROM 子句指定一个或者多个主题， 当任何消息发布到指定的主题时都会触发该规则。

示例

SELECT
  *
FROM
  "t/#"

输出

{
  "clientid": "c_emqx",
  "event": "message.publish",
  "event_type": "message_publish",
  "flags": {},
  "id": "0005E27C1D24E44FF440000017520000",
  "metadata": {
    "rule_id": "sql_tester:099ddfa9c466d1ca"
  },
  "node": "emqx@127.0.0.1",
  "payload": "abc",
  "peerhost": "192.168.0.10",
  "pub_props": {
    "Message-Expiry-Interval": 30,
    "Payload-Format-Indicator": 0,
    "User-Property": {
      "foo": "bar"
    },
    "User-Property-Pairs": [
      {
        "key": "foo"
      },
      {
        "value": "bar"
      }
    ]
  },
  "publish_received_at": 1656397576334,
  "qos": 1,
  "timestamp": 1656397576334,
  "topic": "t/a",
  "username": "u_emqx"
}

MQTT 事件

规则的 SQL 语句既可以处理消息(消息发布)，也可以处理事件(客户端上下线、客户端订阅等)。对于消息，FROM 子句后面直接跟主题名；对于事件，FROM 子句后面跟事件主题。

事件消息的主题以 "$events/" 开头，比如 "$events/client_connected", "$events/session_subscribed"。 如果想让 emqx 将事件消息发布出来，可以在 emqx_rule_engine.conf 文件中配置。

FROM 子句可用的事件主题

消息投递事件 (“$events/message_delivered”)

当消息被放入底层socket时触发规则。

示例

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

输出

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

消息确认事件 (“$events/message_acked”)

当消息发送到客户端，并收到客户端回复的ack时触发规则，仅QOS1，QOS2会触发。

示例

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

输出

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

消息在转发的过程中被丢弃事件 (“$events/message_dropped”)

当一条消息无任何订阅者时触发规则。

示例

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

输出

{
  "topic": "t/a",
  "timestamp": 1645003103004,
  "reason": "no_subscribers",
  "qos": 1,
  "node": "emqx@127.0.0.1"
}

消息在投递的过程中被丢弃事件 (“$events/delivery_dropped”)

当订阅者的消息队列已满时触发规则。

示例

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

输出

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

终端连接成功事件 (“$events/client_connected”)

当终端连接成功时触发规则。

示例

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

输出

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

终端连接断开事件 (“$events/client_disconnected”)

当终端连接断开时触发规则。

示例

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

输出

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

连接确认事件 (“$events/client_connack”)

当服务端向客户端发送CONNACK报文时触发规则，reason_code 包含各种错误原因代码。

MQTT v5.0 协议将返回码重命名为原因码，增加了一个原因码来指示更多类型的错误(Reason code and ACK - MQTT 5.0 new features (opens new window))。 因此reason_code 在MQTT v3.1.1与MQTT v5.0中有很大的不同。

MQTT v3.1.1

MQTT v5.0

示例

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

输出

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

鉴权完成事件 (“$events/client_check_authz_complete”)

当客户端鉴权结束时触发规则。

示例

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

输出

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

终端订阅成功事件 (“$events/session_subscribed”)

当终端订阅成功时触发规则。

示例

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

输出

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

取消终端订阅成功事件 (“$events/session_unsubscribed”)

当取消终端订阅成功时触发规则。

示例

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

输出

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