EMQX Dedicated REST API (v5)

Download OpenAPI specification:Download

EMQX Dedicated REST API

Authentication

Get user from built-in database authenticator.

Authorizations:

basicAuthbearerAuth

path Parameters

user_id

required

string

User ID.

Responses

Response samples

200
404

Content type

application/json

Example

Regular user

{"user_id": "user1"
}

Update user in built-in database authenticator.

Authorizations:

basicAuthbearerAuth

path Parameters

user_id

required

string

User ID.

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

Delete user in built-in database authenticator.

Authorizations:

basicAuthbearerAuth

path Parameters

user_id

required

string

User ID.

Responses

Response samples

404

Content type

application/json

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

Import users into built-in database authenticator.

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

List users in built-in database authenticator.

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 Fuzzy search user_id (username or clientid).
is_superuser	boolean Is superuser

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
}
}

Create users for built-in database authenticator.

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

Show the list of rules for users

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 Fuzzy search `username` as substring

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
}
}

Add new rule for 'username'

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

Array

	Array of objects (emqx_authz_api_mnesia.rule_item)
username required	string Username

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

Show the list of rules for clients

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 Fuzzy search `clientid` as substring

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
}
}

Add new rule for 'clientid'

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

Array

	Array of objects (emqx_authz_api_mnesia.rule_item)
clientid required	string ClientID

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

Get rule for 'clientid'

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

Example: client1

ClientID

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

Set rule for 'clientid'

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

Example: client1

ClientID

Request Body schema: application/json

	Array of objects (emqx_authz_api_mnesia.rule_item)
clientid required	string ClientID

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

Delete rule for 'clientid'

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

Example: client1

ClientID

Responses

Response samples

400
404

Content type

application/json

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

Get rule for 'username'

Authorizations:

basicAuthbearerAuth

path Parameters

username

required

string

Example: user1

Username

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

Set rule for 'username'

Authorizations:

basicAuthbearerAuth

path Parameters

username

required

string

Example: user1

Username

Request Body schema: application/json

	Array of objects (emqx_authz_api_mnesia.rule_item)
username required	string Username

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

Delete rule for 'username'

Authorizations:

basicAuthbearerAuth

path Parameters

username

required

string

Example: user1

Username

Responses

Response samples

400
404

Content type

application/json

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

Delete all rules for all 'users', 'clients' and 'a

Delete all rules for all 'users', 'clients' and 'all'

Authorizations:

basicAuthbearerAuth

Responses

Response samples

400

Content type

application/json

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

Show the list of rules for '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"
]
}
]
}

Delete rules for 'all'

Authorizations:

basicAuthbearerAuth

Responses

Create/Update the list of rules for '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

List all currently banned client IDs, usernames an

List all currently banned client IDs, usernames and IP addresses.

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
}
}

Clear all banned data.

Authorizations:

basicAuthbearerAuth

Responses

Add a client ID, username or IP address to the bla

Add a client ID, username or IP address to the blacklist.

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

as required	string Enum: "clientid" "username" "peerhost" "clientid_re" "username_re" "peerhost_net" Ban method, which can be exact client ID, client ID regular expression, exact username, username regular expression, IP address or an IP address range.
who required	string Ban object, specific client ID, username or IP address.
by	string Initiator of the ban.
reason	string Ban reason, record the reason why the current object was banned.
	integer or string The start time of the ban, the format is rfc3339, the default is the time when the operation was initiated.
	integer or string The end time of the ban, the format is rfc3339, the default is the time when the operation was initiated + 1 year.

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"
}
]
}

Remove a client ID, username or IP address from th

Remove a client ID, username or IP address from the blacklist.

Authorizations:

basicAuthbearerAuth

path Parameters

as required	string Enum: "clientid" "username" "peerhost" "clientid_re" "username_re" "peerhost_net" Example: username Ban method, which can be exact client ID, client ID regular expression, exact username, username regular expression, IP address or an IP address range.
who required	string Example: Badass Ban object, specific client ID, username or IP address.

