EMQX Dedicated REST API (v5)

Download OpenAPI specification:Download

EMQX Dedicated REST API

Authentication

获取内置数据库认证器中的指定用户数据。

Authorizations:

basicAuthbearerAuth

path Parameters

user_id

required

string

用户 ID，根据设置的账号类型不同，可以是 username 或 clientid

Responses

Response samples

200
404

Content type

application/json

Example

Regular user

{"user_id": "user1"
}

更新内置数据库认证器中的指定用户数据。

Authorizations:

basicAuthbearerAuth

path Parameters

user_id

required

string

用户 ID，根据设置的账号类型不同，可以是 username 或 clientid

Request Body schema: application/json

password required	string
is_superuser	boolean Default: false

Responses

Request samples

Payload

Content type

application/json

Example

Update regular user

{"password": "******"
}

Response samples

200
400
404

Content type

application/json

{"regular_user": {"value": {"user_id": "user1"
},
"summary": "Regular user"
},
"super_user": {"value": {"is_superuser": true,
"user_id": "user2"
},
"summary": "Superuser"
}
}

删除内置数据库认证器中的指定用户数据。

Authorizations:

basicAuthbearerAuth

path Parameters

user_id

required

string

用户 ID，根据设置的账号类型不同，可以是 username 或 clientid

Responses

Response samples

404

Content type

application/json

{"code": "NOT_FOUND",
"message": "string"
}

为内置数据库认证器导入用户数据。

Authorizations:

basicAuthbearerAuth

query Parameters

type

required

string

Enum: "plain" "hash"

Example: type=hash

The import file template type, enum with plain,hash

Request Body schema: application/json

Import body

object

Responses

Request samples

Payload

Content type

application/json

[{"is_superuser": true,
"password": "password1",
"user_id": "user1"
},
{"is_superuser": false,
"password": "password2",
"user_id": "user2"
}
]

Response samples

400
404

Content type

application/json

{"code": "BAD_REQUEST",
"message": "string"
}

获取内置数据库认证器中的用户列表。

Authorizations:

basicAuthbearerAuth

query Parameters

page	integer >= 1 Default: 1 Example: page=1 Page number of the results to fetch.
limit	integer [ 1 .. 10000 ] Default: 100 Example: limit=50 Results per page(max 10000)
like_user_id	string 使用用户 ID （username 或 clientid）模糊查询。
is_superuser	boolean 是否是超级用户。

Responses

Response samples

200
404

Content type

application/json

{"data": [{"user_id": "user1"
},
{"is_superuser": true,
"user_id": "user2"
}
],
"meta": {"count": 300,
"limit": 20,
"page": 0
}
}

向内置数据库认证器添加用户数据。

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

user_id required	string
password required	string
is_superuser	boolean Default: false

Responses

Request samples

Payload

Content type

application/json

Example

Regular user

{"password": "******",
"user_id": "user1"
}

Response samples

201
400
404

Content type

application/json

Example

Regular user

{"user_id": "user1"
}

Authorization

显示用户名规则列表。

Authorizations:

basicAuthbearerAuth

query Parameters

page	integer >= 1 Default: 1 Example: page=1 Page number of the results to fetch.
limit	integer [ 1 .. 10000 ] Default: 100 Example: limit=50 Results per page(max 10000)
like_username	string 使用字串匹配模糊搜索用户名

Responses

Response samples

200

Content type

application/json

{"data": [{"rules": [{"action": "publish",
"permission": "allow",
"topic": "test/topic/1"
},
{"action": "subscribe",
"permission": "allow",
"topic": "test/topic/2"
},
{"action": "all",
"permission": "deny",
"topic": "eq test/#"
},
{"action": "publish",
"retain": "true",
"permission": "allow",
"topic": "test/topic/3",
"qos": ["1"
]
},
{"action": "publish",
"retain": "all",
"permission": "allow",
"topic": "test/topic/4",
"qos": ["0",
"1",
"2"
]
}
],
"username": "user1"
}
],
"meta": {"count": 1,
"limit": 100,
"page": 1
}
}

为指定的用户添加规则。

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

Array

	Array of objects (emqx_authz_api_mnesia.rule_item)
username required	string 用户名

Responses

Request samples

Payload

Content type

application/json

[{"rules": [{"action": "publish",
"permission": "allow",
"topic": "test/topic/1"
},
{"action": "subscribe",
"permission": "allow",
"topic": "test/topic/2"
},
{"action": "all",
"permission": "deny",
"topic": "eq test/#"
},
{"action": "publish",
"retain": "true",
"permission": "allow",
"topic": "test/topic/3",
"qos": ["1"
]
},
{"action": "publish",
"retain": "all",
"permission": "allow",
"topic": "test/topic/4",
"qos": ["0",
"1",
"2"
]
}
],
"username": "user1"
}
]

Response samples

400
409

Content type

application/json

{"code": "BAD_REQUEST",
"message": "string"
}

显示客户端规则列表。

Authorizations:

basicAuthbearerAuth

query Parameters

page	integer >= 1 Default: 1 Example: page=1 Page number of the results to fetch.
limit	integer [ 1 .. 10000 ] Default: 100 Example: limit=50 Results per page(max 10000)
like_clientid	string 使用字串匹配模糊搜索客户端 ID

Responses

Response samples

200

Content type

application/json

{"data": [{"rules": [{"action": "publish",
"permission": "allow",
"topic": "test/topic/1"
},
{"action": "subscribe",
"permission": "allow",
"topic": "test/topic/2"
},
{"action": "all",
"permission": "deny",
"topic": "eq test/#"
},
{"action": "publish",
"retain": "true",
"permission": "allow",
"topic": "test/topic/3",
"qos": ["1"
]
},
{"action": "publish",
"retain": "all",
"permission": "allow",
"topic": "test/topic/4",
"qos": ["0",
"1",
"2"
]
}
],
"clientid": "client1"
}
],
"meta": {"count": 1,
"limit": 100,
"page": 1
}
}

为指定客户端添加规则。

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

Array

	Array of objects (emqx_authz_api_mnesia.rule_item)
clientid required	string 客户端 ID

Responses

Request samples

Payload

Content type

application/json

[{"rules": [{"action": "publish",
"permission": "allow",
"topic": "test/topic/1"
},
{"action": "subscribe",
"permission": "allow",
"topic": "test/topic/2"
},
{"action": "all",
"permission": "deny",
"topic": "eq test/#"
},
{"action": "publish",
"retain": "true",
"permission": "allow",
"topic": "test/topic/3",
"qos": ["1"
]
},
{"action": "publish",
"retain": "all",
"permission": "allow",
"topic": "test/topic/4",
"qos": ["0",
"1",
"2"
]
}
],
"clientid": "client1"
}
]

Response samples

400

Content type

application/json

{"code": "BAD_REQUEST",
"message": "string"
}

获取指定客户端的规则

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

Example: client1

客户端 ID

Responses

Response samples

200
404

Content type

application/json

{"rules": [{"action": "publish",
"permission": "allow",
"topic": "test/topic/1"
},
{"action": "subscribe",
"permission": "allow",
"topic": "test/topic/2"
},
{"action": "all",
"permission": "deny",
"topic": "eq test/#"
},
{"action": "publish",
"retain": "true",
"permission": "allow",
"topic": "test/topic/3",
"qos": ["1"
]
},
{"action": "publish",
"retain": "all",
"permission": "allow",
"topic": "test/topic/4",
"qos": ["0",
"1",
"2"
]
}
],
"clientid": "client1"
}

设置指定客户端的规则

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

Example: client1

客户端 ID

Request Body schema: application/json

	Array of objects (emqx_authz_api_mnesia.rule_item)
clientid required	string 客户端 ID

Responses

Request samples

Payload

Content type

application/json

{"rules": [{"action": "publish",
"permission": "allow",
"topic": "test/topic/1"
},
{"action": "subscribe",
"permission": "allow",
"topic": "test/topic/2"
},
{"action": "all",
"permission": "deny",
"topic": "eq test/#"
},
{"action": "publish",
"retain": "true",
"permission": "allow",
"topic": "test/topic/3",
"qos": ["1"
]
},
{"action": "publish",
"retain": "all",
"permission": "allow",
"topic": "test/topic/4",
"qos": ["0",
"1",
"2"
]
}
],
"clientid": "client1"
}

Response samples

400

Content type

application/json

{"code": "BAD_REQUEST",
"message": "string"
}

删除指定客户端的规则

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

Example: client1

客户端 ID

Responses

Response samples

400
404

Content type

application/json

{"code": "BAD_REQUEST",
"message": "string"
}

获取指定用户名规则

Authorizations:

basicAuthbearerAuth

path Parameters

username

required

string

Example: user1

用户名

Responses

Response samples

200
404

Content type

application/json

{"rules": [{"action": "publish",
"permission": "allow",
"topic": "test/topic/1"
},
{"action": "subscribe",
"permission": "allow",
"topic": "test/topic/2"
},
{"action": "all",
"permission": "deny",
"topic": "eq test/#"
},
{"action": "publish",
"retain": "true",
"permission": "allow",
"topic": "test/topic/3",
"qos": ["1"
]
},
{"action": "publish",
"retain": "all",
"permission": "allow",
"topic": "test/topic/4",
"qos": ["0",
"1",
"2"
]
}
],
"username": "user1"
}

设置指定用户名的规则

Authorizations:

basicAuthbearerAuth

path Parameters

username

required

string

Example: user1

用户名

Request Body schema: application/json

	Array of objects (emqx_authz_api_mnesia.rule_item)
username required	string 用户名

Responses

Request samples

Payload

Content type

application/json

{"rules": [{"action": "publish",
"permission": "allow",
"topic": "test/topic/1"
},
{"action": "subscribe",
"permission": "allow",
"topic": "test/topic/2"
},
{"action": "all",
"permission": "deny",
"topic": "eq test/#"
},
{"action": "publish",
"retain": "true",
"permission": "allow",
"topic": "test/topic/3",
"qos": ["1"
]
},
{"action": "publish",
"retain": "all",
"permission": "allow",
"topic": "test/topic/4",
"qos": ["0",
"1",
"2"
]
}
],
"username": "user1"
}

