数据转发

RealMQ基于消息流与设备事件，通过资源、规则和动作清晰、灵活、简洁的配置即可与用户的业务快速集成，提升用户易用性，降低业务系统与RealMQ的耦合度。

支持的数据转发列表

关系数据库	时序数据库	消息队列	WebServer	NoSQL
MySQL	TDengine	Kafka	WebHook	Redis
PostgreSQL		RocketMQ

数据流转的语法和例子

事件、主题 选择触发事件或者主题
筛选字段 对数据进行变换，并选择出感兴趣的字段
条件 筛选出来的某个字段施加条件过滤

事件列表

事件列表	说明
消息发布	客户端消息发布
消息发布前	客户端消息发布前
连接	客户端连接成功
连接断开	客户端断开连接成功
消息投递	消息投递给订阅者
消息投递丢弃	消息投递给订阅者时丢弃
消息应答	订阅者接收到消息的应答
消息丢弃	客户端发布消息时丢弃
订阅	客户端订阅主题
取消订阅	客户端取消订阅主题
鉴权结果	客户端访问控制鉴权

运算符号和比较符号

函数名	函数作用	返回值
`+`	加法，或字符串拼接	加和，或拼接之后的字符串
`-`	减法	差值
`*`	乘法	乘积
`/`	除法	商值
`div`	整数除法	整数商值
`mod`	取模	模
`=`	比较两者是否完全相等。可用于比较变量和主题	true/false
`=~`	比较主题(topic)是否能够匹配到主题过滤器(topic filter)。只能用于主题匹配	true/false
`>`	大于	true/false
`<`	小于	true/false
`<=`	小于等于	true/false
`>=`	大于等于	true/false
`<>`	不等于	true/false
`!=`	不等于	true/false

数据流转中可用的字段

消息发布

字段	解释
id	MQTT 消息 ID
clientid	消息来源 Client ID
username	消息来源用户名
payload	MQTT 消息体
peerhost	客户端的 IPAddress
topic	MQTT 主题
qos	MQTT 消息的 QoS
flags	MQTT 消息的 Flags
headers	MQTT 消息内部与流程处理相关的额外数据
pub_props	PUBLISH Properties (仅适用于 MQTT 5.0)
timestamp	事件触发时间 (ms)
publish_received_at	PUBLISH 消息到达 Broker 的时间 (ms)
node	事件触发所在节点

客户端连接成功

当客户端连接成功时触发规则

字段	解释
clientid	消息目的 Client ID
username	消息目的用户名
mountpoint	主题挂载点(主题前缀)
peername	终端的 IPAddress 和 Port
sockname	监听的 IPAddress 和 Port
proto_name	协议名字
proto_ver	协议版本
keepalive	MQTT 保活间隔
clean_start	MQTT clean_start
expiry_interval	MQTT Session 过期时间
is_bridge	是否为 MQTT bridge 连接
connected_at	终端连接完成时间 (ms)
conn_props	CONNECT Properties (仅适用于 MQTT 5.0)
timestamp	事件触发时间 (ms)
node	事件触发所在节点

客户端连接断开

当客户端连接断开时触发规则

字段	解释
reason	终端连接断开原因： normal：客户端主动断开 kicked：服务端踢出，通过 REST API keepalive_timeout: keepalive 超时 not_authorized: 认证失败，或者 acl_nomatch = disconnect 时没有权限的 Pub/Sub 会主动断开客户端 tcp_closed: 对端关闭了网络连接 discarded: 因为相同 ClientID 的客户端上线且设置 `clean_start = true` takeovered: 因为相同 ClientID 的客户端上线且设置 `clean_start = false` internal_error: 畸形报文或其他未知错误
clientid	消息目的 Client ID
username	消息目的用户名
peername	终端的 IPAddress 和 Port
sockname	监听的 IPAddress 和 Port
disconnected_at	终端连接断开时间 (ms)
disconn_props	DISCONNECT Properties (仅适用于 MQTT 5.0)
timestamp	事件触发时间 (ms)
node	事件触发所在节点

消息投递

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