Responses

Response samples

404

Content type

application/json

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

Clients

Kick out a batch of client by client IDs

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

Array

string

Responses

Request samples

Payload

Content type

application/json

["emqx_clientid_985bb09d",
"emqx_clientid_211cc01c"
]

Get client authz cache in the cluster.

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
}

Clean client authz cache in the cluster.

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

Responses

Response samples

404

Content type

application/json

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

Get clients info by client ID

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
}

Kick out client by client ID

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

Responses

Response samples

404

Content type

application/json

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

Unsubscribe bulk

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

Subscribe bulk

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
}
]

Subscribe

Authorizations:

basicAuthbearerAuth

path Parameters

clientid

required

string

Request Body schema: application/json

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
}

Get client mqueue messages

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

Get client in-flight messages

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

Unsubscribe

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

Get client subscriptions

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
}
]

List clients

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

List 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

Possible HTTP response status code are:

200: All messages are delivered to at least one subscriber;

202: At least one message was not delivered to any subscriber;

400: At least one message is invalid. For example bad topic name, or QoS is out of range;

503: Failed to deliver at least one of the messages;

In case there is at lest one invalid message in the batch, the HTTP response body
is the same as for /publish API.

Otherwise the HTTP response body is an array of JSON objects indicating the publish
result of each individual message in the batch.

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

Array

payload_encoding	string Default: "plain" Enum: "plain" "base64" MQTT Payload Encoding, `base64` or `plain`. When set to `base64`, the message is decoded before it is published.
topic required	string Topic Name
qos	integer [ 0 .. 2 ] Default: 0 MQTT message QoS
clientid	string Deprecated
payload required	string The MQTT message payload.
	object (emqx_mgmt_api_publish.message_properties)
retain	boolean Default: false A boolean field to indicate if this message should be retained.

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

Possible HTTP status response codes are:

200: The message is delivered to at least one subscriber;

202: No matched subscribers;

400: Message is invalid. for example bad topic name, or QoS is out of range;

503: Failed to deliver the message to subscriber(s)

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

payload_encoding	string Default: "plain" Enum: "plain" "base64" MQTT Payload Encoding, `base64` or `plain`. When set to `base64`, the message is decoded before it is published.
topic required	string Topic Name
qos	integer [ 0 .. 2 ] Default: 0 MQTT message QoS
clientid	string Deprecated
payload required	string The MQTT message payload.
	object (emqx_mgmt_api_publish.message_properties)
retain	boolean Default: false A boolean field to indicate if this message should be retained.

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

List retained messages.

Authorizations:

basicAuthbearerAuth

query Parameters

topic	string Topic filter, supports wildcards, omit this to match all messages.
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
}
}

Delete all retained messages

Authorizations:

basicAuthbearerAuth

Responses

Lookup a message by a topic without wildcards.

Authorizations:

basicAuthbearerAuth

path Parameters

topic

required

string

Topic.

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

Delete matching messages.

Authorizations:

basicAuthbearerAuth

path Parameters

topic

required

string

Topic.

Responses

Response samples

400
404

Content type

application/json

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

Metrics

EMQX metrics

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

[ ]

Current monitor (statistics) data, e.g. number of

Current monitor (statistics) data, e.g. number of connections and connection rate in the whole cluster.

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

Test creating a new connector.

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

One of

type required	string Value: "timescale" The type of the connector.
name required	string The name of the connector.
enable	boolean Default: true Enable (true) or disable (false) this connector.
tags	Array of strings Tags to annotate this config entry.
description	string Default: "" Descriptive text.
server required	string The IPv4 or IPv6 address or the hostname to connect to. A host entry has the following form: `Host[:Port]`. The PostgreSQL default port 5432 is used if `[:Port]` is not specified.
database required	string Database name.
pool_size	integer >= 1 Default: 8 Size of the connection pool towards the bridge target service.
username required	string The username associated with the bridge in the external database used for authentication or identification purposes.
password	string <password> The password associated with the bridge, used for authentication with the external database.
auto_reconnect	boolean Deprecated Default: true Deprecated. Enable automatic reconnect to the database.
	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