Response samples

400

Content type

application/json

{"code": "BAD_REQUEST",
"message": "string"
}

删除指定用户名的规则

Authorizations:

basicAuthbearerAuth

path Parameters

username

required

string

Example: user1

用户名

Responses

Response samples

400
404

Content type

application/json

{"code": "BAD_REQUEST",
"message": "string"
}

清除内置数据库中的所有类型（'users' 、'clients' 、'all'）的所有规则

Authorizations:

basicAuthbearerAuth

Responses

Response samples

400

Content type

application/json

{"code": "BAD_REQUEST",
"message": "string"
}

列出适用于所有客户端规则（即 'all' 规则）。

Authorizations:

basicAuthbearerAuth

Responses

Response samples

200

Content type

application/json

{"rules": [{"action": "publish",
"permission": "allow",
"topic": "test/topic/1"
},
{"action": "subscribe",
"permission": "allow",
"topic": "test/topic/2"
},
{"action": "all",
"permission": "deny",
"topic": "eq test/#"
},
{"action": "publish",
"retain": "true",
"permission": "allow",
"topic": "test/topic/3",
"qos": ["1"
]
},
{"action": "publish",
"retain": "all",
"permission": "allow",
"topic": "test/topic/4",
"qos": ["0",
"1",
"2"
]
}
]
}

删除 'all' 规则

Authorizations:

basicAuthbearerAuth

Responses

创建或更新适用于所有客户端的规则（即 'all' 规则）。

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

Array of objects (emqx_authz_api_mnesia.rule_item)

Responses

Request samples

Payload

Content type

application/json

{"rules": [{"action": "publish",
"permission": "allow",
"topic": "test/topic/1"
},
{"action": "subscribe",
"permission": "allow",
"topic": "test/topic/2"
},
{"action": "all",
"permission": "deny",
"topic": "eq test/#"
},
{"action": "publish",
"retain": "true",
"permission": "allow",
"topic": "test/topic/3",
"qos": ["1"
]
},
{"action": "publish",
"retain": "all",
"permission": "allow",
"topic": "test/topic/4",
"qos": ["0",
"1",
"2"
]
}
]
}

Response samples

400

Content type

application/json

{"code": "BAD_REQUEST",
"message": "string"
}

Banned

列出目前所有被封禁的客户端 ID、用户名和 IP 地址。

Authorizations:

basicAuthbearerAuth

query Parameters

page	integer >= 1 Default: 1 Example: page=1 Page number of the results to fetch.
limit	integer [ 1 .. 10000 ] Default: 100 Example: limit=50 Results per page(max 10000)

Responses

Response samples

200

Content type

application/json

{"data": [{"as": "username",
"who": "Banned name",
"by": "mgmt_api",
"reason": "Too many requests",
"at": "2021-10-25T21:48:47+08:00",
"until": "2021-10-25T21:53:47+08:00"
}
],
"meta": {"page": 1,
"limit": 50,
"count": 0,
"hasnext": true
}
}

清除全部黑名单数据。

Authorizations:

basicAuthbearerAuth

Responses

添加一个客户端 ID、用户名或者 IP 地址到黑名单。

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

as required	string Enum: "clientid" "username" "peerhost" "clientid_re" "username_re" "peerhost_net" 封禁方式，可以通过客户端 ID、用户名或者 IP 地址等方式进行封禁。
who required	string 封禁对象，具体的客户端 ID、用户名或者 IP 地址。
by	string 封禁的发起者。
reason	string 封禁原因，记录当前对象被封禁的原因。
	integer or string 封禁的起始时间，格式为 rfc3339，默认为发起操作的时间。
	integer or string 封禁的结束时间，格式为 rfc3339，默认值为发起操作的时间 + 1 年。

Responses

Request samples

Payload

Content type

application/json

{"as": "username",
"who": "Banned name",
"by": "mgmt_api",
"reason": "Too many requests",
"at": "2021-10-25T21:48:47+08:00",
"until": "2021-10-25T21:53:47+08:00"
}

Response samples

200
400

Content type

application/json

{"data": [{"as": "username",
"who": "Banned name",
"by": "mgmt_api",
"reason": "Too many requests",
"at": "2021-10-25T21:48:47+08:00",
"until": "2021-10-25T21:53:47+08:00"
}
]
}

将一个客户端 ID、用户名或者 IP 地址从黑名单中删除。

Authorizations:

basicAuthbearerAuth

path Parameters

as required	string Enum: "clientid" "username" "peerhost" "clientid_re" "username_re" "peerhost_net" Example: username 封禁方式，可以通过客户端 ID、用户名或者 IP 地址等方式进行封禁。
who required	string Example: Badass 封禁对象，具体的客户端 ID、用户名或者 IP 地址。

Responses

Response samples

404

Content type

application/json

{"code": "NOT_FOUND",
"message": "string"
}

Clients

----

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

Array

string

Responses

Request samples

Payload

Content type

application/json

["emqx_clientid_985bb09d",
"emqx_clientid_211cc01c"
]

获取指定客户端的授权缓存

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

Responses

Response samples

200
404

Content type

application/json

{"access": "publish",
"result": "allow",
"topic": "testtopic/1",
"updated_time": 1687850712989
}

清除指定客户端的授权缓存

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

Responses

Response samples

404

Content type

application/json

{"code": "CLIENTID_NOT_FOUND",
"message": "string"
}

获取指定客户端的详细信息

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

Responses

Response samples

200
404

Content type

application/json

{"heap_size": 610,
"send_msg.dropped.expired": 0,
"send_oct": 31,
"recv_msg.qos1": 0,
"is_persistent": false,
"send_pkt": 4,
"clean_start": true,
"inflight_cnt": 0,
"node": "emqx@127.0.0.1",
"send_msg.dropped.queue_full": 0,
"awaiting_rel_cnt": 0,
"inflight_max": 32,
"created_at": "2024-01-01T12:34:56.789+08:00",
"subscriptions_cnt": 1,
"mailbox_len": 0,
"send_cnt": 4,
"connected": true,
"ip_address": "127.0.0.1",
"awaiting_rel_max": 100,
"recv_msg.qos2": 0,
"proto_ver": 5,
"mountpoint": "null",
"proto_name": "MQTT",
"port": 52571,
"connected_at": "2024-01-01T12:34:56.789+08:00",
"enable_authn": true,
"expiry_interval": 0,
"username": null,
"recv_msg": 0,
"recv_oct": 49,
"send_msg.dropped.too_large": 0,
"keepalive": 60,
"send_msg.qos1": 0,
"send_msg.qos2": 0,
"recv_msg.qos0": 0,
"send_msg.qos0": 0,
"subscriptions_max": "infinity",
"mqueue_max": 1000,
"mqueue_dropped": 0,
"clientid": "01",
"is_bridge": false,
"peerport": 52571,
"send_msg": 0,
"listener": "tcp:default",
"recv_cnt": 4,
"recv_pkt": 4,
"recv_msg.dropped": 0,
"send_msg.dropped": 0,
"recv_msg.dropped.await_pubrel_timeout": 0,
"reductions": 6836,
"mqueue_len": 0
}

踢出指定客户端

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

Responses

Response samples

404

Content type

application/json

{"code": "CLIENTID_NOT_FOUND",
"message": "string"
}

批量取消订阅

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

Request Body schema: application/json

Array

topic

string

Topic

Responses

Request samples

Payload

Content type

application/json

[{"topic": "testtopic/#"
}
]

Response samples

404

Content type

application/json

{"code": "CLIENTID_NOT_FOUND",
"message": "string"
}

批量订阅

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

Request Body schema: application/json

Array

topic required	string Topic
qos	integer [ 0 .. 2 ] Default: 0 QoS
nl	integer Default: 0 No Local
rap	integer Default: 0 Retain as Published
rh	integer Default: 0 Retain Handling

Responses

Request samples

Payload

Content type

application/json

[{"topic": "testtopic/#",
"qos": 0,
"nl": 0,
"rap": 0,
"rh": 0
}
]

Response samples

200
404

Content type

application/json

[{"node": "emqx@127.0.0.1",
"topic": "testtopic/1",
"clientid": "emqx_clientid_xx128cdhfc",
"qos": 0,
"nl": 0,
"rap": 0,
"rh": 0
}
]

获取客户端消息队列中的消息。

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

query Parameters

payload	string Default: "base64" Enum: "none" "base64" "plain" Client's inflight/mqueue messages payload encoding. If set to `none`, no payload is returned in the response.
max_payload_bytes	string Default: "1MB" Example: max_payload_bytes=32MB Client's inflight/mqueue messages payload limit. The total payload size of all messages in the response will not exceed this value. Messages beyond the limit will be silently omitted in the response. The only exception to this rule is when the first message payload is already larger than the limit. In this case, the first message will be returned in the response.
	string or string or string Example: position=none An opaque token that can then be in subsequent requests to get the next chunk of results: "?position={prev_response.meta.position}" It is used instead of "page" parameter to traverse highly volatile data. Can be omitted or set to "none" to get the first chunk of data.
limit	integer [ 1 .. 10000 ] Default: 100 Example: limit=50 Results per page(max 10000)

Responses

Response samples

200
400
404
501

Content type

application/json

{"data": [{"from_clientid": "mqttx_59ac0a87",
"from_username": "test-user",
"mqueue_priority": 0,
"msgid": "000611F460D57FA9F44500000D360002",
"payload": "eyJmb28iOiAiYmFyIn0=",
"publish_at": 1709055346487,
"qos": 0,
"topic": "t/test"
}
],
"meta": {"count": 100,
"last": "1710785444656449826_10"
}
}

获取客户端未确认消息。

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

query Parameters

payload	string Default: "base64" Enum: "none" "base64" "plain" Client's inflight/mqueue messages payload encoding. If set to `none`, no payload is returned in the response.
max_payload_bytes	string Default: "1MB" Example: max_payload_bytes=32MB Client's inflight/mqueue messages payload limit. The total payload size of all messages in the response will not exceed this value. Messages beyond the limit will be silently omitted in the response. The only exception to this rule is when the first message payload is already larger than the limit. In this case, the first message will be returned in the response.
	string or string or string Example: position=none An opaque token that can then be in subsequent requests to get the next chunk of results: "?position={prev_response.meta.position}" It is used instead of "page" parameter to traverse highly volatile data. Can be omitted or set to "none" to get the first chunk of data.
