GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-17 04:40:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

mqtt schema support values format

Browse Source
This commit is contained in:
fengzero
2025-02-05 08:38:39 +00:00
parent 1344584aa8
commit 11d716eb9b
4 changed files with 96 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/mqtt/mqtt.json
+2 -2
View File
@@ -56,8 +56,8 @@
"format": {
"name": "Upload Format",
"name_zh": "上报数据格式",
"description": "JSON format of the data reported. In Values-format mode, data are split into `values` and `errors` sub objects. In Tags-format mode, tag data are put in a single array. ECP-format is the format for connecting to ECP data storage. Custom format supports user-defined data format. For variable definition specifications, please refer to [https://docs.emqx.com/en/neuronex/latest/configuration/north-apps/mqtt/api.html].",
"description_zh": "上报数据的 JSON 格式。在 Values-format 格式下，数据被分为 `values` 和 `errors` 两个子对象。在 Tags-format 格式下，数据被放在一个数组中。 ECP-format 为对接 ECP 数据存储的格式。自定义格式，支持用户自定义上报数据格式，变量定义规范请参考【https://docs.emqx.com/zh/neuronex/latest/configuration/north-apps/mqtt/api.html】。",
"description": "JSON format of the data reported. In Values-format mode, data are split into `values` and `errors` sub objects. In Tags-format mode, tag data are put in a single array. ECP-format is the format for connecting to ECP data storage. Custom format supports user-defined data format. For variable definition specifications, please refer to 'https://docs.emqx.com/en/neuronex/latest/configuration/north-apps/mqtt/api.html'.",
"description_zh": "上报数据的 JSON 格式。在 Values-format 格式下，数据被分为 `values` 和 `errors` 两个子对象。在 Tags-format 格式下，数据被放在一个数组中。 ECP-format 为对接 ECP 数据存储的格式。自定义格式，支持用户自定义上报数据格式，变量定义规范请参考'https://docs.emqx.com/zh/neuronex/latest/configuration/north-apps/mqtt/api.html'。",
"attribute": "required",
"type": "map",
"default": 0,
plugins/mqtt/schema.c
+82 -5
View File
@@ -48,10 +48,16 @@ int mqtt_schema_validate(const char *schema, mqtt_schema_vt_t **vts,
vt->vt = MQTT_SCHEMA_NODE_NAME;
} else if (strcmp(str_val, "${group}") == 0) {
vt->vt = MQTT_SCHEMA_GROUP_NAME;
} else if (strcmp(str_val, "${tag_values}") == 0) {
vt->vt = MQTT_SCHEMA_TAG_VALUES;
} else if (strcmp(str_val, "${tags}") == 0) {
vt->vt = MQTT_SCHEMA_TAGS;
} else if (strcmp(str_val, "${static_tags}") == 0) {
vt->vt = MQTT_SCHEMA_STATIC_TAG_VALUES;
vt->vt = MQTT_SCHEMA_STATIC_TAGS;
} else if (strcmp(str_val, "${tag_values}") == 0) {
vt->vt = MQTT_SCHEMA_TAGVALUES;
} else if (strcmp(str_val, "${static_tag_values}") == 0) {
vt->vt = MQTT_SCHEMA_STATIC_TAGVALUES;
} else if (strcmp(str_val, "${tag_error_values}") == 0) {
vt->vt = MQTT_SCHEMA_TAG_ERROR_VALUES;
} else if (strcmp(str_val, "${tag_errors}") == 0) {
vt->vt = MQTT_SCHEMA_TAG_ERRORS;
} else {
@@ -95,7 +101,7 @@ int mqtt_schema_encode(char *driver, char *group, neu_json_read_resp_t *tags,
elem.t = NEU_JSON_STR;
elem.v.val_str = group;
break;
case MQTT_SCHEMA_TAG_VALUES: {
case MQTT_SCHEMA_TAGS: {
void *values_array = neu_json_array();
neu_json_read_resp_tag_t *p_tag = tags->tags;
@@ -130,7 +136,37 @@ int mqtt_schema_encode(char *driver, char *group, neu_json_read_resp_t *tags,
elem.v.val_object = values_array;
break;
}
case MQTT_SCHEMA_STATIC_TAG_VALUES: {
case MQTT_SCHEMA_TAGVALUES: {
void *values_array = neu_json_encode_new();
neu_json_read_resp_tag_t *p_tag = tags->tags;
for (int j = 0; j < tags->n_tag; j++) {
neu_json_elem_t tag_elems[1 + NEU_TAG_META_SIZE] = { 0 };
if (p_tag->error == 0) {
tag_elems[0].name = p_tag->name;
tag_elems[0].t = p_tag->t;
tag_elems[0].v = p_tag->value;
for (int k = 0; k < p_tag->n_meta; k++) {
tag_elems[1 + k].name = p_tag->metas[k].name;
tag_elems[1 + k].t = p_tag->metas[k].t;
tag_elems[1 + k].v = p_tag->metas[k].value;
}
neu_json_encode_field(values_array, tag_elems,
1 + p_tag->n_meta);
}
p_tag++;
}
elem.t = NEU_JSON_OBJECT;
elem.v.val_object = values_array;
break;
}
case MQTT_SCHEMA_STATIC_TAGS: {
void *static_array = neu_json_array();
for (size_t k = 0; k < n_s_tags; k++) {
@@ -153,6 +189,24 @@ int mqtt_schema_encode(char *driver, char *group, neu_json_read_resp_t *tags,
break;
}
case MQTT_SCHEMA_STATIC_TAGVALUES: {
void *static_array = neu_json_encode_new();
for (size_t k = 0; k < n_s_tags; k++) {
neu_json_elem_t tag_elems[1 + NEU_TAG_META_SIZE] = { 0 };
tag_elems[0].name = s_tags[k].name;
tag_elems[0].t = s_tags[k].jtype;
tag_elems[0].v = s_tags[k].jvalue;
neu_json_encode_field(static_array, tag_elems, 1);
}
elem.t = NEU_JSON_OBJECT;
elem.v.val_object = static_array;
break;
}
case MQTT_SCHEMA_TAG_ERRORS: {
void *errors_array = neu_json_array();
@@ -181,6 +235,29 @@ int mqtt_schema_encode(char *driver, char *group, neu_json_read_resp_t *tags,
elem.v.val_object = errors_array;
break;
}
case MQTT_SCHEMA_TAG_ERROR_VALUES: {
void *errors_array = neu_json_encode_new();
neu_json_read_resp_tag_t *p_tag = tags->tags;
for (int k = 0; k < tags->n_tag; k++) {
neu_json_elem_t tag_elems[1 + NEU_TAG_META_SIZE] = { 0 };
if (p_tag->error != 0) {
tag_elems[0].name = p_tag->name;
tag_elems[0].t = NEU_JSON_INT;
tag_elems[0].v.val_int = p_tag->error;
neu_json_encode_field(errors_array, tag_elems, 1);
}
p_tag++;
}
elem.t = NEU_JSON_OBJECT;
elem.v.val_object = errors_array;
break;
}
case MQTT_SCHEMA_UD:
elem.t = NEU_JSON_STR;
elem.v.val_str = vts[i].ud;
plugins/mqtt/schema.h
+10 -7
View File
@@ -32,13 +32,16 @@ extern "C" {
#include "plugin.h"
typedef enum {
MQTT_SCHEMA_TIMESTAMP = 0,
MQTT_SCHEMA_NODE_NAME = 1,
MQTT_SCHEMA_GROUP_NAME = 2,
MQTT_SCHEMA_TAG_VALUES = 3,
MQTT_SCHEMA_STATIC_TAG_VALUES = 4,
MQTT_SCHEMA_TAG_ERRORS = 5,
MQTT_SCHEMA_UD = 6,
MQTT_SCHEMA_TIMESTAMP = 0,
MQTT_SCHEMA_NODE_NAME = 1,
MQTT_SCHEMA_GROUP_NAME = 2,
MQTT_SCHEMA_TAGS = 3,
MQTT_SCHEMA_STATIC_TAGS = 4,
MQTT_SCHEMA_TAGVALUES = 5,
MQTT_SCHEMA_STATIC_TAGVALUES = 6,
MQTT_SCHEMA_TAG_ERRORS = 7,
MQTT_SCHEMA_TAG_ERROR_VALUES = 8,
MQTT_SCHEMA_UD = 9,
} mqtt_schema_vt_e;
typedef struct {
tests/ut/mqtt_schema_test.cc
+2 -2
View File
@@ -29,10 +29,10 @@ TEST(validate_success, schema)
EXPECT_EQ(vts[2].vt, MQTT_SCHEMA_GROUP_NAME);
EXPECT_STREQ(vts[3].name, "tags");
EXPECT_EQ(vts[3].vt, MQTT_SCHEMA_TAG_VALUES);
EXPECT_EQ(vts[3].vt, MQTT_SCHEMA_TAGVALUES);
EXPECT_STREQ(vts[4].name, "static");
EXPECT_EQ(vts[4].vt, MQTT_SCHEMA_STATIC_TAG_VALUES);
EXPECT_EQ(vts[4].vt, MQTT_SCHEMA_STATIC_TAGS);
EXPECT_STREQ(vts[5].name, "errors");
EXPECT_EQ(vts[5].vt, MQTT_SCHEMA_TAG_ERRORS);