Get a connector by id.

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: http:my_http_connector

The connector id. Must be of format {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

Update a connector by id.

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: http:my_http_connector

The connector id. Must be of format {type}:{name}.

Request Body schema: application/json

One of

enable	boolean Default: true Enable (true) or disable (false) this connector.
tags	Array of strings Tags to annotate this config entry.
description	string Default: "" Descriptive text.
server required	string The IPv4 or IPv6 address or the hostname to connect to. A host entry has the following form: `Host[:Port]`. The PostgreSQL default port 5432 is used if `[:Port]` is not specified.
database required	string Database name.
pool_size	integer >= 1 Default: 8 Size of the connection pool towards the bridge target service.
username required	string The username associated with the bridge in the external database used for authentication or identification purposes.
password	string <password> The password associated with the bridge, used for authentication with the external database.
auto_reconnect	boolean Deprecated Default: true Deprecated. Enable automatic reconnect to the database.
	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

Delete a connector by id.

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: http:my_http_connector

The connector id. Must be of format {type}:{name}.

Responses

Response samples

400
404
503

Content type

application/json

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

Manually start a connector

Start connector on all nodes in the cluster.

Authorizations:

basicAuthbearerAuth

path Parameters

id required	string Example: http:my_http_connector The connector id. Must be of format {type}:{name}.
operation required	string Value: "start" Example: start Operation can be one of: 'start'.

Responses

Response samples

400
404
501
503

Content type

application/json

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

Enable or disable connector

Enable or Disable connector on all nodes in the cluster.

Authorizations:

basicAuthbearerAuth

path Parameters

id required	string Example: http:my_http_connector The connector id. Must be of format {type}:{name}.
enable required	boolean Example: true Whether to enable this connector.

Responses

Response samples

404
503

Content type

application/json

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

List connectors

List all created 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"
}
],
"acceptors": 16,
"handshake_timeout": "16s",
"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

Create a new connector by type and name.

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

One of

type required	string Value: "timescale" The type of the connector.
name required	string The name of the connector.
enable	boolean Default: true Enable (true) or disable (false) this connector.
tags	Array of strings Tags to annotate this config entry.
description	string Default: "" Descriptive text.
server required	string The IPv4 or IPv6 address or the hostname to connect to. A host entry has the following form: `Host[:Port]`. The PostgreSQL default port 5432 is used if `[:Port]` is not specified.
database required	string Database name.
pool_size	integer >= 1 Default: 8 Size of the connection pool towards the bridge target service.
username required	string The username associated with the bridge in the external database used for authentication or identification purposes.
password	string <password> The password associated with the bridge, used for authentication with the external database.
auto_reconnect	boolean Deprecated Default: true Deprecated. Enable automatic reconnect to the database.
	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: {}

The context of the event for testing

sql

required

string

The SQL of the rule for testing

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

List all rules

Authorizations:

basicAuthbearerAuth

query Parameters

enable	boolean Filter enable/disable rules
from	string Filter rules by from(topic), exact match
like_id	string Filter rules by id, Substring matching
like_from	string Filter rules by from(topic), Substring matching
like_description	string Filter rules by description, Substring matching
match_from	string Filter rules by from(topic), Mqtt topic matching
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

Create a new rule using given Id

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