limit	integer [ 1 .. 10000 ] Default: 100 Example: limit=50 Results per page(max 10000)

Responses

Response samples

200
400
404
501

Content type

application/json

{"data": [{"from_clientid": "mqttx_59ac0a87",
"from_username": "test-user",
"msgid": "000611F460D57FA9F44500000D360002",
"payload": "eyJmb28iOiAiYmFyIn0=",
"publish_at": 1709055346487,
"qos": 0,
"topic": "t/test"
}
],
"meta": {"count": 100,
"last": "1710785444656449826"
}
}

取消订阅

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

Request Body schema: application/json

topic

string

Topic

Responses

Request samples

Payload

Content type

application/json

{"topic": "testtopic/#"
}

Response samples

404

Content type

application/json

{"code": "CLIENTID_NOT_FOUND",
"message": "string"
}

获取指定客户端的订阅列表

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

Responses

Response samples

200
404

Content type

application/json

[{"node": "emqx@127.0.0.1",
"topic": "testtopic/1",
"clientid": "emqx_clientid_xx128cdhfc",
"qos": 0,
"nl": 0,
"rap": 0,
"rh": 0
}
]

列出客户端

Authorizations:

basicAuthbearerAuth

query Parameters

page	integer >= 1 Default: 1 Example: page=1 Page number of the results to fetch.
limit	integer [ 1 .. 10000 ] Default: 100 Example: limit=50 Results per page(max 10000)
node	string Example: node=emqx@127.0.0.1 Node name
username	Array of strings User name, multiple values can be specified by repeating the parameter: username=u1&username=u2
ip_address	string Example: ip_address=127.0.0.1 Client's IP address
conn_state	string Enum: "connected" "idle" "disconnected" The current connection status of the client, the possible values are connected,idle,disconnected
clean_start	boolean Whether the client uses a new session
proto_ver	string Client protocol version
like_clientid	string Fuzzy search `clientid` as substring
like_username	string Fuzzy search `username` as substring
	integer or string Search client session creation time by greater than or equal method, rfc3339 or timestamp(millisecond)
	integer or string Search client session creation time by less than or equal method, rfc3339 or timestamp(millisecond)
	integer or string Search client connection creation time by greater than or equal method, rfc3339 or timestamp(epoch millisecond)
	integer or string Search client connection creation time by less than or equal method, rfc3339 or timestamp(millisecond)
clientid	Array of strings Client ID, multiple values can be specified by repeating the parameter: clientid=c1&clientid=c2
	Array of strings or string Default: "all" Comma separated list of client fields to return in the response

Responses

Response samples

200
400

Content type

application/json

{"data": [{"heap_size": 610,
"send_msg.dropped.expired": 0,
"send_oct": 31,
"recv_msg.qos1": 0,
"is_persistent": false,
"send_pkt": 4,
"clean_start": true,
"inflight_cnt": 0,
"node": "emqx@127.0.0.1",
"send_msg.dropped.queue_full": 0,
"awaiting_rel_cnt": 0,
"inflight_max": 32,
"created_at": "2024-01-01T12:34:56.789+08:00",
"subscriptions_cnt": 1,
"mailbox_len": 0,
"send_cnt": 4,
"connected": true,
"ip_address": "127.0.0.1",
"awaiting_rel_max": 100,
"recv_msg.qos2": 0,
"proto_ver": 5,
"mountpoint": "null",
"proto_name": "MQTT",
"port": 52571,
"connected_at": "2024-01-01T12:34:56.789+08:00",
"enable_authn": true,
"expiry_interval": 0,
"username": null,
"recv_msg": 0,
"recv_oct": 49,
"send_msg.dropped.too_large": 0,
"keepalive": 60,
"send_msg.qos1": 0,
"send_msg.qos2": 0,
"recv_msg.qos0": 0,
"send_msg.qos0": 0,
"subscriptions_max": "infinity",
"mqueue_max": 1000,
"mqueue_dropped": 0,
"clientid": "01",
"is_bridge": false,
"peerport": 52571,
"send_msg": 0,
"listener": "tcp:default",
"recv_cnt": 4,
"recv_pkt": 4,
"recv_msg.dropped": 0,
"send_msg.dropped": 0,
"recv_msg.dropped.await_pubrel_timeout": 0,
"reductions": 6836,
"mqueue_len": 0
}
],
"meta": {"count": 1,
"hasnext": false,
"limit": 50,
"page": 1
}
}

Subscriptions

获取集群的订阅列表。

Authorizations:

basicAuthbearerAuth

query Parameters

page	integer >= 1 Default: 1 Example: page=1 Page number of the results to fetch.
limit	integer [ 1 .. 10000 ] Default: 100 Example: limit=50 Results per page(max 10000)
node	string Example: node=emqx@127.0.0.1 Node name
clientid	string Client ID
qos	integer [ 0 .. 2 ] Example: qos=0 QoS
topic	string Topic, url encoding
match_topic	string Match topic string, url encoding
share_group	string Shared subscription group name

Responses

Response samples

200
400
500

Content type

application/json

[{"node": "emqx@127.0.0.1",
"topic": "testtopic/1",
"clientid": "emqx_clientid_xx128cdhfc",
"qos": 0,
"nl": 0,
"rap": 0,
"rh": 0
}
]

Publish

Publish a batch of messages

批量发布多条消息。

可能的 HTTP 状态码如下：

200: 所有的消息都被成功发送到至少一个订阅。

202: 至少有一个消息没有匹配到任何订阅。

400: 至少有一个消息编码错误，如非法主题，或 QoS 超出范围等。

503: 至少有一个小因为服务重启的原因导致转发失败。

请求的 Body 或者 Body 中包含的某个消息无法通过 API 规范的类型检查时，HTTP 响应的消息与发布单个消息的 API
/publish 是一样的。
如果所有的消息都是合法的，那么 HTTP 返回的内容是一个 JSON 数组，每个元素代表了该消息转发的状态。

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

Array

payload_encoding	string Default: "plain" Enum: "plain" "base64" MQTT 消息体的编码方式，可以是 `base64` 或 `plain`。当设置为 `base64` 时，消息在发布前会先被解码，可以用于发布二进制等数据。
topic required	string 主题
qos	integer [ 0 .. 2 ] Default: 0 QoS
clientid	string Deprecated
payload required	string MQTT 消息体.
	object (emqx_mgmt_api_publish.message_properties)
retain	boolean Default: false 保留消息，布尔型字段，用于表示该消息是否是保留消息。

Responses

Request samples

Payload

Content type

application/json

[{"payload_encoding": "plain",
"topic": "api/example/topic",
"qos": 0,
"clientid": "string",
"payload": "hello emqx api",
"properties": {"payload_format_indicator": 0,
"message_expiry_interval": 0,
"response_topic": "some_other_topic",
"correlation_data": "string",
"user_properties": {"foo": "bar"
},
"content_type": "text/plain"
},
"retain": false
}
]

Response samples

200
202
400
503

Content type

application/json

[{"id": "string"
}
]

Publish a message

发布单条信息。

可能的 HTTP 状态码如下：

200: 消息被成功发送到至少一个订阅。

202: 没有匹配到任何订阅。

400: 消息编码错误，如非法主题，或 QoS 超出范围等。

503: 服务重启等过程中导致转发失败。

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

payload_encoding	string Default: "plain" Enum: "plain" "base64" MQTT 消息体的编码方式，可以是 `base64` 或 `plain`。当设置为 `base64` 时，消息在发布前会先被解码，可以用于发布二进制等数据。
topic required	string 主题
qos	integer [ 0 .. 2 ] Default: 0 QoS
clientid	string Deprecated
payload required	string MQTT 消息体.
	object (emqx_mgmt_api_publish.message_properties)
retain	boolean Default: false 保留消息，布尔型字段，用于表示该消息是否是保留消息。

Responses

Request samples

Payload

Content type

application/json

{"payload_encoding": "plain",
"topic": "api/example/topic",
"qos": 0,
"clientid": "string",
"payload": "hello emqx api",
"properties": {"payload_format_indicator": 0,
"message_expiry_interval": 0,
"response_topic": "some_other_topic",
"correlation_data": "string",
"user_properties": {"foo": "bar"
},
"content_type": "text/plain"
},
"retain": false
}

Response samples

200
202
400
503

Content type

application/json

{"id": "string"
}

Retainer

查看保留消息列表

Authorizations:

basicAuthbearerAuth

query Parameters

topic	string 主题过滤器，支持通配符，省略此项以匹配所有消息。
page	integer >= 1 Default: 1 Example: page=1 Page number of the results to fetch.
limit	integer [ 1 .. 10000 ] Default: 100 Example: limit=50 Results per page(max 10000)

Responses

Response samples

200
400

Content type

application/json

{"data": [{"msgid": "string",
"topic": "string",
"qos": 0,
"publish_at": "string",
"from_clientid": "string",
"from_username": "string"
}
],
"meta": {"page": 1,
"limit": 50,
"count": 0,
"hasnext": true
}
}

删除所有保留消息

Authorizations:

basicAuthbearerAuth

Responses

不支持主题通配符

Authorizations:

basicAuthbearerAuth

path Parameters

topic

required

string

主题

Responses

Response samples

200
400
404

Content type

application/json

{"payload": "string",
"msgid": "string",
"topic": "string",
"qos": 0,
"publish_at": "string",
"from_clientid": "string",
"from_username": "string"
}

删除对应的消息

Authorizations:

basicAuthbearerAuth

path Parameters

topic

required

string

主题

Responses

Response samples

400
404

Content type

application/json

{"code": "BAD_REQUEST",
"message": "string"
}

Metrics

EMQX 监控指标信息

Authorizations:

basicAuthbearerAuth

query Parameters

aggregate

boolean

Whether to aggregate all nodes Metrics

Responses

Response samples

200

Content type

application/json

Example

emqx_mgmt_api_metrics.node_metrics

[ ]

当前监控（统计）数据，例如整个集群中的连接数和连接速率。

