启用集群中所有节点上的数据桥接。
id required | string Example: http:my_http_action 数据桥接 ID,格式必须为 {type}:{name}。 |
operation required | string Value: "start" Example: start 集群可用操作:'启动'。 |
{- "code": "BAD_REQUEST",
- "message": "string"
}
启用或禁用集群内所有节点上的数据桥接。
id required | string Example: http:my_http_action 数据桥接 ID,格式必须为 {type}:{name}。 |
enable required | boolean Example: true 是否启用该数据桥接。 |
{- "code": "NOT_FOUND",
- "message": "string"
}
[- {
- "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": "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_linger_bytes": "10MB",
- "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",
- "max_linger_time": "5ms"
}, - "connector": "my_confluent_producer_connector",
- "node_status": [
- {
- "node": "emqx@localhost",
- "status": "connected"
}
], - "local_topic": "mqtt/local/topic"
}, - {
- "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"
}
}, - {
- "status": "connected",
- "description": "my action",
- "enable": true,
- "parameters": {
- "mode": "direct",
- "content": "${.payload}",
- "blob": "${.payload.blob}",
- "container": "${.payload.container}"
}, - "connector": "my_connector",
- "node_status": [
- {
- "node": "emqx@localhost",
- "status": "connected"
}
], - "resource_opts": {
- "batch_size": 1,
- "worker_pool_size": 16,
- "query_mode": "sync",
- "batch_time": "0ms",
- "inflight_window": 100,
- "request_ttl": "45s",
- "health_check_interval": "30s"
}
}, - {
- "status": "connected",
- "description": "my action",
- "enable": true,
- "parameters": {
- "sql": "insert into mqtt (key, value) values (${.id}, ${.payload})"
}, - "connector": "my_connector",
- "node_status": [
- {
- "node": "emqx@localhost",
- "status": "connected"
}
], - "resource_opts": {
- "worker_pool_size": 16,
- "query_mode": "sync",
- "inflight_window": 100,
- "request_ttl": "45s",
- "health_check_interval": "30s"
}
}, - {
- "name": "datalayers_action",
- "status": "connected",
- "type": "datalayers",
- "description": "My example datalayers action",
- "enable": true,
- "parameters": {
- "precision": "ms",
- "write_syntax": "${topic},clientid=${clientid} payload=${payload},${clientid}_int_value=${payload.int_key}i,bool=${payload.bool}"
}, - "connector": "datalayers_connector",
- "node_status": [
- {
- "node": "emqx@localhost",
- "status": "connected"
}
], - "resource_opts": {
- "health_check_interval": "30s"
}
}, - {
- "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_linger_bytes": "10MB",
- "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",
- "max_linger_time": "5ms"
}, - "connector": "my_azure_event_hub_producer_connector",
- "node_status": [
- {
- "node": "emqx@localhost",
- "status": "connected"
}
], - "local_topic": "mqtt/local/topic"
}, - {
- "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": "10MB",
- "per_partition_limit": "2GB"
}, - "compression": "no_compression",
- "topic": "kafka-topic",
- "max_inflight": 10,
- "required_acks": "all_isr",
- "max_linger_bytes": "10MB",
- "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",
- "max_linger_time": "5ms"
}, - "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": "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": "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": "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": "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,
- "publish_confirmation_timeout": "30s",
- "delivery_mode": "non_persistent"
}, - "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": "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": {
- "hash_key": "clientid",
- "table": "mqtt_msg",
- "template": ""
}
}, - {
- "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"
}
}, - {
- "status": "connected",
- "description": "My upload action",
- "enable": true,
- "parameters": {
- "mode": "direct",
- "key": "${topic}",
- "content": "${payload}",
- "bucket": "${clientid}",
- "acl": "public_read"
}, - "connector": "my_s3_connector",
- "node_status": [
- {
- "node": "emqx@localhost",
- "status": "connected"
}
], - "resource_opts": {
- "query_mode": "sync",
- "inflight_window": 10
}
}, - {
- "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": "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"
}
}, - {
- "status": "connected",
- "description": "my action",
- "enable": true,
- "parameters": {
- "mode": "aggregated",
- "connect_timeout": "15s",
- "pool_size": 16,
- "private_key": "file:///path/to/secret.pem",
- "max_retries": 3,
- "schema": "public",
- "database": "testdatabase",
- "stage": "teststage",
- "pipe": "testpipe",
- "pipelining": 100,
- "aggregation": {
- "container": {
- "type": "csv"
}, - "max_records": 1000,
- "time_interval": "60s"
}, - "pipe_user": "pipeuser"
}, - "connector": "my_connector",
- "node_status": [
- {
- "node": "emqx@localhost",
- "status": "connected"
}
], - "resource_opts": {
- "batch_size": 10000,
- "worker_pool_size": 16,
- "query_mode": "sync",
- "batch_time": "60s",
- "inflight_window": 100,
- "request_ttl": "45s",
- "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": "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": "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": "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": "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)"
}
}, - {
- "status": "connected",
- "description": "my action",
- "enable": true,
- "parameters": {
- "mode": "aggregated",
- "blob": "${action}/${node}/${datetime.rfc3339}/${sequence}",
- "container": "mycontainer",
- "aggregation": {
- "container": {
- "type": "csv",
- "column_order": [
- "a",
- "b"
]
}, - "max_records": 10000,
- "time_interval": "4s"
}
}, - "connector": "my_connector",
- "node_status": [
- {
- "node": "emqx@localhost",
- "status": "connected"
}
], - "resource_opts": {
- "batch_size": 100,
- "worker_pool_size": 16,
- "query_mode": "sync",
- "batch_time": "10ms",
- "inflight_window": 100,
- "request_ttl": "45s",
- "health_check_interval": "30s"
}
}, - {
- "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"
}
}, - {
- "name": "pulsar_action",
- "status": "connected",
- "type": "pulsar",
- "description": "My example pulsar action",
- "enable": true,
- "parameters": {
- "message": {
- "value": "${.}",
- "key": "${.clientid}"
}, - "sync_timeout": "5s",
- "pulsar_topic": "test_topic"
}, - "connector": "pulsar_connector",
- "node_status": [
- {
- "node": "emqx@localhost",
- "status": "connected"
}
], - "resource_opts": {
- "health_check_interval": "30s"
}
}, - {
- "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
}
}, - {
- "status": "connected",
- "description": "My aggregated upload action",
- "enable": true,
- "parameters": {
- "mode": "aggregated",
- "key": "${action}/${node}/${datetime.rfc3339utc}_N${sequence}.csv",
- "bucket": "mqtt-aggregated",
- "acl": "public_read",
- "aggregation": {
- "max_records": 100000,
- "time_interval": "15m"
}, - "container": {
- "type": "csv",
- "column_order": [
- "clientid",
- "topic",
- "publish_received_at"
]
}
}, - "connector": "my_s3_connector",
- "node_status": [
- {
- "node": "emqx@localhost",
- "status": "connected"
}
], - "resource_opts": {
- "query_mode": "async",
- "inflight_window": 100,
- "health_check_interval": "10s"
}
}, - {
- "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": "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": "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)"
}
}
]
通过类型和名称创建一个新的数据桥接。
type required | string Value: "timescale" Bridge 类型 |
name required | string 桥接名字 |
local_topic | string MQTT 主题或主题过滤器作为数据源(动作输入)。 如果规则动作用作数据源,则应将此配置留空,否则消息将在远程系统中重复。 |
required | object (bridge_pgsql.action_parameters) |
enable | boolean Default: true 启用(是)或停用(否)此动作。 |
connector required | string 由动作指定的连接器名称,用于选择外部资源。 |
tags | Array of strings 连接器的标签 |
description | string Default: "" 描述性文本。 |
object (actions_and_sources.action_resource_opts) |
{- "name": "my_http_action",
- "type": "http",
- "enable": true,
- "parameters": {
- "path": "/room/${room_no}",
- "body": "${.}",
- "headers": { },
- "method": "post"
}, - "connector": "my_http_connector",
- "resource_opts": {
- "worker_pool_size": 16,
- "query_mode": "async",
- "health_check_interval": "15s"
}
}
{- "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"
}
}
测试创建一个新的数据桥接。
type required | string Value: "timescale" Bridge 类型 |
name required | string 桥接名字 |
local_topic | string MQTT 主题或主题过滤器作为数据源(动作输入)。 如果规则动作用作数据源,则应将此配置留空,否则消息将在远程系统中重复。 |
required | object (bridge_pgsql.action_parameters) |
enable | boolean Default: true 启用(是)或停用(否)此动作。 |
connector required | string 由动作指定的连接器名称,用于选择外部资源。 |
tags | Array of strings 连接器的标签 |
description | string Default: "" 描述性文本。 |
object (actions_and_sources.action_resource_opts) |
{- "name": "my_http_action",
- "type": "http",
- "enable": true,
- "parameters": {
- "path": "/room/${room_no}",
- "body": "${.}",
- "headers": { },
- "method": "post"
}, - "connector": "my_http_connector",
- "resource_opts": {
- "worker_pool_size": 16,
- "query_mode": "async",
- "health_check_interval": "15s"
}
}
{- "code": "TEST_FAILED",
- "message": "string"
}
[- "http",
- "rocketmq",
- "rabbitmq",
- "cassandra",
- "influxdb",
- "greptimedb",
- "iotdb",
- "pulsar",
- "hstreamdb",
- "confluent_producer",
- "syskeeper_forwarder",
- "gcp_pubsub_producer",
- "mongodb",
- "azure_event_hub_producer",
- "kafka_producer",
- "sqlserver",
- "opents",
- "kinesis",
- "datalayers",
- "s3",
- "redis",
- "mqtt",
- "clickhouse",
- "snowflake",
- "couchbase",
- "azure_blob_storage",
- "timescale",
- "tdengine",
- "oracle",
- "matrix",
- "elasticsearch",
- "pgsql",
- "mysql",
- "dynamo"
]
通过 id 获取一个数据桥接
id required | string Example: http:my_http_action 数据桥接 ID,格式必须为 {type}:{name}。 |
{- "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"
}
}
通过 id 更新数据桥接
id required | string Example: http:my_http_action 数据桥接 ID,格式必须为 {type}:{name}。 |
local_topic | string MQTT 主题或主题过滤器作为数据源(动作输入)。 如果规则动作用作数据源,则应将此配置留空,否则消息将在远程系统中重复。 |
required | object (bridge_pgsql.action_parameters) |
enable | boolean Default: true 启用(是)或停用(否)此动作。 |
connector required | string 由动作指定的连接器名称,用于选择外部资源。 |
tags | Array of strings 连接器的标签 |
description | string Default: "" 描述性文本。 |
object (actions_and_sources.action_resource_opts) |
{- "enable": true,
- "parameters": {
- "path": "/room/${room_no}",
- "body": "${.}",
- "headers": { },
- "method": "post"
}, - "connector": "my_http_connector",
- "resource_opts": {
- "worker_pool_size": 16,
- "query_mode": "async",
- "health_check_interval": "15s"
}
}
{- "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"
}
}
通过 id 删除数据桥接
id required | string Example: http:my_http_action 数据桥接 ID,格式必须为 {type}:{name}。 |
also_delete_dep_actions | boolean Default: false 是否级联删除依赖的动作。 |
{- "rules": [
- "string"
], - "code": "BAD_REQUEST",
- "message": "string"
}
通过 id 来获取数据桥接的指标信息。
id required | string Example: http:my_http_action 数据桥接 ID,格式必须为 {type}:{name}。 |
{- "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
}
}
]
}
更新内置数据库认证器中的指定用户数据。
user_id required | string 用户 ID,根据设置的账号类型不同,可以是 username 或 clientid |
password required | string |
is_superuser | boolean Default: false |
{- "password": "******"
}
{- "regular_user": {
- "value": {
- "user_id": "user1"
}, - "summary": "Regular user"
}, - "super_user": {
- "value": {
- "is_superuser": true,
- "user_id": "user2"
}, - "summary": "Superuser"
}
}
为内置数据库认证器导入用户数据。
type required | string Enum: "plain" "hash" Example: type=hash The import file template type, enum with |
Import body
[- {
- "is_superuser": true,
- "password": "password1",
- "user_id": "user1"
}, - {
- "is_superuser": false,
- "password": "password2",
- "user_id": "user2"
}
]
{- "code": "BAD_REQUEST",
- "message": "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) |
like_user_id | string 使用用户 ID (username 或 clientid)模糊查询。 |
is_superuser | boolean 是否是超级用户。 |
{- "data": [
- {
- "user_id": "user1"
}, - {
- "is_superuser": true,
- "user_id": "user2"
}
], - "meta": {
- "count": 300,
- "limit": 20,
- "page": 0
}
}
向内置数据库认证器添加用户数据。
user_id required | string |
password required | string |
is_superuser | boolean Default: false |
{- "password": "******",
- "user_id": "user1"
}
{- "user_id": "user1"
}
获取指定客户端的详细信息
clientid required | string |
{- "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",
- "durable": false,
- "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
}
批量取消订阅
clientid required | string |
topic | string Topic |
[- {
- "topic": "testtopic/#"
}
]
{- "code": "CLIENTID_NOT_FOUND",
- "message": "string"
}
批量订阅
clientid required | string |
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 |
[- {
- "topic": "testtopic/#",
- "qos": 0,
- "nl": 0,
- "rap": 0,
- "rh": 0
}
]
[- {
- "node": "emqx@127.0.0.1",
- "topic": "testtopic/1",
- "clientid": "emqx_clientid_xx128cdhfc",
- "qos": 0,
- "nl": 0,
- "rap": 0,
- "rh": 0,
- "durable": false
}
]
订阅
clientid required | string |
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 |
{- "topic": "testtopic/#",
- "qos": 0,
- "nl": 0,
- "rap": 0,
- "rh": 0
}
{- "node": "emqx@127.0.0.1",
- "topic": "testtopic/1",
- "clientid": "emqx_clientid_xx128cdhfc",
- "qos": 0,
- "nl": 0,
- "rap": 0,
- "rh": 0,
- "durable": false
}
获取客户端消息队列中的消息。
clientid required | string |
payload | string Default: "base64" Enum: "none" "base64" "plain" Client's inflight/mqueue messages payload encoding. If set to |
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}" | |
limit | integer [ 1 .. 10000 ] Default: 100 Example: limit=50 Results per page(max 10000) |
{- "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"
}
}
获取客户端未确认消息。
clientid required | string |
payload | string Default: "base64" Enum: "none" "base64" "plain" Client's inflight/mqueue messages payload encoding. If set to |
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}" | |
limit | integer [ 1 .. 10000 ] Default: 100 Example: limit=50 Results per page(max 10000) |
{- "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"
}
}
取消订阅
clientid required | string |
topic | string Topic |
{- "topic": "testtopic/#"
}
{- "code": "CLIENTID_NOT_FOUND",
- "message": "string"
}
获取指定客户端的订阅列表
clientid required | string |
[- {
- "node": "emqx@127.0.0.1",
- "topic": "testtopic/1",
- "clientid": "emqx_clientid_xx128cdhfc",
- "qos": 0,
- "nl": 0,
- "rap": 0,
- "rh": 0,
- "durable": false
}
]
列出客户端
page | integer >= 1 Default: 1 Example: page=1 Page number of the results to fetch. |
node | string Example: node=emqx@127.0.0.1 Node name |
limit | integer [ 1 .. 10000 ] Default: 100 Example: limit=50 Results per page(max 10000) |
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 |
like_username | string Fuzzy search |
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 |
{- "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",
- "durable": false,
- "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
}
}
测试创建一个新的连接器。
type required | string Value: "timescale" 连接器的类型。 |
name required | string 连接器的名称 |
enable | boolean Default: true 启用 (是) 或 停用 (否) 该连接器。 |
tags | Array of strings 连接器的标签 |
description | string Default: "" 描述性文本。 |
server required | string 用于连接的 IPv4 或 IPv6 地址或主机名。 |
disable_prepared_statements | boolean Default: false 在连接中禁用预处理语句的使用。 |
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) |
{- "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"
}
}
{- "code": "TEST_FAILED",
- "message": "string"
}
通过 id 获取一个连接器。
id required | string Example: http:my_http_connector 连接器 id。必须是 {type}:{name} 的格式。 |
{- "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"
]
}
通过 id 更新一个连接器。
id required | string Example: http:my_http_connector 连接器 id。必须是 {type}:{name} 的格式。 |
enable | boolean Default: true 启用 (是) 或 停用 (否) 该连接器。 |
tags | Array of strings 连接器的标签 |
description | string Default: "" 描述性文本。 |
server required | string 用于连接的 IPv4 或 IPv6 地址或主机名。 |
disable_prepared_statements | boolean Default: false 在连接中禁用预处理语句的使用。 |
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) |
{- "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"
}
}
{- "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"
]
}
在集群的所有节点上启动连接器。
id required | string Example: http:my_http_connector 连接器 id。必须是 {type}:{name} 的格式。 |
operation required | string Value: "start" Example: start 集群可用操作:'start'。 |
{- "code": "BAD_REQUEST",
- "message": "string"
}
在集群的所有节点上启用或禁用连接器。
id required | string Example: http:my_http_connector 连接器 id。必须是 {type}:{name} 的格式。 |
enable required | boolean Example: true 是否启用此连接器。 |
{- "code": "NOT_FOUND",
- "message": "string"
}
[- {
- "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"
}, - "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,
- "iotdb_version": "v1.3.x"
}, - {
- "status": "connected",
- "ssl": {
- "enable": true
}, - "description": "My connector",
- "server": "couchbase:8093",
- "enable": true,
- "password": "******",
- "username": "admin",
- "node_status": [
- {
- "node": "emqx@localhost",
- "status": "connected"
}
], - "resource_opts": {
- "start_timeout": "5s",
- "start_after_created": true,
- "health_check_interval": "45s"
}
}, - {
- "name": "datalayers_connector",
- "status": "connected",
- "type": "datalayers",
- "ssl": {
- "enable": false
}, - "description": "My example datalayers connector",
- "server": "127.0.0.1:8361",
- "enable": true,
- "parameters": {
- "password": "******",
- "username": "example_username",
- "database": "example_database"
}, - "node_status": [
- {
- "node": "emqx@localhost",
- "status": "connected"
}
], - "actions": [
- "my_action"
]
}, - {
- "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",
- "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,
- "region": "us-west-2",
- "resource_opts": {
- "health_check_interval": "15s",
- "start_timeout": "5s"
},
}, - {
- "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,
- "enable": true,
- "node_status": [
- {
- "node": "emqx@localhost",
- "status": "connected"
}
], - "actions": [
- "my_action"
]
}, - {
- "status": "connected",
- "description": "My connector",
- "enable": true,
- "node_status": [
- {
- "node": "emqx@localhost",
- "status": "connected"
}
], - "account_key": "******",
- "resource_opts": {
- "start_timeout": "5s",
- "start_after_created": true,
- "health_check_interval": "45s"
}, - "account_name": "my_account_name"
}, - {
- "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"
}, - {
- "status": "connected",
- "description": "My connector",
- "pool_size": 8,
- "server": "myorg-myaccount.snowflakecomputing.com",
- "enable": true,
- "password": "******",
- "username": "admin",
- "node_status": [
- {
- "node": "emqx@localhost",
- "status": "connected"
}
], - "resource_opts": {
- "start_timeout": "5s",
- "start_after_created": true,
- "health_check_interval": "45s"
}, - "account": "myorg-myaccount",
- "dsn": "snowflake"
}, - {
- "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"
},
}, - {
- "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"
}
], - "servers": "127.0.0.1:9042",
- "actions": [
- "my_action"
], - "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",
- "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",
}
}, - {
- "name": "clickhouse_connector",
- "status": "connected",
- "type": "clickhouse",
- "description": "My example clickhouse connector",
- "pool_size": 8,
- "password": "******",
- "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"
}
], - "servers": "pulsar://127.0.0.1:6650",
- "actions": [
- "my_action"
]
}, - {
- "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",
- "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",
}
}, - {
- "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,
- "servers": "127.0.0.1:6379,127.0.0.2:6379",
- "sentinel": "myredismaster",
- "redis_type": "sentinel"
}, - "node_status": [
- {
- "node": "emqx@localhost",
- "status": "connected"
}
]
}
]
通过类型和名称创建一个新的连接器。
type required | string Value: "timescale" 连接器的类型。 |
name required | string 连接器的名称 |
enable | boolean Default: true 启用 (是) 或 停用 (否) 该连接器。 |
tags | Array of strings 连接器的标签 |
description | string Default: "" 描述性文本。 |
server required | string 用于连接的 IPv4 或 IPv6 地址或主机名。 |
disable_prepared_statements | boolean Default: false 在连接中禁用预处理语句的使用。 |
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) |
{- "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"
}
}
{- "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_avro_schema",
- "type": "avro",
- "description": "My Avro Schema",
- "source": "{\"type\":\"record\",\"fields\":[{\"type\":\"int\",\"name\":\"i\"},{\"type\":\"string\",\"name\":\"s\"}]}"
}
]
注册一个新的 schema。
name required | string Schema 的一个名称,将用作其标识符。 |
type required | string Value: "json" 对于 JSON schema,必须是 json。 |
source required | string Schema 的源文本。 |
description | string Default: "" 改 schema 的描述。 |
{- "name": "my_avro_schema",
- "type": "avro",
- "description": "My Avro Schema",
- "source": "{\"type\":\"record\",\"fields\":[{\"type\":\"int\",\"name\":\"i\"},{\"type\":\"string\",\"name\":\"s\"}]}"
}
{- "name": "my_avro_schema",
- "type": "avro",
- "description": "My Avro Schema",
- "source": "{\"type\":\"record\",\"fields\":[{\"type\":\"int\",\"name\":\"i\"},{\"type\":\"string\",\"name\":\"s\"}]}"
}
通过名称获取一个 schema。
name required | string Example: my_schema Schema 的名称。 |
{- "name": "my_avro_schema",
- "type": "avro",
- "description": "My Avro Schema",
- "source": "{\"type\":\"record\",\"fields\":[{\"type\":\"int\",\"name\":\"i\"},{\"type\":\"string\",\"name\":\"s\"}]}"
}
更新一个现有的 schema。
name required | string Example: my_schema Schema 的名称。 |
type required | string Value: "json" 对于 JSON schema,必须是 json。 |
source required | string Schema 的源文本。 |
description | string Default: "" 改 schema 的描述。 |
{- "name": "my_avro_schema",
- "type": "avro",
- "description": "My Avro Schema",
- "source": "{\"type\":\"record\",\"fields\":[{\"type\":\"int\",\"name\":\"i\"},{\"type\":\"string\",\"name\":\"s\"}]}"
}
{- "name": "my_avro_schema",
- "type": "avro",
- "description": "My Avro Schema",
- "source": "{\"type\":\"record\",\"fields\":[{\"type\":\"int\",\"name\":\"i\"},{\"type\":\"string\",\"name\":\"s\"}]}"
}
批量发布多条消息。
可能的 HTTP 状态码如下:
200: 所有的消息都被成功发送到至少一个订阅。
202: 至少有一个消息没有匹配到任何订阅。
400: 至少有一个消息编码错误,如非法主题,或 QoS 超出范围等。
503: 至少有一个小因为服务重启的原因导致转发失败。
请求的 Body 或者 Body 中包含的某个消息无法通过 API 规范的类型检查时,HTTP 响应的消息与发布单个消息的 API
/publish
是一样的。
如果所有的消息都是合法的,那么 HTTP 返回的内容是一个 JSON 数组,每个元素代表了该消息转发的状态。
payload_encoding | string Default: "plain" Enum: "plain" "base64" MQTT 消息体的编码方式,可以是 |
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 保留消息,布尔型字段,用于表示该消息是否是保留消息。 |
[- {
- "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
}
]
[- {
- "id": "string"
}
]
发布单条信息。
可能的 HTTP 状态码如下:
200: 消息被成功发送到至少一个订阅。
202: 没有匹配到任何订阅。
400: 消息编码错误,如非法主题,或 QoS 超出范围等。
503: 服务重启等过程中导致转发失败。
payload_encoding | string Default: "plain" Enum: "plain" "base64" MQTT 消息体的编码方式,可以是 |
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 保留消息,布尔型字段,用于表示该消息是否是保留消息。 |
{- "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
}
{- "id": "string"
}
依据指定的消息内容和运行环境执行相应的规则。
id required | string Example: my_rule_id |
rule_engine.ctx_message_transformation_failed (object) or rule_engine.ctx_schema_validation_failed (object) or rule_engine.ctx_delivery_dropped (object) or rule_engine.ctx_bridge_mqtt (object) or rule_engine.ctx_check_authn_complete (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: {} 测试事件的上下文 | |
stop_action_after_template_rendering | boolean Default: true 如果操作在模板渲染后应该停止,请将此设置为 true(默认为 true)。 |
{- "context": { },
- "stop_action_after_template_rendering": true
}
{- "code": "BAD_REQUEST",
- "message": "string"
}
列出所有规则
enable | boolean 根据规则是否开启条件过滤 |
from | string 根据规则来源 Topic 过滤, 需要完全匹配 |
like_id | string 根据规则 id 过滤, 使用子串模糊匹配 |
like_from | string 根据规则来源 Topic 过滤, 使用子串模糊匹配 |
like_description | string 根据规则描述过滤, 使用子串模糊匹配 |
match_from | string 根据规则来源 Topic 过滤, 使用 MQTT Topic 匹配 |
action | Array of strings 筛选包含任意给定动作 ID 的规则。当与 source ID 筛选一起使用时,规则必须包含匹配某些条件的 source 和动作。 |
source | Array of strings 筛选包含任意给定 source ID 的规则。当与动作 ID 筛选一起使用时,规则必须包含匹配某些条件的 source 和动作。 |
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) |
{- "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
}
}
通过指定 ID 创建规则
name | string Default: "" 规则名字 |
sql required | string 用于处理消息的 SQL 。 |
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: [] 规则的动作列表。 | |
enable | boolean Default: true 启用或禁用规则引擎 |
description | string Default: "" 规则的描述 |
metadata | object 规则的元数据,不要手动修改 |
{- "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": { }
}
{- "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": { }
}
通过给定的 Id 获得规则的指标数据
id required | string Example: my_rule_id |
{- "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,
- "actions.discarded": 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,
- "actions.discarded": 0
}
]
}
通过 ID 查询规则
id required | string Example: my_rule_id |
{- "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": { }
}
通过 ID 更新集群里所有节点上的规则
id required | string Example: my_rule_id |
name | string Default: "" 规则名字 |
sql required | string 用于处理消息的 SQL 。 |
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: [] 规则的动作列表。 | |
enable | boolean Default: true 启用或禁用规则引擎 |
description | string Default: "" 规则的描述 |
metadata | object 规则的元数据,不要手动修改 |
{- "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": { }
}
{- "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": { }
}
查看保留消息列表
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) |
{- "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
}
}
获取集群的订阅列表。
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 |
durable | boolean Filter subscriptions by durability |
[- {
- "node": "emqx@127.0.0.1",
- "topic": "testtopic/1",
- "clientid": "emqx_clientid_xx128cdhfc",
- "qos": 0,
- "nl": 0,
- "rap": 0,
- "rh": 0,
- "durable": false
}
]
列出目前所有被封禁的客户端 ID、用户名和 IP 地址。
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) |
clientid | string Query the banned objects with an exact client ID. |
username | string Query the banned objects with an exact username. |
peerhost | string Example: peerhost=127.0.0.1 Query the banned objects with an exact IP address. |
like_clientid | string Fuzzy query banned objects with a regular expression for client ID. |
like_username | string Fuzzy query banned objects with a regular expression for username. |
like_peerhost | string Example: like_peerhost=127.0.0.1 Fuzzy query banned objects with a regular expression for IP address. |
like_peerhost_net | string Example: like_peerhost_net=192.1.0.0/16 Fuzzy query banned objects with a regular expression for CIDR. |
{- "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
}
}
添加一个客户端 ID、用户名或者 IP 地址到黑名单。
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) or string Default: "infinity" 封禁的结束时间,格式为 rfc3339,默认值为发起操作的时间 + 1 年。 |
{- "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"
}
{- "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 地址从黑名单中删除。
as required | string Enum: "clientid" "username" "peerhost" "clientid_re" "username_re" "peerhost_net" Example: username 封禁方式,可以通过客户端 ID、用户名或者 IP 地址等方式进行封禁。 |
who required | string Example: Badass 封禁对象,具体的客户端 ID、用户名或者 IP 地址。 |
{- "code": "NOT_FOUND",
- "message": "string"
}