name	string Default: "" The name of the rule
sql required	string SQL query to transform the messages. Example: `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: [] A list of actions of the rule. An action can be a string that refers to the channel ID of an EMQX bridge, or an object that refers to a function. There a some built-in functions like "republish" and "console", and we also support user provided functions in the format: "{module}:{function}". The actions in the list are executed sequentially. This means that if one of the action is executing slowly, all the following actions will not be executed until it returns. If one of the action crashed, all other actions come after it will still be executed, in the original order. If there's any error when running an action, there will be an error message, and the 'failure' counter of the function action or the bridge channel will increase.
enable	boolean Default: true Enable or disable the rule
description	string Default: "" The description of the rule
metadata	object Rule metadata, do not change manually

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

Get a rule's metrics by given 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

List all events can be used in rules

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

Reset a 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

Get a rule by given 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

Update a rule by given Id to all nodes in the cluster

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: my_rule_id

Request Body schema: application/json

name	string Default: "" The name of the rule
sql required	string SQL query to transform the messages. Example: `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: [] A list of actions of the rule. An action can be a string that refers to the channel ID of an EMQX bridge, or an object that refers to a function. There a some built-in functions like "republish" and "console", and we also support user provided functions in the format: "{module}:{function}". The actions in the list are executed sequentially. This means that if one of the action is executing slowly, all the following actions will not be executed until it returns. If one of the action crashed, all other actions come after it will still be executed, in the original order. If there's any error when running an action, there will be an error message, and the 'failure' counter of the function action or the bridge channel will increase.
enable	boolean Default: true Enable or disable the rule
description	string Default: "" The description of the rule
metadata	object Rule metadata, do not change manually

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

Delete a rule by given Id from all nodes in the cluster

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

Reset a bridge metrics by id.

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: http:my_http_action

The bridge id. Must be of format {type}:{name}.

Responses

Response samples

404

Content type

application/json

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

Manually start a bridge

Start bridge on all nodes in the cluster.

Authorizations:

basicAuthbearerAuth

path Parameters

id required	string Example: http:my_http_action The bridge id. Must be of format {type}:{name}.
operation required	string Value: "start" Example: start Operation can be one of: 'start'.

Responses

Response samples

400
404
501
503

Content type

application/json

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

Enable or disable bridge

Enable or Disable bridge on all nodes in the cluster.

Authorizations:

basicAuthbearerAuth

path Parameters

id required	string Example: http:my_http_action The bridge id. Must be of format {type}:{name}.
enable required	boolean Example: true Whether to enable this bridge.

Responses

Response samples

404
503

Content type

application/json

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

List bridges

List all created 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",
"memory_overload_protection": true,
"segment_bytes": "100MB",
"per_partition_limit": "2GB"
},
"topic": "topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"partition_count_refresh_interval": "60s",
"kafka_header_value_encode_mode": "none",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_headers": "${.pub_props}",
"partition_strategy": "random"
},
"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",
"memory_overload_protection": true,
"segment_bytes": "100MB",
"per_partition_limit": "2GB"
},
"topic": "topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"partition_count_refresh_interval": "60s",
"kafka_header_value_encode_mode": "none",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_headers": "${.pub_props}",
"partition_strategy": "random"
},
"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",
"memory_overload_protection": true,
"segment_bytes": "100MB",
"per_partition_limit": "2GB"
},
"compression": "no_compression",
"topic": "kafka-topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"partitions_limit": "all_partitions",
"partition_count_refresh_interval": "60s",
"kafka_header_value_encode_mode": "none",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_headers": "${pub_props}",
"partition_strategy": "random"
},
"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

Create a new bridge by type and name.

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

One of

type required	string Value: "timescale" The Bridge Type
name required	string Bridge name.
local_topic	string MQTT topic or topic filter as data source (action input). If rule action is used as data source, this config should be left empty, otherwise messages will be duplicated in the remote system.
enable	boolean Default: true Enable (true) or disable (false) this action.
connector required	string Name of the connector specified by the action, used for external resource selection.
tags	Array of strings Tags to annotate this config entry.
description	string Default: "" Descriptive text.
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",
"memory_overload_protection": true,
"segment_bytes": "100MB",
"per_partition_limit": "2GB"
},
"topic": "topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"partition_count_refresh_interval": "60s",
"kafka_header_value_encode_mode": "none",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_headers": "${.pub_props}",
"partition_strategy": "random"
},
"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",
"memory_overload_protection": true,
"segment_bytes": "100MB",
"per_partition_limit": "2GB"
},
"topic": "topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"partition_count_refresh_interval": "60s",
"kafka_header_value_encode_mode": "none",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_headers": "${.pub_props}",
"partition_strategy": "random"
},
"connector": "my_azure_event_hub_producer_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"local_topic": "mqtt/local/topic"
}