Authorizations:

basicAuthbearerAuth

Responses

Response samples

200

Content type

application/json

{"dropped_msg_rate": 0,
"sent_msg_rate": 0,
"received_msg_rate": 0,
"subscriptions": 0,
"topics": 0,
"connections": 0,
"live_connections": 0,
"retained_msg_count": 0,
"shared_subscriptions": 0,
"license_quota": 0
}

Connectors

Test creating connector

测试创建一个新的连接器。

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

One of

type required	string Value: "timescale" 连接器的类型。
name required	string 连接器的名称
enable	boolean Default: true 启用 (是) 或停用 (否) 该连接器。
tags	Array of strings 连接器的标签
description	string Default: "" 描述性文本。
server required	string 用于连接的 IPv4 或 IPv6 地址或主机名。一个主机条目具有以下格式：Host[:Port]。如果未指定 [:Port]，将使用 PostgreSQL 默认端口5432。
database required	string 数据库名字。
pool_size	integer >= 1 Default: 8 桥接远端服务时使用的连接池大小。
username required	string 内部数据库的用户名。
password	string <password> 内部数据库密码。
auto_reconnect	boolean Deprecated Default: true 已弃用。自动重连数据库。
	object (emqx.ssl_client_opts)
	object (connector_postgres.resource_opts)

Responses

Request samples

Payload

Content type

application/json

Example

InfluxDB HTTP API V2 Connector

{"name": "influxdb_connector",
"type": "influxdb",
"ssl": {"enable": false
},
"description": "My example influxdb connector",
"server": "127.0.0.1:8086",
"enable": true,
"parameters": {"org": "examlpe_org",
"token": "example_token",
"bucket": "example_bucket",
"influxdb_type": "influxdb_api_v2"
}
}

Response samples

400

Content type

application/json

{"code": "TEST_FAILED",
"message": "string"
}

Get connector

通过 id 获取一个连接器。

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: http:my_http_connector

连接器 id。必须是 {type}:{name} 的格式。

Responses

Response samples

200
404

Content type

application/json

Example

InfluxDB HTTP API V2 Connector

{"name": "influxdb_connector",
"status": "connected",
"type": "influxdb",
"ssl": {"enable": false
},
"description": "My example influxdb connector",
"server": "127.0.0.1:8086",
"enable": true,
"parameters": {"org": "examlpe_org",
"token": "example_token",
"bucket": "example_bucket",
"influxdb_type": "influxdb_api_v2"
},
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
]
}

Update connector

通过 id 更新一个连接器。

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: http:my_http_connector

连接器 id。必须是 {type}:{name} 的格式。

Request Body schema: application/json

One of

enable	boolean Default: true 启用 (是) 或停用 (否) 该连接器。
tags	Array of strings 连接器的标签
description	string Default: "" 描述性文本。
server required	string 用于连接的 IPv4 或 IPv6 地址或主机名。一个主机条目具有以下格式：Host[:Port]。如果未指定 [:Port]，将使用 PostgreSQL 默认端口5432。
database required	string 数据库名字。
pool_size	integer >= 1 Default: 8 桥接远端服务时使用的连接池大小。
username required	string 内部数据库的用户名。
password	string <password> 内部数据库密码。
auto_reconnect	boolean Deprecated Default: true 已弃用。自动重连数据库。
	object (emqx.ssl_client_opts)
	object (connector_postgres.resource_opts)

Responses

Request samples

Payload

Content type

application/json

Example

InfluxDB HTTP API V2 Connector

{"ssl": {"enable": false
},
"description": "My example influxdb connector",
"server": "127.0.0.1:8086",
"enable": true,
"parameters": {"org": "examlpe_org",
"token": "example_token",
"bucket": "example_bucket",
"influxdb_type": "influxdb_api_v2"
}
}

Response samples

200
400
404

Content type

application/json

Example

InfluxDB HTTP API V2 Connector

{"name": "influxdb_connector",
"status": "connected",
"type": "influxdb",
"ssl": {"enable": false
},
"description": "My example influxdb connector",
"server": "127.0.0.1:8086",
"enable": true,
"parameters": {"org": "examlpe_org",
"token": "example_token",
"bucket": "example_bucket",
"influxdb_type": "influxdb_api_v2"
},
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
]
}

Delete connector

通过 id 删除一个连接器。

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: http:my_http_connector

连接器 id。必须是 {type}:{name} 的格式。

Responses

Response samples

400
404
503

Content type

application/json

{"code": "BAD_REQUEST",
"message": "string"
}

Manually start a connector

在集群的所有节点上启动连接器。

Authorizations:

basicAuthbearerAuth

path Parameters

id required	string Example: http:my_http_connector 连接器 id。必须是 {type}:{name} 的格式。
operation required	string Value: "start" Example: start 集群可用操作：'start'。

Responses

Response samples

400
404
501
503

Content type

application/json

{"code": "BAD_REQUEST",
"message": "string"
}

Enable or disable connector

在集群的所有节点上启用或禁用连接器。

Authorizations:

basicAuthbearerAuth

path Parameters

id required	string Example: http:my_http_connector 连接器 id。必须是 {type}:{name} 的格式。
enable required	boolean Example: true 是否启用此连接器。

Responses

Response samples

404
503

Content type

application/json

{"code": "NOT_FOUND",
"message": "string"
}

List connectors

列出所有创建的连接器。

Authorizations:

basicAuthbearerAuth

Responses

Response samples

200

Content type

application/json