字段	解释
id	MQTT 消息 ID
from_clientid	消息来源 Client ID
from_username	消息来源用户名
clientid	消息目的 Client ID
username	消息目的用户名
payload	MQTT 消息体
peerhost	客户端的 IPAddress
topic	MQTT 主题
qos	MQTT 消息的 QoS
flags	MQTT 消息的 Flags
pub_props	PUBLISH Properties (仅适用于 MQTT 5.0)
timestamp	事件触发时间 (ms)
publish_received_at	PUBLISH 消息到达 Broker 的时间 (ms)
node	事件触发所在节点

消息确认

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

字段	解释
id	MQTT 消息 ID
from_clientid	消息来源 Client ID
from_username	消息来源用户名
clientid	消息目的 Client ID
username	消息目的用户名
payload	MQTT 消息体
peerhost	客户端的 IPAddress
topic	MQTT 主题
qos	MQTT 消息的 QoS
flags	MQTT 消息的 Flags
pub_props	PUBLISH Properties (仅适用于 MQTT 5.0)
puback_props	PUBACK Properties (仅适用于 MQTT 5.0)
timestamp	事件触发时间 (ms)
publish_received_at	PUBLISH 消息到达 Broker 的时间 (ms)
node	事件触发所在节点

消息在转发的过程中被丢弃

当一条消息被丢弃时触发规则

字段	解释
id	MQTT 消息 ID
reason	消息丢弃原因，可能的原因： no_subscribers: 没有订阅者 receive_maximum_exceeded: awaiting_rel 队列已满 packet_identifier_inuse: 消息 ID 已被使用
clientid	消息来源 Client ID
username	消息来源用户名
payload	MQTT 消息体
peerhost	客户端的 IPAddress
topic	MQTT 主题
qos	MQTT 消息的 QoS
flags	MQTT 消息的 Flags
pub_props	PUBLISH Properties (仅适用于 MQTT 5.0)
timestamp	事件触发时间 (ms)
publish_received_at	PUBLISH 消息到达 Broker 的时间 (ms)
node	事件触发所在节点

消息在投递的过程中被丢弃

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

字段	解释
id	MQTT 消息 ID
reason	消息丢弃原因，可能的原因： queue_full: 消息队列已满(QoS>0) no_local: 不允许客户端接收自己发布的消息 expired: 消息或者会话过期 qos0_msg: QoS0 的消息因为消息队列已满被丢弃
from_clientid	消息来源 Client ID
from_username	消息来源用户名
clientid	消息目的 Client ID
username	消息目的用户名
payload	MQTT 消息体
peerhost	客户端的 IPAddress
topic	MQTT 主题
qos	MQTT 消息的 QoS
flags	MQTT 消息的 Flags
pub_props	PUBLISH Properties (仅适用于 MQTT 5.0)
timestamp	事件触发时间 (ms)
publish_received_at	PUBLISH 消息到达 Broker 的时间 (ms)
node	事件触发所在节点

鉴权结果

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

字段	解释
clientid	消息目的 Client ID
username	消息目的用户名
peerhost	客户端的 IPAddress
topic	MQTT 主题
action	publish or subscribe, 发布或者订阅事件
result	allow or deny，鉴权结果
is_cache	true or false，鉴权时数据的来源 is_cache为true时，鉴权数据来源于cache is_cache为false时，鉴权数据来源于插件
timestamp	事件触发时间 (ms)
node	事件触发所在节点

客户端订阅成功

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

字段	解释
clientid	消息目的 Client ID
username	消息目的用户名
peerhost	客户端的 IPAddress
topic	MQTT 主题
qos	MQTT 消息的 QoS
sub_props	SUBSCRIBE Properties (仅适用于 5.0)
timestamp	事件触发时间 (ms)
node	事件触发所在节点

取消客户端订阅

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

字段	解释
clientid	消息目的 Client ID
username	消息目的用户名
peerhost	客户端的 IPAddress
topic	MQTT 主题
qos	MQTT 消息的 QoS
unsub_props	UNSUBSCRIBE Properties (仅适用于 5.0)
timestamp	事件触发时间 (ms)
node	事件触发所在节点

示例

可用的函数

数据类型判断函数

