emqx(https://github.com/emqx/emqx) 数据处理插件(实验阶段)
提供 API 访问各种数据库、其他消息中间件以及各种协议网关。提供各种 emqx 内部操作 API。 外加一套简单、灵活的数据处理 DSL。
数据源可能有:
- mqtt
- gateways: lwm2m 等各种协议网关
- http api
- 订阅自其他 broker 的消息
数据输出可能包括:
- 数据库和其他 broker
- HTTP 回调
- MQTT
- gateways
数据源提供 triggers 数据输出提供 send/query functions.
DSL 需要选择一个或者多个 trigger 作为输入,调用数据库和中间件等提供的 function 做消息转移。
DSL 最好提供类似 switch 语言的 playground 那种 debug 能力。
I. 场景举例
当收到主题为 't/1' 的 MQTT 消息,首先做 Avro 解码。 解码后发送给 lwm2m 设备(指令下发),然后根据消息类型分别发送到 mysql 和 kafka 做消息备份。
task:
## define a function sending data according to the data type
func send_data('write', data) {
## call the 'produce' API of a kafka resource 'res_4'
kafka.produce('res_4', 't/1', data)
}
func send_data('read', data) {
mysql.query('res_3', 'insert into table values (data.x, data.y))
}
trigger 'mqtt:publish' when topic = 't/1' {
## decode it by avro schema
data = schema_registry.decode(avro, 'schema-1' payload)
## log it out for debugging
emqx.log('info', data)
## send to the lwm2m device with endpoint name = 'client-a'
## call the 'send' API of the lwm2m resource named 'res_2'
lwm2m.send('res_2', 'client-a', '/3/0/1', data)
## call the self-defined function
send_data(data.type)
}
II. DSL 功能
- 语法
基本:
- 数据类型: strings, numbers, maps, lists
- 数据操作: 下标读取,Map 插入等等
- 函数定义:
- 函数调用:
- 计算: 数学计算
- 条件分支: case
- 循环: while, foreach
高级:
- pipe
- list comprehension
- stdlibs
- strings
- binary
- crypto
- time
- jq (json string processing)
- emqx (getting connection info, publishing messages, etc. All the functions that are provided by HTTP API should be supported here)
- triggers
-
mqtt.client_connected
-
mqtt.publish
-
mqtt.auth_request ...
-
coap.put
-
coap.post
-
coap.auth_request ...
-
lwm2m.registered
-
lwm2m.notify
-
lwm2m.auth_request ...
-
kafka.msg_in ...
-
timers.interval
-
timers.schedule_at ...
- function provided by plugins/gateways/sinks
- coap:res_1.send('post', peer_ip_port, '/path').
- lwm2m:res_2.send('client-a', '/3/0/1', payload).
- mysql:res_3.query('SELECT * FROM table', timeout).
- kafka:res_4.produce('t/1', payload)
- in memory key-value db
- kv.put(key, value)
- kv.get(key, default)