[{"name": "influxdb_connector",
"status": "connected",
"type": "influxdb",
"ssl": {"enable": false
},
"description": "My example influxdb connector",
"server": "127.0.0.1:8086",
"enable": true,
"parameters": {"org": "examlpe_org",
"token": "example_token",
"bucket": "example_bucket",
"influxdb_type": "influxdb_api_v2"
},
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
]
},
{"name": "my_http_connector",
"status": "connected",
"type": "http",
"connect_timeout": "15s",
"pool_size": 1,
"enable": true,
"headers": {"content-type": "application/json"
},
"url": "http://localhost:8080/api/v1",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"pool_type": "hash",
"enable_pipelining": 100
},
{"name": "my_confluent_producer_connector",
"status": "connected",
"type": "confluent_producer",
"ssl": {"verify": "verify_none",
"enable": true,
"versions": ["tlsv1.3",
"tlsv1.2"
],
"server_name_indication": "auto"
},
"authentication": {"password": "******"
},
"connect_timeout": "5s",
"enable": true,
"socket_opts": {"nodelay": true,
"recbuf": "1024KB",
"sndbuf": "1024KB",
"tcp_keepalive": "none"
},
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"min_metadata_refresh_interval": "3s",
"bootstrap_hosts": "xyz.sa-east1.gcp.confluent.cloud:9092",
"metadata_request_timeout": "4s"
},
{"name": "iotdb_connector",
"status": "connected",
"type": "iotdb",
"ssl": {"enable": false
},
"authentication": {"password": "******",
"username": "root"
},
"description": "My example iotdb connector",
"connect_timeout": "15s",
"pool_size": 8,
"enable": true,
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"pool_type": "random",
"enable_pipelining": 100,
"base_url": "http://iotdb.local:18080/",
"iotdb_version": "v1.1.x"
},
{"name": "my_azure_event_hub_producer_connector",
"status": "connected",
"type": "azure_event_hub_producer",
"ssl": {"verify": "verify_none",
"enable": true,
"versions": ["tlsv1.3",
"tlsv1.2"
],
"server_name_indication": "auto"
},
"authentication": {"password": "******"
},
"connect_timeout": "5s",
"enable": true,
"socket_opts": {"nodelay": true,
"recbuf": "1024KB",
"sndbuf": "1024KB",
"tcp_keepalive": "none"
},
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"min_metadata_refresh_interval": "3s",
"bootstrap_hosts": "namespace.servicebus.windows.net:9093",
"metadata_request_timeout": "4s"
},
{"name": "my_kafka_producer_connector",
"status": "connected",
"type": "kafka_producer",
"authentication": {"mechanism": "plain",
"password": "******",
"username": "username"
},
"connect_timeout": "5s",
"enable": true,
"socket_opts": {"nodelay": true,
"recbuf": "1024KB",
"sndbuf": "1024KB",
"tcp_keepalive": "none"
},
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"min_metadata_refresh_interval": "3s",
"bootstrap_hosts": "localhost:9092",
"metadata_request_timeout": "4s"
},
{"name": "kinesis_connector",
"status": "connected",
"type": "kinesis",
"description": "My example kinesis connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"aws_access_key_id": "your_access_key",
"aws_secret_access_key": "aws_secret_key",
"endpoint": "http://localhost:4566",
"max_retries": 2,
"pool_size": 8
},
{"name": "syskeeper_proxy",
"status": "connected",
"type": "syskeeper_proxy",
"listen": "127.0.0.1:9092",
"enable": true,
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"handshake_timeout": "16s",
"acceptors": 16,
"actions": ["my_action"
]
},
{"name": "my_pgsql_connector",
"status": "connected",
"type": "pgsql",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"database": "emqx_data",
"enable": true,
"password": "public",
"pool_size": 8,
"server": "127.0.0.1:5432",
"ssl": {"ciphers": [ ],
"depth": 10,
"enable": false,
"hibernate_after": "5s",
"log_level": "notice",
"reuse_sessions": true,
"secure_renegotiate": true,
"verify": "verify_peer",
"versions": ["tlsv1.3",
"tlsv1.2"
]
},
"username": "postgres"
},
{"name": "mysql_connector",
"status": "connected",
"type": "mysql",
"description": "My example mysql connector",
"pool_size": 8,
"server": "127.0.0.1:3306",
"password": "******",
"username": "root",
"database": "test",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"resource_opts": {"health_check_interval": "20s"
}
},
{"timeout": "5s",
"name": "rabbitmq_connector",
"port": 5672,
"status": "connected",
"type": "rabbitmq",
"ssl": {"enable": false
},
"description": "My example rabbitmq connector",
"pool_size": 8,
"server": "127.0.0.1",
"enable": true,
"password": "******",
"username": "guest",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"heartbeat": "30s",
"actions": ["my_action"
],
"virtual_host": "/"
},
{"name": "dynamo_connector",
"status": "connected",
"type": "dynamo",
"description": "My example dynamo connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"aws_access_key_id": "root",
"aws_secret_access_key": "******",
"enable": true,
"pool_size": 8,
"resource_opts": {"health_check_interval": "15s",
"start_timeout": "5s"
},
"url": "http://127.0.0.1:8000"
},
{"name": "tdengine_connector",
"status": "connected",
"type": "tdengine",
"description": "My example tdengine connector",
"pool_size": 8,
"server": "127.0.0.1:6041",
"enable": true,
"password": "******",
"username": "root",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
]
},
{"name": "mongodb_single_connector",
"status": "connected",
"type": "mongodb_single",
"description": "My example mongodb_single connector",
"parameters": {"pool_size": 8,
"enable": true,
"password": "******",
"username": "myuser",
"database": "mqtt",
"srv_record": false
},
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
]
},
{"name": "demo",
"status": "connected",
"type": "greptimedb",
"ssl": {"enable": false
},
"description": "My example greptimedb connector",
"server": "127.0.0.1:4001",
"enable": true,
"password": "******",
"username": "example_username",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"resource_opts": {"batch_size": 100,
"batch_time": "20ms"
},
"local_topic": "local/topic/#",
"dbname": "example_db"
},
{"name": "opents_connector",
"status": "connected",
"type": "opents",
"description": "My example opents connector",
"pool_size": 8,
"server": "http://localhost:4242/",
"enable": true,
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
]
},
{"name": "mongodb_sharded_connector",
"status": "connected",
"type": "mongodb_sharded",
"description": "My example mongodb_sharded connector",
"parameters": {"pool_size": 8,
"enable": true,
"password": "******",
"username": "myuser",
"database": "mqtt",
"srv_record": false
},
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
]
},
{"name": "mongodb_rs_connector",
"status": "connected",
"type": "mongodb_rs",
"description": "My example mongodb_rs connector",
"parameters": {"pool_size": 8,
"enable": true,
"password": "******",
"username": "myuser",
"database": "mqtt",
"srv_record": false
},
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
]
},
{"status": "connected",
"ssl": {"enable": false
},
"description": "My redis single connector",
"enable": true,
"parameters": {"pool_size": 8,
"server": "127.0.0.1:6379",
"password": "******",
"username": "test",
"database": 1,
"redis_type": "single"
},
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
]
},
{"name": "oracle_connector",
"status": "connected",
"type": "oracle",
"description": "My example oracle connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"password": "oracle",
"pool_size": 8,
"resource_opts": {"health_check_interval": "15s",
"start_timeout": "5s"
},
"server": "127.0.0.1:1521",
"service_name": "XE",
"sid": "XE",
"username": "system"
},
{"name": "hstreamdb_connector",
"status": "connected",
"type": "hstreamdb",
"description": "My example hstreamdb connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"grpc_timeout": "30s",
"resource_opts": {"health_check_interval": "15s",
"start_timeout": "5s"
},
"ssl": {"enable": false,
"verify": "verify_peer"
},
"url": "http://127.0.0.1:6570"
},
{"name": "influxdb_connector",
"status": "connected",
"type": "influxdb",
"ssl": {"enable": false
},
"description": "My example influxdb connector",
"server": "127.0.0.1:8086",
"enable": true,
"parameters": {"password": "******",
"username": "example_username",
"database": "example_database",
"influxdb_type": "influxdb_api_v1"
},
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
]
},
{"port": 443,
"status": "connected",
"host": "s3.eu-east-1.amazonaws.com",
"description": "My S3 connector",
"enable": true,
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"transport_options": {"ssl": {"verify": "verify_peer",
"enable": true
},
"connect_timeout": "1s",
"pool_size": 4,
"max_retries": 1,
"request_timeout": "60s",
"enable_pipelining": 1
},
"secret_access_key": "SECRET",
"access_key_id": "ACCESS"
},
{"name": "cassandra_connector",
"status": "connected",
"type": "cassandra",
"description": "My example cassandra connector",
"pool_size": 8,
"password": "******",
"username": "root",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"servers": "127.0.0.1:9042",
"keyspace": "mqtt"
},
{"name": "my_connector",
"status": "connected",
"type": "gcp_pubsub_producer",
"description": "my connector",
"connect_timeout": "15s",
"pool_size": 8,
"enable": true,
"max_retries": 2,
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"resource_opts": {"start_timeout": "5s",
"start_after_created": true,
"health_check_interval": "30s"
},
"pipelining": 100,
"service_account_json": {"type": "service_account",
"private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvQI...",
"client_id": "123812831923812319190",
"auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
"auth_uri": "https://accounts.google.com/o/oauth2/auth",
"client_email": "test@myproject.iam.gserviceaccount.com",
"client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/test%40myproject.iam.gserviceaccount.com",
"private_key_id": "kid",
"project_id": "myproject",
"token_uri": "https://oauth2.googleapis.com/token"
}
},
{"name": "clickhouse_connector",
"status": "connected",
"type": "clickhouse",
"description": "My example clickhouse connector",
"pool_size": 8,
"password": "******",
"url": "http://localhost:8123",
"username": "default",
"database": "mqtt",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
]
},
{"driver": "ms-sql",
"name": "sqlserver_connector",
"status": "connected",
"type": "sqlserver",
"description": "My example sqlserver connector",
"pool_size": 8,
"server": "127.0.0.1:1433",
"password": "******",
"username": "sa",
"database": "test",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"resource_opts": {"health_check_interval": "20s"
}
},
{"status": "connected",
"ssl": {"enable": false
},
"description": "My redis cluster connector",
"enable": true,
"parameters": {"pool_size": 8,
"password": "******",
"username": "test",
"servers": "127.0.0.1:6379,127.0.0.2:6379",
"redis_type": "cluster"
},
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
]
},
{"name": "my_matrix_connector",
"status": "connected",
"type": "matrix",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"database": "emqx_data",
"enable": true,
"password": "public",
"pool_size": 8,
"server": "127.0.0.1:5432",
"ssl": {"ciphers": [ ],
"depth": 10,
"enable": false,
"hibernate_after": "5s",
"log_level": "notice",
"reuse_sessions": true,
"secure_renegotiate": true,
"verify": "verify_peer",
"versions": ["tlsv1.3",
"tlsv1.2"
]
},
"username": "postgres"
},
{"name": "my_connector",
"status": "connected",
"type": "kafka_consumer",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"start_timeout": "5s",
"start_after_created": true,
"health_check_interval": "30s"
},
"bootstrap_hosts": "kafka.emqx.net:9092"
},
{"name": "rocketmq_connector",
"status": "connected",
"type": "rocketmq",
"description": "My example rocketmq connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"enable": true,
"pool_size": 8,
"resource_opts": {"health_check_interval": "15s",
"start_after_created": true,
"start_timeout": "5s"
},
"servers": "127.0.0.1:9876"
},
{"name": "pulsar_connector",
"status": "connected",
"type": "pulsar",
"ssl": {"enable": false
},
"authentication": "none",
"description": "My example pulsar connector",
"connect_timeout": "5s",
"enable": true,
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"servers": "pulsar://127.0.0.1:6650"
},
{"name": "syskeeper_forwarder",
"status": "connected",
"type": "syskeeper_forwarder",
"pool_size": 16,
"server": "127.0.0.1:9092",
"enable": true,
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"ack_timeout": "10s",
"actions": ["my_action"
],
"ack_mode": "no_ack"
},
{"name": "elasticsearch_connector",
"status": "connected",
"type": "elasticsearch",
"ssl": {"enable": false
},
"authentication": {"password": "******",
"username": "root"
},
"description": "My example elasticsearch connector",
"connect_timeout": "15s",
"pool_size": 8,
"server": "127.0.0.1:9200",
"enable": true,
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"pool_type": "random",
"enable_pipelining": 100
},
{"name": "my_connector",
"status": "connected",
"type": "gcp_pubsub_producer",
"connect_timeout": "10s",
"pool_size": 8,
"enable": true,
"max_retries": 2,
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"resource_opts": {"request_ttl": "60s"
},
"pipelining": 100,
"service_account_json": {"type": "service_account",
"private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvQI...",
"client_id": "123812831923812319190",
"auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
"auth_uri": "https://accounts.google.com/o/oauth2/auth",
"client_email": "test@myproject.iam.gserviceaccount.com",
"client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/test%40myproject.iam.gserviceaccount.com",
"private_key_id": "kid",
"project_id": "myproject",
"token_uri": "https://oauth2.googleapis.com/token"
}
},
{"name": "my_timescale_connector",
"status": "connected",
"type": "timescale",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
],
"database": "emqx_data",
"enable": true,
"password": "public",
"pool_size": 8,
"server": "127.0.0.1:5432",
"ssl": {"ciphers": [ ],
"depth": 10,
"enable": false,
"hibernate_after": "5s",
"log_level": "notice",
"reuse_sessions": true,
"secure_renegotiate": true,
"verify": "verify_peer",
"versions": ["tlsv1.3",
"tlsv1.2"
]
},
"username": "postgres"
},
{"status": "connected",
"ssl": {"enable": false
},
"description": "My redis sentinel connector",
"enable": true,
"parameters": {"pool_size": 8,
"password": "******",
"username": "test",
"database": 1,
"sentinel": "myredismaster",
"servers": "127.0.0.1:6379,127.0.0.2:6379",
"redis_type": "sentinel"
},
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
]
}
]

Create connector

通过类型和名称创建一个新的连接器。

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

One of

type required	string Value: "timescale" 连接器的类型。
name required	string 连接器的名称
enable	boolean Default: true 启用 (是) 或停用 (否) 该连接器。
tags	Array of strings 连接器的标签
description	string Default: "" 描述性文本。
server required	string 用于连接的 IPv4 或 IPv6 地址或主机名。一个主机条目具有以下格式：Host[:Port]。如果未指定 [:Port]，将使用 PostgreSQL 默认端口5432。
database required	string 数据库名字。
pool_size	integer >= 1 Default: 8 桥接远端服务时使用的连接池大小。
username required	string 内部数据库的用户名。
password	string <password> 内部数据库密码。
auto_reconnect	boolean Deprecated Default: true 已弃用。自动重连数据库。
	object (emqx.ssl_client_opts)
	object (connector_postgres.resource_opts)