函数名	函数作用	参数	返回值
is_null	判断变量是否为空值。注意：此函数无法判断 JSON 的 `null` 类型，请使用 `is_null_var` 替代	Data	Boolean 类型的数据。如果为空值(undefined) 则返回 true，否则返回 false
is_not_null	判断变量是否为非空值。注意：此函数无法判断 JSON 的 `null` 类型，请使用 `is_null_var` 替代	Data	Boolean 类型的数据。如果为空值(undefined) 则返回 false，否则返回 true
is_null_var	判断变量是否为空值	Data	Boolean 类型的数据。如果为空值(undefined) 则返回 true，否则返回 false
is_not_null_var	判断变量是否为非空值	Data	Boolean 类型的数据。如果为空值(undefined) 则返回 false，否则返回 true
is_str	判断变量是否为 String 类型	Data	Boolean 类型的数据。
is_bool	判断变量是否为 Boolean 类型	Data	Boolean 类型的数据。
is_int	判断变量是否为 Integer 类型	Data	Boolean 类型的数据。
is_float	判断变量是否为 Float 类型	Data	Boolean 类型的数据。
is_num	判断变量是否为数字类型，包括 Integer 和 Float 类型	Data	Boolean 类型的数据。
is_map	判断变量是否为 Map 类型	Data	Boolean 类型的数据。
is_array	判断变量是否为 Array 类型	Data	Boolean 类型的数据。

is_null(undefined_var) = true
is_null(mget('a', json_decode('{"a": null}'))) = false
is_not_null(1) = true
is_not_null(mget('a', json_decode('{"a": null}'))) = true

is_null_var(undefined_var) = true
is_null_var(mget('a', json_decode('{"a": null}'))) = true
is_not_null_var(1) = true
is_not_null_var(mget('a', json_decode('{"a": null}'))) = false

is_str(1) = false
is_str('val') = true
is_bool(true) = true
is_int(1) = true
is_float(1) = false
is_float(1.234) = true
is_num(2.3) = true
is_num('val') = false

数据类型转换函数

函数名	函数作用	参数	返回值
str	将数据转换为 String 类型	Data	String 类型的数据。无法转换将会导致 SQL 匹配失败
str_utf8	将数据转换为 UTF-8 String 类型	Data	UTF-8 String 类型的数据。无法转换将会导致 SQL 匹配失败
bool	将数据转换为 Boolean 类型	Data	Boolean 类型的数据。无法转换将会导致 SQL 匹配失败
int	将数据转换为整数类型	Data	整数类型的数据。无法转换将会导致 SQL 匹配失败
float	将数据转换为浮点型类型	Data	浮点型类型的数据。无法转换将会导致 SQL 匹配失败
float2str	将浮点型数字以指定精度转换为字符串	1. 浮点型数字 2. 精度	字符串
map	将数据转换为 Map 类型	Data	Map 类型的数据。无法转换将会导致 SQL 匹配失败

str(1234) = '1234'
str_utf8(1234) = '1234'
bool('true') = true
int('1234') = 1234
float('3.14') = 3.14
float2str(20.2, 10) = '20.2'
float2str(20.2, 17) = '20.19999999999999928'

注意浮点型转换为字符串的时候，输出结果会受到精度的影响，详情见：https://floating-point-gui.de/

字符串函数

