diff --git a/plugins/mqtt/mqtt.json b/plugins/mqtt/mqtt.json index f4e2330c..ae7bb187 100644 --- a/plugins/mqtt/mqtt.json +++ b/plugins/mqtt/mqtt.json @@ -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, diff --git a/plugins/mqtt/schema.c b/plugins/mqtt/schema.c index ac4ff49a..ced0c7a0 100644 --- a/plugins/mqtt/schema.c +++ b/plugins/mqtt/schema.c @@ -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; diff --git a/plugins/mqtt/schema.h b/plugins/mqtt/schema.h index 5fab6edb..e0195412 100644 --- a/plugins/mqtt/schema.h +++ b/plugins/mqtt/schema.h @@ -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 { diff --git a/tests/ut/mqtt_schema_test.cc b/tests/ut/mqtt_schema_test.cc index a04f0b5f..5c7ee01e 100644 --- a/tests/ut/mqtt_schema_test.cc +++ b/tests/ut/mqtt_schema_test.cc @@ -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);