Responses

Request samples

Payload

Content type

application/json

Example

InfluxDB HTTP API V2 Connector

{"name": "influxdb_connector",
"type": "influxdb",
"ssl": {"enable": false
},
"description": "My example influxdb connector",
"server": "127.0.0.1:8086",
"enable": true,
"parameters": {"org": "examlpe_org",
"token": "example_token",
"bucket": "example_bucket",
"influxdb_type": "influxdb_api_v2"
}
}

Response samples

201
400

Content type

application/json

Example

InfluxDB HTTP API V2 Connector

{"name": "influxdb_connector",
"status": "connected",
"type": "influxdb",
"ssl": {"enable": false
},
"description": "My example influxdb connector",
"server": "127.0.0.1:8086",
"enable": true,
"parameters": {"org": "examlpe_org",
"token": "example_token",
"bucket": "example_bucket",
"influxdb_type": "influxdb_api_v2"
},
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"actions": ["my_action"
]
}

Rules

Test a rule

测试一个规则

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

rule_engine.ctx_delivery_dropped (object) or rule_engine.ctx_bridge_mqtt (object) or rule_engine.ctx_check_authz_complete (object) or rule_engine.ctx_connack (object) or rule_engine.ctx_disconnected (object) or rule_engine.ctx_connected (object) or rule_engine.ctx_dropped (object) or rule_engine.ctx_acked (object) or rule_engine.ctx_delivered (object) or rule_engine.ctx_unsub (object) or rule_engine.ctx_sub (object) or rule_engine.ctx_pub (object)

Default: {}

测试事件的上下文

sql

required

string

测试的 SQL

Responses

Request samples

Payload

Content type

application/json

{"context": { },
"sql": "string"
}

Response samples

400
412

Content type

application/json

{"code": "BAD_REQUEST",
"message": "string"
}

List rules

列出所有规则

Authorizations:

basicAuthbearerAuth

query Parameters

enable	boolean 根据规则是否开启条件过滤
from	string 根据规则来源 Topic 过滤, 需要完全匹配
like_id	string 根据规则 id 过滤, 使用子串模糊匹配
like_from	string 根据规则来源 Topic 过滤, 使用子串模糊匹配
like_description	string 根据规则描述过滤, 使用子串模糊匹配
match_from	string 根据规则来源 Topic 过滤, 使用 MQTT Topic 匹配
page	integer >= 1 Default: 1 Example: page=1 Page number of the results to fetch.
limit	integer [ 1 .. 10000 ] Default: 100 Example: limit=50 Results per page(max 10000)

Responses

Response samples

200
400

Content type

application/json

{"data": [{"id": "293fb66f",
"from": "t/#",
"created_at": "2021-12-01T15:00:43.153+08:00",
"name": "foo",
"sql": "SELECT * FROM \"test/topic\" WHERE payload.x = 1",
"actions": ["webhook:my_webhook",
{"args": {"payload": "${payload}",
"topic": "t/1"
},
"function": "republish"
},
{"function": "console"
}
],
"enable": true,
"description": "Some description",
"metadata": { }
}
],
"meta": {"page": 1,
"limit": 50,
"count": 0,
"hasnext": true
}
}

Create a rule

通过指定 ID 创建规则

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

name	string Default: "" 规则名字
sql required	string 用于处理消息的 SQL 。示例：`SELECT * FROM "test/topic" WHERE payload.x = 1`
	Array of rule_engine.user_provided_function (object) or rule_engine.builtin_action_console (object) or rule_engine.builtin_action_republish (object) or strings Default: [] 规则的动作列表。动作可以是指向 EMQX 数据桥接的引用，也可以是一个指向函数的对象。我们支持一些内置函数，如“republish”和“console”，我们还支持用户提供的函数，它的格式为：“{module}:{function}”。列表中的动作按顺序执行。这意味着如果其中一个动作执行缓慢，则以下所有动作都不会被执行直到它返回。如果其中一个动作崩溃，在它之后的所有动作仍然会被按照原始顺序执行。如果运行动作时出现任何错误，则会出现错误消息，并且相应的计数器会增加。
enable	boolean Default: true 启用或禁用规则引擎
description	string Default: "" 规则的描述
metadata	object 规则的元数据，不要手动修改

Responses

Request samples

Payload

Content type

application/json

{"name": "foo",
"sql": "SELECT * FROM \"test/topic\" WHERE payload.x = 1",
"actions": ["webhook:my_webhook",
{"args": {"payload": "${payload}",
"topic": "t/1"
},
"function": "republish"
},
{"function": "console"
}
],
"enable": true,
"description": "Some description",
"metadata": { }
}

Response samples

201
400

Content type

application/json

{"id": "293fb66f",
"from": "t/#",
"created_at": "2021-12-01T15:00:43.153+08:00",
"name": "foo",
"sql": "SELECT * FROM \"test/topic\" WHERE payload.x = 1",
"actions": ["webhook:my_webhook",
{"args": {"payload": "${payload}",
"topic": "t/1"
},
"function": "republish"
},
{"function": "console"
}
],
"enable": true,
"description": "Some description",
"metadata": { }
}

Get rule metrics

通过给定的 Id 获得规则的指标数据

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: my_rule_id

Responses

Response samples

200
404

Content type

application/json

{"id": "293fb66f",
"metrics": {"matched": 0,
"matched.rate": 0,
"matched.rate.max": 0,
"matched.rate.last5m": 0,
"passed": 0,
"failed": 0,
"failed.exception": 0,
"failed.unknown": 0,
"actions.total": 0,
"actions.success": 0,
"actions.failed": 0,
"actions.failed.out_of_service": 0,
"actions.failed.unknown": 0
},
"node_metrics": [{"node": "emqx@127.0.0.1",
"matched": 0,
"matched.rate": 0,
"matched.rate.max": 0,
"matched.rate.last5m": 0,
"passed": 0,
"failed": 0,
"failed.exception": 0,
"failed.unknown": 0,
"actions.total": 0,
"actions.success": 0,
"actions.failed": 0,
"actions.failed.out_of_service": 0,
"actions.failed.unknown": 0
}
]
}

List rule events

列出所有能被规则使用的事件

Authorizations:

basicAuthbearerAuth

Responses

Response samples

200

Content type

application/json

{"event": "$events/client_connected",
"title": "some title",
"description": "some desc",
"columns": { },
"test_columns": { },
"sql_example": "string"
}

Reset rule metrics

重置规则计数

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: my_rule_id

Responses

Response samples

404

Content type

application/json

{"code": "NOT_FOUND",
"message": "string"
}

Get rule

通过 ID 查询规则

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: my_rule_id

Responses

Response samples

200
404

Content type

application/json

{"id": "293fb66f",
"from": "t/#",
"created_at": "2021-12-01T15:00:43.153+08:00",
"name": "foo",
"sql": "SELECT * FROM \"test/topic\" WHERE payload.x = 1",
"actions": ["webhook:my_webhook",
{"args": {"payload": "${payload}",
"topic": "t/1"
},
"function": "republish"
},
{"function": "console"
}
],
"enable": true,
"description": "Some description",
"metadata": { }
}

Update rule

通过 ID 更新集群里所有节点上的规则

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: my_rule_id

Request Body schema: application/json

name	string Default: "" 规则名字
sql required	string 用于处理消息的 SQL 。示例：`SELECT * FROM "test/topic" WHERE payload.x = 1`
	Array of rule_engine.user_provided_function (object) or rule_engine.builtin_action_console (object) or rule_engine.builtin_action_republish (object) or strings Default: [] 规则的动作列表。动作可以是指向 EMQX 数据桥接的引用，也可以是一个指向函数的对象。我们支持一些内置函数，如“republish”和“console”，我们还支持用户提供的函数，它的格式为：“{module}:{function}”。列表中的动作按顺序执行。这意味着如果其中一个动作执行缓慢，则以下所有动作都不会被执行直到它返回。如果其中一个动作崩溃，在它之后的所有动作仍然会被按照原始顺序执行。如果运行动作时出现任何错误，则会出现错误消息，并且相应的计数器会增加。
enable	boolean Default: true 启用或禁用规则引擎
description	string Default: "" 规则的描述
metadata	object 规则的元数据，不要手动修改

Responses

Request samples

Payload

Content type

application/json

{"name": "foo",
"sql": "SELECT * FROM \"test/topic\" WHERE payload.x = 1",
"actions": ["webhook:my_webhook",
{"args": {"payload": "${payload}",
"topic": "t/1"
},
"function": "republish"
},
{"function": "console"
}
],
"enable": true,
"description": "Some description",
"metadata": { }
}

Response samples

200
400

Content type

application/json

{"id": "293fb66f",
"from": "t/#",
"created_at": "2021-12-01T15:00:43.153+08:00",
"name": "foo",
"sql": "SELECT * FROM \"test/topic\" WHERE payload.x = 1",
"actions": ["webhook:my_webhook",
{"args": {"payload": "${payload}",
"topic": "t/1"
},
"function": "republish"
},
{"function": "console"
}
],
"enable": true,
"description": "Some description",
"metadata": { }
}

Delete rule

通过 ID 删除集群里所有节点上的规则

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: my_rule_id

Responses

Response samples

404

Content type

application/json

{"code": "NOT_FOUND",
"message": "string"
}

Actions

Reset action metrics

通过 id 重置数据桥接指标。

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: http:my_http_action

数据桥接 ID，格式必须为 {type}:{name}。

Responses

Response samples

404

Content type

application/json

{"code": "NOT_FOUND",
"message": "string"
}

Manually start a bridge

启用集群中所有节点上的数据桥接。

Authorizations:

basicAuthbearerAuth

path Parameters

id required	string Example: http:my_http_action 数据桥接 ID，格式必须为 {type}:{name}。
operation required	string Value: "start" Example: start 集群可用操作：'启动'。

Responses

Response samples

400
404
501
503

Content type

application/json

{"code": "BAD_REQUEST",
"message": "string"
}

Enable or disable bridge

启用或禁用集群内所有节点上的数据桥接。