函数名	函数作用	参数	返回值
lower	转为小写	1. 原字符串	小写字符串
upper	转为大写	1. 原字符串	大写字符串
trim	去掉左右空格	1. 原字符串	去掉空格后的字符串
ltrim	去掉左空格	1. 原字符串	去掉空格后的字符串
rtrim	去掉右空格	1. 原字符串	去掉空格后的字符串
reverse	字符串反转	1. 原字符串	翻转后的字符串
strlen	取字符串长度	1. 原字符串	整数值，字符长度
substr	取字符的子串	1. 原字符串 2. 起始位置. 注意: 下标从 0 开始	子串
substr	取字符的子串	1. 原字符串 2. 起始位置 3. 要取出的子串长度. 注意: 下标从 0 开始	子串
split	字符串分割	1. 原字符串 2. 分割符子串	分割后的字符串数组
split	字符串分割, 只查找左边第一个分隔符	1. 原字符串 2. 分割符子串 3. 'leading'	分割后的字符串数组
split	字符串分割, 只查找右边第一个分隔符	1. 原字符串 2. 分割符子串 3. 'trailing'	分割后的字符串数组
concat	字符串拼接	1. 左字符串 2. 右符子串	拼接后的字符串
tokens	字符串分解(按照指定字符串符分解)	1. 输入字符串 2. 分割符或字符串	分解后的字符串数组
tokens	字符串分解(按照指定字符串和换行符分解)	1. 输入字符串 2. 分割符或字符串 3. 'nocrlf'	分解后的字符串数组
sprintf	字符串格式化, 格式字符串的用法详见 https://erlang.org/doc/man/io.html#fwrite-1 里的 Format 部分	1. 格式字符串 2,3,4... 参数列表。参数个数不定	分解后的字符串数组
pad	字符串补足长度，补空格，从尾部补足	1. 原字符串 2. 字符总长度	补足后的字符串
pad	字符串补足长度，补空格，从尾部补足	1. 原字符串 2. 字符总长度 3. 'trailing'	补足后的字符串
pad	字符串补足长度，补空格，从两边补足	1. 原字符串 2. 字符总长度 3. 'both'	补足后的字符串
pad	字符串补足长度，补空格，从头部补足	1. 原字符串 2. 字符总长度 3. 'leading'	补足后的字符串
pad	字符串补足长度，补指定字符，从尾部补足	1. 原字符串 2. 字符总长度 3. 'trailing' 4. 指定用于补足的字符	补足后的字符串
pad	字符串补足长度，补指定字符，从两边补足	1. 原字符串 2. 字符总长度 3. 'both' 4. 指定用于补足的字符	补足后的字符串
pad	字符串补足长度，补指定字符，从头部补足	1. 原字符串 2. 字符总长度 3. 'leading' 4. 指定用于补足的字符	补足后的字符串
replace	替换字符串中的某子串，查找所有匹配子串替换	1. 原字符串 2. 要被替换的子串 3. 指定用于替换的字符串	替换后的字符串
replace	替换字符串中的某子串，查找所有匹配子串替换	1. 原字符串 2. 要被替换的子串 3. 指定用于替换的字符串 4. 'all'	替换后的字符串
replace	替换字符串中的某子串，从尾部查找第一个匹配子串替换	1. 原字符串 2. 要被替换的子串 3. 指定用于替换的字符串 4. 'trailing'	替换后的字符串
replace	替换字符串中的某子串，从头部查找第一个匹配子串替换	1. 原字符串 2. 要被替换的子串 3. 指定用于替换的字符串 4. 'leading'	替换后的字符串
regex_match	判断字符串是否与某正则表达式匹配	1. 原字符串 2. 正则表达式	true 或 false
regex_replace	替换字符串中匹配到某正则表达式的子串	1. 原字符串 2. 正则表达式 3. 指定用于替换的字符串	替换后的字符串
ascii	返回字符对应的 ASCII 码	1. 字符	整数值，字符对应的 ASCII 码
find	查找并返回字符串中的某个子串，从头部查找	1. 原字符串 2. 要查找的子串	查抄到的子串，如找不到则返回空字符串
find	查找并返回字符串中的某个子串，从头部查找	1. 原字符串 2. 要查找的子串 3. 'leading'	查抄到的子串，如找不到则返回空字符串
find	查找并返回字符串中的某个子串，从尾部查找	1. 原字符串 2. 要查找的子串 3. 'trailing'	查抄到的子串，如找不到则返回空字符串
join_to_string	拼接数组元素为字符串	1. 数组	拼接后的字符串，拼接符为逗号加空格 (`,` )
join_to_string	拼接数组元素为字符串	1. 拼接符 2. 数组	拼接后的字符串
join_to_sql_values_string	拼接数组元素为字符串，如果元素格式为字符串，将使用单引号包裹。此函数的字符串可以用作拼接 SQL 语句的 VALUES 子句	1. 数组	拼接后的字符串，拼接符为逗号加空格 (`,` )

lower('AbC') = 'abc'
lower('abc') = 'abc'

upper('AbC') = 'ABC'` `lower('ABC') = 'ABC'

trim(' hello  ') = 'hello'

ltrim(' hello  ') = 'hello  '

rtrim(' hello  ') = ' hello'

reverse('hello') = 'olleh'

