规则引擎语法与示列

    • 子句将规则挂载到某个主题上
    • SELECT 子句用于对数据进行变换，并选择出感兴趣的字段
    • WHERE 子句用于对 SELECT 选择出来的某个字段施加条件过滤
    1. ## SELECT 语句用于决定最终的输出结果里的字段。比如:
    2. ## 下面 SQL 的输出结果中将只有两个字段 "a" 和 "b":
    4. SELECT a, b FROM "t/#"
    6. # 选取 username 为 'abc' 的终端发来的消息，输出结果为所有可用字段:
    8. SELECT * FROM "#" WHERE username = 'abc'
    10. ## 选取 clientid 为 'abc' 的终端发来的消息，输出结果将只有 cid 一个字段。
    11. ## 注意 cid 变量是在 SELECT 语句中定义的，故可在 WHERE 语句中使用:
    13. SELECT clientid as cid FROM "#" WHERE cid = 'abc'
    15. ## 选取 username 为 'abc' 的终端发来的消息，输出结果将只有 cid 一个字段。
    16. ## 注意虽然 SELECT 语句中只选取了 cid 一个字段，所有消息发布事件中的可用字段 (比如 clientid, username 等) 仍然可以在 WHERE 语句中使用:
    18. SELECT clientid as cid FROM "#" WHERE username = 'abc'
    20. ## 但下面这个 SQL 语句就不能工作了，因为变量 xyz 既不是消息发布事件中的可用字段，又没有在 SELECT 语句中定义:
    22. SELECT clientid as cid FROM "#" WHERE xyz = 'abc'

    FROM 语句用于选择事件来源。如果是消息发布则填写消息的主题，如果是事件则填写对应的事件主题。

    FOREACH、DO 和 INCASE 子句

    如果对于一个数组数据，想针对数组中的每个元素分别执行一些操作并执行 Actions，需要使用 FOREACH-DO-INCASE 语法。其基本格式为:

    1. FOREACH <字段名> [DO <条件>] [INCASE <条件>] FROM <主题> [WHERE <条件>]
    • FOREACH 子句用于选择需要做 foreach 操作的字段，注意选择出的字段必须为数组类型
    • DO 子句用于对 FOREACH 选择出来的数组中的每个元素进行变换，并选择出感兴趣的字段
    • INCASE 子句用于对 DO 选择出来的某个字段施加条件过滤
    2. FOREACH
    3.     payload.sensors as e ## 选择出的字段必须为数组类型
    4. DO                       ## DO相当于针对当前循环中对象的 SELECT 子句, 决定最终的输出结果里的字段
    5.     clientid,
    6.     e.name as name,
    7.     e.idx as idx
    8. INCASE                  ## INCASE 相当于针对当前循环中对象的 WHERE 语句
    9.     e.idx >= 1          ## 对DO选择出来的某个字段施加条件过滤
    10. FROM "t/#"              ## 子句将规则挂载到某个主题上

    其中 DO 和 INCASE 子句都是可选的。

    比较符号

    函数名函数作用返回值
    >大于true/false
    <小于true/false
    <=小于等于true/false
    >=大于等于true/false
    <>不等于true/false
    !=不等于true/false

    SQL 语句示例

    • 从 topic 为 “t/a” 的消息中提取所有字段:
    1. SELECT * FROM "t/a"
    • 从 topic 为 “t/a” 或 “t/b” 的消息中提取所有字段:
    1. SELECT * FROM "t/a","t/b"
    • 从 topic 能够匹配到 ‘t/#’ 的消息中提取所有字段。
    1. SELECT * FROM "t/#"
    • 从 topic 能够匹配到 ‘t/#’ 的消息中提取 qos, username 和 clientid 字段:
    1. SELECT qos, username, clientid FROM "t/#"
    • 从任意 topic 的消息中提取 username 字段，并且筛选条件为 username = ‘Steven’:
    1. SELECT username FROM "#" WHERE username='Steven'
    • 从任意 topic 的 JSON 消息体(payload) 中提取 x 字段，并创建别名 x 以便在 WHERE 子句中使用。WHERE 子句限定条件为 x = 1。下面这个 SQL 语句可以匹配到消息体 {“x”: 1}, 但不能匹配到消息体 {“x”: 2}:
    1. SELECT payload FROM "#" WHERE payload.x = 1
    • 类似于上面的 SQL 语句，但嵌套地提取消息体中的数据，下面的 SQL 语句可以匹配到 JSON 消息体 {“x”: {“y”: 1}}:
    1. SELECT payload FROM "#" WHERE payload.x.y = 1
    • 在 clientid = ‘c1’ 尝试连接时，提取其来源 IP 地址和端口号:
    1. SELECT peername as ip_port FROM "$events/client_connected" WHERE clientid = 'c1'
    • 筛选所有订阅 ‘t/#’ 主题且订阅级别为 QoS1 的 clientid:
    1. SELECT clientid FROM "$events/session_subscribed" WHERE topic = 't/#' and qos = 1
    • 筛选所有订阅主题能匹配到 ‘t/#’ 且订阅级别为 QoS1 的 clientid。注意与上例不同的是，这里用的是主题匹配操作符 ‘=~’，所以会匹配订阅 ‘t’ 或 ‘t/+/a’ 的订阅事件:
    • 对于一个 MQTT 5.0 PUBLISH 消息，筛选出 Key 为 “foo” 的 User Property:
    1. SELECT pub_props.'User-Property'.foo as foo FROM "t/#"

    提示

    • FROM 子句后面的主题需要用双引号 ""，或者单引号 '' 引起来。
    • WHERE 子句后面接筛选条件，如果使用到字符串需要用单引号 '' 引起来。
    • FROM 子句里如有多个主题，需要用逗号 "," 分隔。例如 SELECT * FROM “t/1”, “t/2” 。
    • 可以使用使用 "." 符号对 payload 进行嵌套选择。

    遍历语法(FOREACH-DO-INCASE) 举例

    假设有 ClientID 为 c_steve、主题为 t/1 的消息，消息体为 JSON 格式，其中 sensors 字段为包含多个 Object 的数组:

    1. {
    2.     "date": "2020-04-24",
    3.     "sensors": [
    5.         {"name": "b", "idx":1},
    6.         {"name": "c", "idx":2}
    7.     ]
    8. }

    示例1: 要求将 sensors 里的各个对象，分别作为数据输入重新发布消息到 sensors/${idx} 主题，内容为 ${name}。即最终规则引擎将会发出 3 条消息:

    1. 主题：sensors/0 内容：a
    2. 主题：sensors/1 内容：b
    3. 主题：sensors/2 内容：c

    要完成这个规则，我们需要配置如下动作：

    • 动作类型：消息重新发布 (republish)
    • 目的主题：sensors/${idx}
    • 目的 QoS：0
    • 消息内容模板：${name}

    以及如下 SQL 语句：

    1. FOREACH
    2.     payload.sensors
    3. FROM "t/#"

    示例解析:

    这个 SQL 中，FOREACH 子句指定需要进行遍历的数组 sensors，则选取结果为:

    1. [
    2.   {
    3.     "name": "a",
    4.     "idx": 0
    5.   },
    6.   {
    7.     "name": "b",
    8.     "idx": 1
    9.   },
    10.   {
    11.     "name": "c",
    12.     "idx": 2
    13.   }
    14. ]

    FOREACH 语句将会对于结果数组里的每个对象分别执行 “消息重新发布” 动作，所以将会执行重新发布动作 3 次。

    示例2: 要求将 sensors 里的 idx 值大于或等于 1 的对象，分别作为数据输入重新发布消息到 sensors/${idx} 主题，内容为 clientid=${clientid},name=${name},date=${date}。即最终规则引擎将会发出 2 条消息:

    1. 主题：sensors/1 内容：clientid=c_steve,name=b,date=2020-04-24
    2. 主题：sensors/2 内容：clientid=c_steve,name=c,date=2020-04-24
    • 动作类型：消息重新发布 (republish)
    • 目的主题：sensors/${idx}
    • 目的 QoS：0
    • 消息内容模板：clientid=${clientid},name=${name},date=${date}

    以及如下 SQL 语句：

    1. FOREACH
    2.     payload.sensors
    3. DO
    4.     clientid,
    5.     item.name as name,
    6.     item.idx as idx
    7. INCASE
    8.     item.idx >= 1
    9. FROM "t/#"

    示例解析:

    这个 SQL 中，FOREACH 子句指定需要进行遍历的数组 sensors; DO 子句选取每次操作需要的字段，这里我们选了外层的 clientid 字段，以及当前 sensor 对象的 name 和 idx 两个字段，注意 item 代表 sensors 数组中本次循环的对象。INCASE 子句是针对 DO 语句中字段的筛选条件，仅仅当 idx >= 1 满足条件。所以 SQL 的选取结果为:

    1. [
    2.   {
    3.     "name": "b",
    4.     "idx": 1,
    5.     "clientid": "c_emqx"
    6.   },
    7.   {
    8.     "name": "c",
    9.     "idx": 2,
    10.     "clientid": "c_emqx"
    11.   }
    12. ]

    FOREACH 语句将会对于结果数组里的每个对象分别执行 “消息重新发布” 动作，所以将会执行重新发布动作 2 次。

    在 DO 和 INCASE 语句里，可以使用 item 访问当前循环的对象，也可以通过在 FOREACH 使用 as 语法自定义一个变量名。所以本例中的 SQL 语句又可以写为：

    1. FOREACH
    2.     payload.sensors as s
    3. DO
    4.     clientid,
    5.     s.name as name,
    6.     s.idx as idx
    7. INCASE
    8.     s.idx >= 1
    9. FROM "t/#"

    示例3: 在示例2 的基础上，去掉 clientid 字段 c_steve 中的 c_ 前缀

    在 FOREACH 和 DO 语句中可以调用各类 SQL 函数，若要将 c_steve 变为 steve，则可以把例2 中的 SQL 改为：

    1. FOREACH
    2.     payload.sensors as s
    3. DO
    4.     nth(2, tokens(clientid,'_')) as clientid,
    5.     s.name as name,
    6.     s.idx as idx
    8. FROM "t/#"

    另外，FOREACH 子句中也可以放多个表达式，只要最后一个表达式是指定要遍历的数组即可。比如我们将消息体改一下，sensors 外面多套一层 Object：

    1. {
    2.     "date": "2020-04-24",
    3.     "data": {
    4.         "sensors": [
    5.             {"name": "a", "idx":0},
    6.             {"name": "b", "idx":1},
    7.             {"name": "c", "idx":2}
    8.         ]
    9.     }
    10. }

    则 FOREACH 中可以在决定要遍历的数组之前把 data 选取出来：

    1. FOREACH
    2.     payload.data as data
    3.     data.sensors as s
    4. ...

    示例1: 将消息中 x 字段的值范围限定在 0~7 之间。

    1. SELECT
    2.   CASE WHEN payload.x < 0 THEN 0
    3.        WHEN payload.x > 7 THEN 7
    4.        ELSE payload.x
    5.   END as x
    6. FROM "t/#"

    假设消息为:

    1. {"x": 8}

    则上面的 SQL 输出为:

    数组操作语法举例

    1. SELECT
    2.   [1,2,3] as a
    3. FROM
    4.   "t/#"

    下标从 1 开始，上面的 SQL 输出为:

    1. {
    2.   "a": [1, 2, 3]
    3. }

    示例2: 从数组中取出第 N 个元素。下标为负数时，表示从数组的右边取:

    1. SELECT
    2.   [1,2,3] as a,
    3.   a[2] as b,
    4.   a[-2] as c
    5. FROM
    6.   "t/#"

    上面的 SQL 输出为:

    1. {
    2.   "b": 2,
    3.   "c": 2,
    4.   "a": [1, 2, 3]
    5. }

    示例3: 从 JSON 格式的 payload 中嵌套的获取值:

    1. SELECT
    2.   payload.data[1].id as id
    3. FROM
    4.   "t/#"

    假设消息为:

    1. {"data": [
    2.     {"id": 1, "name": "steve"},
    3.     {"id": 2, "name": "bill"}
    4. ]}

    则上面的 SQL 输出为:

    1. {"id": 1}

    示例4: 数组范围(range)操作:

    1. SELECT
    2.   [1..5] as a,
    3.   a[2..4] as b
    4. FROM
    5.   "t/#"

    上面的 SQL 输出为:

    1. {
    2.   "b": [2, 3, 4],
    3.   "a": [1, 2, 3, 4, 5]
    4. }

    示例5: 使用下标语法修改数组中的某个元素:

    1. SELECT
    2.   payload,
    3.   'STEVE' as payload.data[1].name
    4. FROM
    5.   "t/#"

    假设消息为:

    1. {"data": [
    2.     {"id": 1, "name": "steve"},
    3.     {"id": 2, "name": "bill"}
    4. ]}

    则上面的 SQL 输出为:

    1. {
    2.   "payload": {
    3.     "data": [
    4.       {"name": "STEVE", "id": 1},
    5.       {"name": "bill", "id": 2}
    6.     ]

    下一部分，规则 SQL 语句中可用的字段