Authorizations:

basicAuthbearerAuth

path Parameters

id required	string Example: http:my_http_action 数据桥接 ID，格式必须为 {type}:{name}。
enable required	boolean Example: true 是否启用该数据桥接。

Responses

Response samples

404
503

Content type

application/json

{"code": "NOT_FOUND",
"message": "string"
}

List bridges

列出所有创建的数据桥接。

Authorizations:

basicAuthbearerAuth

Responses

Response samples

200

Content type

application/json

[{"name": "my_azure_event_hub_producer_action",
"status": "connected",
"type": "azure_event_hub_producer",
"enable": true,
"parameters": {"message": {"value": "${.}",
"key": "${.clientid}"
},
"buffer": {"mode": "hybrid",
"segment_bytes": "100MB",
"per_partition_limit": "2GB",
"memory_overload_protection": true
},
"topic": "topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"kafka_headers": "${.pub_props}",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_header_value_encode_mode": "none",
"partition_strategy": "random",
"partition_count_refresh_interval": "60s"
},
"connector": "my_azure_event_hub_producer_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"local_topic": "mqtt/local/topic"
},
{"name": "cassandra_action",
"status": "connected",
"type": "cassandra",
"description": "My example cassandra action",
"enable": true,
"parameters": {"cql": "insert into mqtt_msg(msgid, topic, qos, payload, arrived) values (${id}, ${topic},  ${qos}, ${payload}, ${timestamp})"
},
"connector": "cassandra_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
}
},
{"name": "clickhouse_action",
"status": "connected",
"type": "clickhouse",
"description": "My example clickhouse action",
"enable": true,
"parameters": {"sql": "INSERT INTO messages(data, arrived) VALUES ('${payload}', ${timestamp})",
"batch_value_separator": ", "
},
"connector": "clickhouse_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
}
},
{"name": "my_confluent_producer_action",
"status": "connected",
"type": "confluent_producer",
"enable": true,
"parameters": {"message": {"value": "${.}",
"key": "${.clientid}"
},
"buffer": {"mode": "hybrid",
"segment_bytes": "100MB",
"per_partition_limit": "2GB",
"memory_overload_protection": true
},
"topic": "topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"kafka_headers": "${.pub_props}",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_header_value_encode_mode": "none",
"partition_strategy": "random",
"partition_count_refresh_interval": "60s"
},
"connector": "my_confluent_producer_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"local_topic": "mqtt/local/topic"
},
{"name": "dynamo_action",
"status": "connected",
"type": "dynamo",
"description": "My example dynamo action",
"enable": true,
"connector": "dynamo_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
},
"parameters": {"table": "mqtt_msg",
"template": ""
}
},
{"name": "elasticsearch_action",
"status": "connected",
"type": "elasticsearch",
"description": "My example elasticsearch action",
"enable": true,
"parameters": {"index": "${payload.index}",
"action": "create",
"doc": "${payload.doc}",
"overwrite": true
},
"connector": "elasticsearch_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
}
},
{"status": "connected",
"description": "My action",
"enable": true,
"parameters": {"payload_template": "${payload}",
"attributes_template": [{"value": "${payload.attrs.v}",
"key": "${payload.attrs.k}"
}
],
"ordering_key_template": "${payload.ok}",
"pubsub_topic": "mytopic"
},
"connector": "my_connector_name",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"batch_size": 5
},
"local_topic": "local/topic"
},
{"name": "greptimedb_action",
"status": "connected",
"type": "greptimedb",
"description": "My example greptimedb action",
"enable": true,
"parameters": {"precision": "ms",
"write_syntax": "${topic},clientid=${clientid} payload=${payload},${clientid}_int_value=${payload.int_key}i,uint_value=${payload.uint_key}u,bool=${payload.bool}"
},
"connector": "greptimedb_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
}
},
{"name": "hstreamdb_action",
"status": "connected",
"type": "hstreamdb",
"description": "My example hstreamdb action",
"enable": true,
"connector": "hstreamdb_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
},
"parameters": {"aggregation_pool_size": 8,
"partition_key": "hej",
"record_template": "${payload}",
"stream": "mqtt_message",
"writer_pool_size": 8
}
},
{"name": "my_http_action",
"status": "connected",
"type": "http",
"enable": true,
"parameters": {"path": "/room/${room_no}",
"body": "${.}",
"headers": { },
"method": "post"
},
"connector": "my_http_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"worker_pool_size": 16,
"query_mode": "async",
"health_check_interval": "15s"
}
},
{"name": "influxdb_action",
"status": "connected",
"type": "influxdb",
"description": "My example influxdb action",
"enable": true,
"parameters": {"precision": "ms",
"write_syntax": "${topic},clientid=${clientid} payload=${payload},${clientid}_int_value=${payload.int_key}i,bool=${payload.bool}"
},
"connector": "influxdb_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
}
},
{"name": "iotdb_action",
"status": "connected",
"type": "iotdb",
"description": "My example iotdb action",
"enable": true,
"parameters": {"data": [{"timestamp": "now",
"value": "${st}",
"data_type": "BOOLEAN",
"measurement": "status"
}
],
"device_id": "my_device",
"is_aligned": false
},
"connector": "iotdb_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
}
},
{"name": "my_kafka_producer_action",
"status": "connected",
"type": "kafka_producer",
"enable": true,
"parameters": {"message": {"timestamp": "${.timestamp}",
"value": "${.}",
"key": "${.clientid}"
},
"buffer": {"mode": "hybrid",
"segment_bytes": "100MB",
"per_partition_limit": "2GB",
"memory_overload_protection": true
},
"compression": "no_compression",
"topic": "kafka-topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"kafka_headers": "${pub_props}",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_header_value_encode_mode": "none",
"partitions_limit": "all_partitions",
"partition_strategy": "random",
"partition_count_refresh_interval": "60s"
},
"connector": "my_kafka_producer_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "32s"
},
"local_topic": "mqtt/local/topic"
},
{"name": "kinesis_action",
"status": "connected",
"type": "kinesis",
"description": "My example kinesis action",
"enable": true,
"parameters": {"partition_key": "any_key",
"payload_template": "${.}",
"stream_name": "my_stream"
},
"connector": "kinesis_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
}
},
{"name": "my_action",
"status": "connected",
"type": "matrix",
"enable": true,
"connector": "my_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"batch_size": 1,
"worker_pool_size": 16,
"max_buffer_bytes": "256MB",
"batch_time": "50ms",
"inflight_window": 100,
"request_ttl": "45s"
},
"parameters": {"sql": "INSERT INTO client_events(clientid, event, created_at)VALUES (\n  ${clientid},\n  ${event},\n  TO_TIMESTAMP((${timestamp} :: bigint))\n)"
}
},
{"name": "mongodb_action",
"status": "connected",
"type": "mongodb",
"description": "My example mongodb action",
"enable": true,
"parameters": {"collection": "mycol"
},
"connector": "mongodb_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
}
},
{"name": "mqtt_action",
"status": "connected",
"type": "mqtt",
"description": "My example mqtt action",
"enable": true,
"parameters": {"retain": false,
"payload": "${.payload}",
"topic": "remote/topic",
"qos": 2
},
"connector": "mqtt_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
}
},
{"name": "mysql_action",
"status": "connected",
"type": "mysql",
"description": "My example mysql action",
"enable": true,
"parameters": {"sql": "insert into t_mqtt_msg(msgid, topic, qos, payload, arrived) values (${id}, ${topic}, ${qos}, ${payload}, FROM_UNIXTIME(${timestamp}/1000))"
},
"connector": "mysql_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
}
},
{"name": "opents_action",
"status": "connected",
"type": "opents",
"description": "My example opents action",
"enable": true,
"parameters": {"data": [{"value": "${value}",
"tags": "${tags}",
"metric": "${metric}"
}
]
},
"connector": "opents_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
}
},
{"name": "oracle_action",
"status": "connected",
"type": "oracle",
"description": "My example oracle action",
"enable": true,
"parameters": {"sql": "insert into t_mqtt_msgs(msgid, topic, qos, payload) values (${id}, ${topic}, ${qos}, ${payload})"
},
"connector": "oracle_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
}
},
{"name": "my_action",
"status": "connected",
"type": "pgsql",
"enable": true,
"connector": "my_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"batch_size": 1,
"worker_pool_size": 16,
"max_buffer_bytes": "256MB",
"batch_time": "50ms",
"inflight_window": 100,
"request_ttl": "45s"
},
"parameters": {"sql": "INSERT INTO client_events(clientid, event, created_at)VALUES (\n  ${clientid},\n  ${event},\n  TO_TIMESTAMP((${timestamp} :: bigint))\n)"
}
},
{"name": "pulsar_action",
"status": "connected",
"type": "pulsar",
"description": "My example pulsar action",
"enable": true,
"parameters": {"message": {"value": "${.}",
"key": "${.clientid}"
},
"pulsar_topic": "test_topic",
"sync_timeout": "5s"
},
"connector": "pulsar_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
}
},
{"name": "rabbitmq_action",
"status": "connected",
"type": "rabbitmq",
"description": "My example rabbitmq action",
"enable": true,
"parameters": {"exchange": "test_exchange",
"payload_template": "${.payload}",
"routing_key": "/",
"wait_for_publish_confirmations": true,
"delivery_mode": "non_persistent",
"publish_confirmation_timeout": "30s"
},
"connector": "rabbitmq_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
}
},
{"status": "connected",
"description": "My action",
"enable": true,
"parameters": {"command_template": ["LPUSH",
"MSGS",
"${payload}"
]
},
"connector": "my_connector_name",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"batch_size": 1
}
},
{"name": "rocketmq_action",
"status": "connected",
"type": "rocketmq",
"description": "My example rocketmq action",
"enable": true,
"connector": "rocketmq_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
},
"parameters": {"refresh_interval": "3s",
"send_buffer": "1024KB",
"sync_timeout": "3s",
"template": "",
"topic": "TopicTest"
}
},
{"status": "connected",
"description": "My action",
"enable": true,
"parameters": {"key": "${topic}",
"content": "${payload}",
"bucket": "${clientid}",
"acl": "public_read"
},
"connector": "my_s3_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"inflight_window": 10,
"query_mode": "sync"
}
},
{"name": "sqlserver_action",
"status": "connected",
"type": "sqlserver",
"description": "My example sqlserver action",
"enable": true,
"connector": "sqlserver_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
},
"parameters": {"sql": "insert into t_mqtt_msg(msgid, topic, qos, payload) values ( ${id}, ${topic}, ${qos}, ${payload} )"
}
},
{"name": "syskeeper_forwarder",
"status": "connected",
"type": "syskeeper_forwarder",
"enable": true,
"parameters": {"template": "${payload}",
"target_topic": "${topic}"
},
"connector": "syskeeper_forwarder",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"worker_pool_size": 16
}
},
{"name": "tdengine_action",
"status": "connected",
"type": "tdengine",
"description": "My example tdengine action",
"enable": true,
"parameters": {"database": "mqtt",
"sql": "insert into t_mqtt_msg(ts, msgid, mqtt_topic, qos, payload, arrived) values (${ts}, '${id}', '${topic}', ${qos}, '${payload}', ${timestamp})"
},
"connector": "tdengine_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"health_check_interval": "30s"
}
},
{"name": "my_action",
"status": "connected",
"type": "timescale",
"enable": true,
"connector": "my_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"resource_opts": {"batch_size": 1,
"worker_pool_size": 16,
"max_buffer_bytes": "256MB",
"batch_time": "50ms",
"inflight_window": 100,
"request_ttl": "45s"
},
"parameters": {"sql": "INSERT INTO client_events(clientid, event, created_at)VALUES (\n  ${clientid},\n  ${event},\n  TO_TIMESTAMP((${timestamp} :: bigint))\n)"
}
}
]