strlen('hello') = 5

substr('abcdef', 2) = 'cdef'
substr('abcdef', 2, 3) = 'cde'

split('a/b/ c', '/') = ['a', 'b', ' c']
split('a/b/ c', '/', 'leading') = ['a', 'b/ c']
split('a/b/ c', '/', 'trailing') = ['a/b', ' c']

concat('a', '/bc') = 'a/bc'
'a' + '/bc' = 'a/bc'

tokens(' a/b/ c', '/') = [' a', 'b', ' c']
tokens(' a/b/ c', '/ ') = ['a', 'b', 'c']
tokens(' a/b/ c\n', '/ ') = ['a', 'b', 'c\n']
tokens(' a/b/ c\n', '/ ', 'nocrlf') = ['a', 'b', 'c']
tokens(' a/b/ c\r\n', '/ ', 'nocrlf') = ['a', 'b', 'c']

sprintf('hello, ~s!', 'steve') = 'hello, steve!'
sprintf('count: ~p~n', 100) = 'count: 100\n'

pad('abc', 5) = 'abc  '
pad('abc', 5, 'trailing') = 'abc  '
pad('abc', 5, 'both') = ' abc '
pad('abc', 5, 'leading') = '  abc'
pad('abc', 5, 'trailing', '*') = 'abc**'
pad('abc', 5, 'trailing', '*#') = 'abc*#*#'
pad('abc', 5, 'both', '*') = '*abc*'
pad('abc', 5, 'both', '*#') = '*#abc*#'
pad('abc', 5, 'leading', '*') = '**abc'
pad('abc', 5, 'leading', '*#') = '*#*#abc'

replace('ababef', 'ab', 'cd') = 'cdcdef'
replace('ababef', 'ab', 'cd', 'all') = 'cdcdef'
replace('ababef', 'ab', 'cd', 'trailing') = 'abcdef'
replace('ababef', 'ab', 'cd', 'leading') = 'cdabef'

regex_match('abc123', '[a-zA-Z1-9]*') = true

regex_replace('ab1cd3ef', '[1-9]', '[&]') = 'ab[1]cd[3]ef'
regex_replace('ccefacef', 'c+', ':') = ':efa:ef'

ascii('a') = 97

find('eeabcabcee', 'abc') = 'abcabcee'
find('eeabcabcee', 'abc', 'leading') = 'abcabcee'
find('eeabcabcee', 'abc', 'trailing') = 'abcee'

join_to_string(['a', 'b', 'c']) = 'a, b, c'
join_to_string('-', ['a', 'b', 'c']) = 'a-b-c'
join_to_sql_values_string(['a', 'b', 1]) = '\'a\', \'b\', 1'

Map 函数

函数名	函数作用	参数	返回值
map_new	构建一个空的 Map 类型数据	无	空的 Map 类型 (Erlang Map 类型：`#{}`，对应 JSON 中的对象类型 `{}`)
map_get	取 Map 中某个 Key 的值，如果没有则返回空值	1. Key 2. Map	Map 中某个 Key 的值。支持嵌套的 Key，比如 "a.b.c"
map_get	取 Map 中某个 Key 的值，如果没有则返回指定默认值	1. Key 2. Map 3. Default Value	Map 中某个 Key 的值。支持嵌套的 Key，比如 "a.b.c"
map_put	向 Map 中插入值	1. Key 2. Value 3. Map	插入后的 Map。支持嵌套的 Key，比如 "a.b.c"
mget	取 Map 中某个 Key 的值，如果没有则返回空值。作用与 map_get 相同，但不支持嵌套的 Key	1. Key 2. Map	Map 中某个 Key 的值
mget	取 Map 中某个 Key 的值，如果没有则返回指定默认值。作用与 map_get 相同，但不支持嵌套的 Key	1. Key 2. Map 3. Default Value	Map 中某个 Key 的值
mput	向 Map 中插入值。作用与 map_put 相同，但不支持嵌套的 Key	1. Key 2. Value 3. Map	插入后的 Map
map_keys	获取 Map 类型数据的所有键	Map	包含所有键的数组
map_values	获取 Map 类型数据的所有值	Map	包含所有值的数组
map_to_entries	将 Map 转换为 Key-Value 键值对数组	Map	`[#{key => Key}, #{value => Value}]` 格式的数组，对应的 JSON 类型为 `[{"key": Key}, {"value": Value}]`