Test creating bridge

Test creating a new bridge.

Authorizations:

basicAuthbearerAuth

Request Body schema: application/json

One of

type required	string Value: "timescale" The Bridge Type
name required	string Bridge name.
local_topic	string MQTT topic or topic filter as data source (action input). If rule action is used as data source, this config should be left empty, otherwise messages will be duplicated in the remote system.
enable	boolean Default: true Enable (true) or disable (false) this action.
connector required	string Name of the connector specified by the action, used for external resource selection.
tags	Array of strings Tags to annotate this config entry.
description	string Default: "" Descriptive text.
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",
"memory_overload_protection": true,
"segment_bytes": "100MB",
"per_partition_limit": "2GB"
},
"topic": "topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"partition_count_refresh_interval": "60s",
"kafka_header_value_encode_mode": "none",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_headers": "${.pub_props}",
"partition_strategy": "random"
},
"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

Lists the 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

Get a bridge by id.

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: http:my_http_action

The bridge id. Must be of format {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",
"memory_overload_protection": true,
"segment_bytes": "100MB",
"per_partition_limit": "2GB"
},
"topic": "topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"partition_count_refresh_interval": "60s",
"kafka_header_value_encode_mode": "none",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_headers": "${.pub_props}",
"partition_strategy": "random"
},
"connector": "my_azure_event_hub_producer_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"local_topic": "mqtt/local/topic"
}

Update bridge

Update a bridge by id.

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: http:my_http_action

The bridge id. Must be of format {type}:{name}.

Request Body schema: application/json

One of

local_topic	string MQTT topic or topic filter as data source (action input). If rule action is used as data source, this config should be left empty, otherwise messages will be duplicated in the remote system.
enable	boolean Default: true Enable (true) or disable (false) this action.
connector required	string Name of the connector specified by the action, used for external resource selection.
tags	Array of strings Tags to annotate this config entry.
description	string Default: "" Descriptive text.
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",
"memory_overload_protection": true,
"segment_bytes": "100MB",
"per_partition_limit": "2GB"
},
"topic": "topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"partition_count_refresh_interval": "60s",
"kafka_header_value_encode_mode": "none",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_headers": "${.pub_props}",
"partition_strategy": "random"
},
"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",
"memory_overload_protection": true,
"segment_bytes": "100MB",
"per_partition_limit": "2GB"
},
"topic": "topic",
"max_inflight": 10,
"required_acks": "all_isr",
"max_batch_bytes": "896KB",
"partition_count_refresh_interval": "60s",
"kafka_header_value_encode_mode": "none",
"kafka_ext_headers": [{"kafka_ext_header_value": "${clientid}",
"kafka_ext_header_key": "clientid"
},
{"kafka_ext_header_value": "${topic}",
"kafka_ext_header_key": "topic"
}
],
"kafka_headers": "${.pub_props}",
"partition_strategy": "random"
},
"connector": "my_azure_event_hub_producer_connector",
"node_status": [{"node": "emqx@localhost",
"status": "connected"
}
],
"local_topic": "mqtt/local/topic"
}

Delete bridge

Delete a bridge by id.

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: http:my_http_action

The bridge id. Must be of format {type}:{name}.

query Parameters

also_delete_dep_actions

boolean

Default: false

Whether to cascade delete dependent actions.

Responses

Response samples

400
404
503

Content type

application/json

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

Get action metrics

Get bridge metrics by id.

Authorizations:

basicAuthbearerAuth

path Parameters

id

required

string

Example: http:my_http_action

The bridge id. Must be of format {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
}
}
]
}