Create bridge

通过类型和名称创建一个新的数据桥接。

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

One of

type required	string Value: "timescale" Bridge 类型
name required	string 桥接名字
local_topic	string MQTT 主题或主题过滤器作为数据源（动作输入）。如果规则动作用作数据源，则应将此配置留空，否则消息将在远程系统中重复。
enable	boolean Default: true 启用（是）或停用（否）此动作。
connector required	string 由动作指定的连接器名称，用于选择外部资源。
tags	Array of strings 连接器的标签
description	string Default: "" 描述性文本。
required	object (bridge_pgsql.action_parameters)
	object (actions_and_sources.action_resource_opts)

Responses

Request samples

Payload

Content type

application/json

Example

Azure Event Hub Action

{"name": "my_azure_event_hub_producer_action",
"type": "azure_event_hub_producer",
"enable": true,
"parameters": {"message": {"value": "${.}",
"key": "${.clientid}"
},
"buffer": {"mode": "hybrid",
"segment_bytes": "100MB",
"per_partition_limit": "2GB",
"memory_overload_protection": true
},
"topic": "topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"kafka_headers": "${.pub_props}",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_header_value_encode_mode": "none",
"partition_strategy": "random",
"partition_count_refresh_interval": "60s"
},
"connector": "my_azure_event_hub_producer_connector",
"local_topic": "mqtt/local/topic"
}

Response samples

201
400

Content type

application/json

Example

Azure Event Hub Action

{"name": "my_azure_event_hub_producer_action",
"status": "connected",
"type": "azure_event_hub_producer",
"enable": true,
"parameters": {"message": {"value": "${.}",
"key": "${.clientid}"
},
"buffer": {"mode": "hybrid",
"segment_bytes": "100MB",
"per_partition_limit": "2GB",
"memory_overload_protection": true
},
"topic": "topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"kafka_headers": "${.pub_props}",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_header_value_encode_mode": "none",
"partition_strategy": "random",
"partition_count_refresh_interval": "60s"
},
"connector": "my_azure_event_hub_producer_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"local_topic": "mqtt/local/topic"
}

Test creating bridge

测试创建一个新的数据桥接。

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

One of

type required	string Value: "timescale" Bridge 类型
name required	string 桥接名字
local_topic	string MQTT 主题或主题过滤器作为数据源（动作输入）。如果规则动作用作数据源，则应将此配置留空，否则消息将在远程系统中重复。
enable	boolean Default: true 启用（是）或停用（否）此动作。
connector required	string 由动作指定的连接器名称，用于选择外部资源。
tags	Array of strings 连接器的标签
description	string Default: "" 描述性文本。
required	object (bridge_pgsql.action_parameters)
	object (actions_and_sources.action_resource_opts)

Responses

Request samples

Payload

Content type

application/json

Example

Azure Event Hub Action

{"name": "my_azure_event_hub_producer_action",
"type": "azure_event_hub_producer",
"enable": true,
"parameters": {"message": {"value": "${.}",
"key": "${.clientid}"
},
"buffer": {"mode": "hybrid",
"segment_bytes": "100MB",
"per_partition_limit": "2GB",
"memory_overload_protection": true
},
"topic": "topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"kafka_headers": "${.pub_props}",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_header_value_encode_mode": "none",
"partition_strategy": "random",
"partition_count_refresh_interval": "60s"
},
"connector": "my_azure_event_hub_producer_connector",
"local_topic": "mqtt/local/topic"
}

Response samples

400

Content type

application/json

{"code": "TEST_FAILED",
"message": "string"
}

List available action types

列出所有可用的动作类型。

Authorizations:

basicAuthbearerAuth

Responses

Response samples

200

Content type

application/json

["http",
"rocketmq",
"cassandra",
"rabbitmq",
"pulsar",
"greptimedb",
"hstreamdb",
"influxdb",
"mongodb",
"azure_event_hub_producer",
"confluent_producer",
"gcp_pubsub_producer",
"iotdb",
"kafka_producer",
"sqlserver",
"syskeeper_forwarder",
"kinesis",
"opents",
"redis",
"s3",
"mqtt",
"clickhouse",
"tdengine",
"oracle",
"timescale",
"elasticsearch",
"matrix",
"mysql",
"pgsql",
"dynamo"
]

Get bridge

通过 id 获取一个数据桥接

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: http:my_http_action

数据桥接 ID，格式必须为 {type}:{name}。

Responses

Response samples

200
404

Content type

application/json

Example

Azure Event Hub Action

{"name": "my_azure_event_hub_producer_action",
"status": "connected",
"type": "azure_event_hub_producer",
"enable": true,
"parameters": {"message": {"value": "${.}",
"key": "${.clientid}"
},
"buffer": {"mode": "hybrid",
"segment_bytes": "100MB",
"per_partition_limit": "2GB",
"memory_overload_protection": true
},
"topic": "topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"kafka_headers": "${.pub_props}",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_header_value_encode_mode": "none",
"partition_strategy": "random",
"partition_count_refresh_interval": "60s"
},
"connector": "my_azure_event_hub_producer_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"local_topic": "mqtt/local/topic"
}

Update bridge

通过 id 更新数据桥接

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: http:my_http_action

数据桥接 ID，格式必须为 {type}:{name}。

Request Body schema: application/json

One of

local_topic	string MQTT 主题或主题过滤器作为数据源（动作输入）。如果规则动作用作数据源，则应将此配置留空，否则消息将在远程系统中重复。
enable	boolean Default: true 启用（是）或停用（否）此动作。
connector required	string 由动作指定的连接器名称，用于选择外部资源。
tags	Array of strings 连接器的标签
description	string Default: "" 描述性文本。
required	object (bridge_pgsql.action_parameters)
	object (actions_and_sources.action_resource_opts)

Responses

Request samples

Payload

Content type

application/json

Example

Azure Event Hub Action

{"enable": true,
"parameters": {"message": {"value": "${.}",
"key": "${.clientid}"
},
"buffer": {"mode": "hybrid",
"segment_bytes": "100MB",
"per_partition_limit": "2GB",
"memory_overload_protection": true
},
"topic": "topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"kafka_headers": "${.pub_props}",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_header_value_encode_mode": "none",
"partition_strategy": "random",
"partition_count_refresh_interval": "60s"
},
"connector": "my_azure_event_hub_producer_connector",
"local_topic": "mqtt/local/topic"
}

Response samples

200
400
404

Content type

application/json

Example

Azure Event Hub Action

{"name": "my_azure_event_hub_producer_action",
"status": "connected",
"type": "azure_event_hub_producer",
"enable": true,
"parameters": {"message": {"value": "${.}",
"key": "${.clientid}"
},
"buffer": {"mode": "hybrid",
"segment_bytes": "100MB",
"per_partition_limit": "2GB",
"memory_overload_protection": true
},
"topic": "topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"kafka_headers": "${.pub_props}",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_header_value_encode_mode": "none",
"partition_strategy": "random",
"partition_count_refresh_interval": "60s"
},
"connector": "my_azure_event_hub_producer_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"local_topic": "mqtt/local/topic"
}

Delete bridge

通过 id 删除数据桥接

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: http:my_http_action

数据桥接 ID，格式必须为 {type}:{name}。

query Parameters

also_delete_dep_actions

boolean

Default: false

是否级联删除依赖的动作。

Responses

Response samples

400
404
503

Content type

application/json

{"rules": ["string"
],
"code": "BAD_REQUEST",
"message": "string"
}

Get action metrics

通过 id 来获取数据桥接的指标信息。

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: http:my_http_action

数据桥接 ID，格式必须为 {type}:{name}。

Responses

Response samples

200
404

Content type

application/json

{"metrics": {"dropped": 0,
"dropped.other": 0,
"dropped.queue_full": 0,
"dropped.resource_not_found": 0,
"dropped.resource_stopped": 0,
"matched": 0,
"queuing": 0,
"retried": 0,
"failed": 0,
"inflight": 0,
"success": 0,
"rate": 0,
"rate_max": 0,
"rate_last5m": 0,
"received": 0
},
"node_metrics": [{"node": "emqx@127.0.0.1",
"metrics": {"dropped": 0,
"dropped.other": 0,
"dropped.queue_full": 0,
"dropped.resource_not_found": 0,
"dropped.resource_stopped": 0,
"matched": 0,
"queuing": 0,
"retried": 0,
"failed": 0,
"inflight": 0,
"success": 0,
"rate": 0,
"rate_max": 0,
"rate_last5m": 0,
"received": 0
}
}
]
}