map_new() = #{}
json_encode(map_new()) = '{}'
map_get('a', json_decode( '{ "a" : 1 }' )) = 1
map_get('b', json_decode( '{ "a" : 1 }' ), 2) = 2
map_get('a.b', json_decode( '{ "a" : {"b": 2} }' )) = 2
map_put('c', 1, map_new()) = #{c => 1}
map_put('c.d', 1, map_new()) = #{c => #{d => 1}}
json_encode(map_put('c.d', 1, map_new())) = '{"c":{"d":1}}'
mget('a.b', json_decode( '{ "a.b" : 1 }' )) = 1
mget('a.b', json_decode( '{ "a" : {"b": 2} }' )) = undefined
mput('c.d', 1, map_new()) = #{<<"c.d">> => 1}
json_encode(mput('c.d', 1, map_new())) = '{"c.d":1}'
json_encode(map_to_entries('{"a": 1, "b": 2}')) = '[{"value":1,"key":"a"}, {"value":2,"key":"b"}]'
map_keys(json_decode('{ "a" : 1, "b" : 2 }')) = ['a', 'b']
map_values(json_decode('{ "a" : 1, "b" : 2 }')) = [1, 2]

数组函数

函数名	函数作用	参数	返回值
nth	取第 n 个元素，下标从 1 开始	1. 起始位置 2. 原数组	第 n 个元素
length	获取数组的长度	1. 原数组	数组长度
sublist	取从第一个元素开始、长度为 len 的子数组。下标从 1 开始	1. 长度 len 2. 原数组	子数组
sublist	取从第 n 个元素开始、长度为 len 的子数组。下标从 1 开始	1. 起始位置 n 2. 长度 len 3. 原数组	子数组
first	取第 1 个元素。下标从 1 开始	1. 原数组	第 1 个元素
last	取最后一个元素。	1. 原数组	最后一个元素
contains	判断数据是否在数组里面	1. 数据 2. 原数组	Boolean 值

nth(2, [1,2,3,4]) = 2
length([1,2,3,4]) = 4
sublist(3, [1,2,3,4]) = [1,2,3,4]
sublist(1,2,[1,2,3,4]) = [1, 2]
first([1,2,3,4]) = 1
last([1,2,3,4]) = 4
contains(2, [1,2,3,4]) = true

哈希函数

函数名	函数功能	参数	返回值
md5	求 MD5 值	数据	MD5 值
sha	求 SHA 值	数据	SHA 值
sha256	求 SHA256 值	数据	SHA256 值

md5('some val') = '1b68352b3e9c2de52ffd322e30bffcc4'
sha('some val') = 'f85ba28ff5ea84a0cbfa118319acb0c5e58ee2b9'
sha256('some val') = '67f97635d8a0e064f60ba6e8846a0ac0be664f18f0c1dc6445cd3542d2b71993'

压缩解压缩函数

函数名	函数功能	参数	返回值
`gzip`	压缩数据，结果包含 gz 数据头和校验和	原始的二进制数据	压缩后的二进制数据
`gunzip`	解压缩数据，原始数据中包含 gz 数据头和校验和	压缩后的二进制数据	原始的二进制数据
`zip`	压缩数据，结果不包含 zlib 数据头和校验和	原始的二进制数据	压缩后的二进制数据
`unzip`	解压缩数据，原始数据中不包含 zlib 数据头和校验和	压缩后的二进制数据	原始的二进制数据
`zip_compress`	压缩数据，结果包含 zlib 数据头和校验和	原始的二进制数据	压缩后的二进制数据
`zip_uncompress`	解压缩数据，原始数据中包含 zlib 数据头和校验和	压缩后的二进制数据	原始的二进制数据

bin2hexstr(gzip('hello world')) = '1F8B0800000000000003CB48CDC9C95728CF2FCA49010085114A0D0B000000'
gunzip(hexstr2bin('1F8B0800000000000003CB48CDC9C95728CF2FCA49010085114A0D0B000000')) = 'hello world'

bin2hexstr(zip('hello world')) = 'CB48CDC9C95728CF2FCA490100'
unzip(hexstr2bin('CB48CDC9C95728CF2FCA490100')) = 'hello world'

bin2hexstr(zip_compress('hello world')) = '789CCB48CDC9C95728CF2FCA4901001A0B045D'
zip_uncompress(hexstr2bin('789CCB48CDC9C95728CF2FCA4901001A0B045D')) = 'hello world'

比特操作函数

函数名	函数功能	参数	返回值
subbits	从二进制数据的起始位置获取指定长度的比特位, 然后转换为无符号整型 (大端).	1. 二进制数据 2. 要获取的长度(bits)	无
subbits	从二进制数据的指定下标位置获取指定长度的比特位, 然后转换为无符号整型 (大端). 下标是从 1 开始的	1. 二进制数据 2. 起	subbits始位置的下标 3. 要获取的长度(bits)
subbits	从二进制数据的指定下标位置获取指定长度的比特位, 然后按照给定的参数转换为想要的数据类型. 下标是从 1 开始的.	1. 二进制数据 2. 起始位置的下标 3. 要获取的长度(bits) 4. 数据类型，可选值：'integer', 'float', 'bits' 5. 符号类型, 只对整型数据有效, 可选值：'unsigned', 'signed', 6. 大端还是小端, 只对整型数据有效, 可选值：'big', 'little'	获取到的数据

subbits('abc', 8) = 97
subbits('abc', 9, 8) = 98
subbits('abc', 17, 8) = 99
subbits('abc', 9, 16, 'integer', 'signed', 'big') = 25187
subbits('abc', 9, 16, 'integer', 'signed', 'little') = 25442

时间与日期函数

Function	Purpose	Parameters	Returned value
`now_timestamp`	返回当前时间的 Unix 秒级时间戳	-	Unix 时间戳
`now_timestamp`	指定时间单位，返回当前时间的 Unix 时间戳	1. 时间单位	Unix 时间戳
`now_rfc3339`	生成当前时间的 RFC3339 字符串，秒级	-	RFC3339 时间字符串
`now_rfc3339`	指定时间单位，生成当前时间的 RFC3339 字符串	1. 时间单位	RFC3339 时间字符串
`unix_ts_to_rfc3339`	将秒级 Unix 时间戳转换为 RFC3339 时间字符串	1. Unix 时间戳（秒）	RFC3339 时间字符串
`unix_ts_to_rfc3339`	指定时间单位，将 Unix 时间戳转换为 RFC3339 时间字符串	1. Unix 时间戳 2. 时间单位	RFC3339 时间字符串
`rfc3339_to_unix_ts`	将秒级 RFC3339 时间字符串转换为 Unix 时间戳	1. RFC3339 时间字符串	Unix 时间戳
`rfc3339_to_unix_ts`	指定时间单位，将 RFC3339 时间字符串转换为 Unix 时间戳	1. RFC3339 时间字符串 2. 时间单位	Unix 时间戳
`format_date`	时间戳转格式化时间	1. 时间戳精度（参考时间戳精度定义） 2. 时间偏移量（参考时间偏移量定义） 3. 日期格式（参考时间字符串编解码格式） 4. 时间戳（可选参数，默认为当前时间）	格式化时间字符串
`date_to_unix_ts`	格式化时间转时间戳	1. 时间戳精度（参考时间戳精度定义） 2. 时间偏移量（可选，未填写时，使用格式化时间字符串中的时间偏移量，参考时间偏移量定义） 3. 日期格式（参考时间字符串编解码格式） 4. 格式化时间字符串	Unix 时间戳

时间戳精度

时间戳精度名称	精度	示例
`second`	秒	1653557821
`millisecond`	毫秒	1653557852982
`microsecond`	微秒	1653557892926417
`nanosecond`	纳秒	1653557916474793000

时间字符串编解码格式

占位符	含义	取值范围
`%Y`	年	0000 - 9999
`%m`	月	01 - 12
`%d`	日	01 - 31
`%H`	时	00 - 12
`%M`	分	00 - 59
`%S`	秒	01 - 59
`%N`	纳秒	000000000 - 999999999
`%3N`	毫秒	000000 - 999999
`%6N`	微秒	000 - 000
`%z`	时间偏移量 [+\|-]HHMM	-1159 至 +1159
`%:z`	时间偏移量 [+\|-]HH:MM	-11:59 至 +11:59
`%::z`	时间偏移量 [+\|-]HH:MM:SS	-11:59:59 至 +11:59:59

时间偏移量定义

格式	含义	示例
`z`	UTC Zulu 时间	固定值 `+00:00`
`Z`	UTC Zulu 时间，与 `z` 相同	固定值 `+00:00`
`local`	系统时间	自动获取，例如北京时间 `+08:00` Zulu 时间 `+00:00` 瑞典斯德哥尔摩时间 `+02:00` 洛杉矶时间 `-08:00`
`[+	-]HHMM`	`%z` 格式
`[+	-]HH:MM`	`%:z` 格式
`[+	-]HH:MM:SS`	`%::z` 格式
integer()	时间偏移量秒数	北京时间 28800 Zulu 时间 0 瑞典斯德哥尔摩时间 7200 洛杉矶时间 -28800

now_timestamp() = 1650874276
now_timestamp('millisecond') = 1650874318331
now_rfc3339() = '2022-04-25T16:08:41+08:00'
now_rfc3339('millisecond') = '2022-04-25T16:10:10.652+08:00'
unix_ts_to_rfc3339(1650874276) = '2022-04-25T16:11:16+08:00'
unix_ts_to_rfc3339(1650874318331, 'millisecond') = '2022-04-25T16:11:58.331+08:00'
rfc3339_to_unix_ts('2022-04-25T16:11:16+08:00') = 1650874276
rfc3339_to_unix_ts('2022-04-25T16:11:58.331+08:00', 'millisecond') = 1650874318331
format_date('second', '+0800', '%Y-%m-%d %H:%M:%S%:z', 1653561612) = '2022-05-26 18:40:12+08:00'
format_date('second', 'local', '%Y-%m-%d %H:%M:%S%:z') = "2022-05-26 18:48:01+08:00"
format_date('second', 0, '%Y-%m-%d %H:%M:%S%:z') = '2022-05-26 10:42:41+00:00'
date_to_unix_ts('second', '%Y-%m-%d %H:%M:%S%:z', '2022-05-26 18:40:12+08:00') = 1653561612
date_to_unix_ts('second', 'local', '%Y-%m-%d %H-%M-%S', '2022-05-26 18:40:12') = 1653561612
date_to_unix_ts('second', '%Y-%m-%d %H-%M-%S', '2022-05-26 10:40:12') = 1653561612

Function	Purpose	Parameters	Returned value
`mongo_date`	生成当前时间的 mongodb ISODate 类型	-	ISODate 类型的时间
`mongo_date`	生成指定 Unix 时间戳的 mongodb ISODate 类型，毫秒级	1. 毫秒级 Unix 时间戳	ISODate 类型的时间
`mongo_date`	指定时间单位，生成指定 Unix 时间戳的 mongodb ISODate 类型	1. Unix 时间戳 2. 时间单位	ISODate 类型的时间

时间单位可以是以下其中之一： 'second', 'millisecond', 'microsecond' or 'nanosecond'.

mongo_date() = 'ISODate("2012-12-19T06:01:17.171Z")'
mongo_date(timestamp) = 'ISODate("2012-12-19T06:01:17.171Z")'
mongo_date(timestamp, 'millisecond') = 'ISODate("2012-12-19T06:01:17.171Z")'

数据转发 ​

支持的数据转发列表 ​

数据流转的语法和例子 ​

事件列表 ​

运算符号和比较符号 ​

数据流转中可用的字段 ​

消息发布 ​

客户端连接成功 ​

客户端连接断开 ​

消息投递 ​

消息确认 ​

消息在转发的过程中被丢弃 ​

消息在投递的过程中被丢弃 ​

鉴权结果 ​

客户端订阅成功 ​

取消客户端订阅 ​

可用的函数 ​

数据类型判断函数 ​

数据类型转换函数 ​

字符串函数 ​

Map 函数 ​

数组函数 ​

哈希函数 ​

压缩解压缩函数 ​

比特操作函数 ​

时间与日期函数 ​