diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 02176e74..5237a4a4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -52,9 +52,9 @@ jobs: run: | ./package-sdk.sh -p ${{ matrix.arch[2] }} -n neuron-sdk-$VERSION - - uses: actions/upload-artifact@v3 + - uses: actions/upload-artifact@v4 with: - name: neuron-sdk + name: neuron-sdk-${{ matrix.arch[2] }} path: | neuron-sdk-*.tar.gz @@ -83,9 +83,9 @@ jobs: run: | ./package-sdk.sh -p ${{ matrix.arch[2] }} -n neuron-sdk-debug-$VERSION - - uses: actions/upload-artifact@v3 + - uses: actions/upload-artifact@v4 with: - name: neuron-sdk + name: neuron-sdk-debug path: | neuron-sdk-debug-*.tar.gz @@ -102,9 +102,19 @@ jobs: env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} - - uses: actions/download-artifact@v3 + - uses: actions/download-artifact@v4 with: - name: neuron-sdk + name: neuron-sdk-debug + + - uses: actions/download-artifact@v4 + with: + name: neuron-sdk-aarch64 + - uses: actions/download-artifact@v4 + with: + name: neuron-sdk-armv4 + - uses: actions/download-artifact@v4 + with: + name: neuron-sdk-x86_64 - uses: softprops/action-gh-release@v1 env: diff --git a/plugins/mqtt/mqtt.json b/plugins/mqtt/mqtt.json index f4e2330c..269da2e5 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, @@ -129,30 +129,6 @@ "length": 255 } }, - "driver-action-req-topic": { - "name": "Driver Action Request Topic", - "name_zh": "驱动执行动作请求主题", - "description": "MQTT topic to receive driver action request messages.", - "description_zh": "驱动执行动作请求的 MQTT 主题。", - "attribute": "required", - "type": "string", - "default": "/neuron/${random_str}/action/req", - "valid": { - "length": 255 - } - }, - "driver-action-resp-topic": { - "name": "Driver Action Response Topic", - "name_zh": "驱动执行动作响应主题", - "description": "MQTT topic to send driver action response messages.", - "description_zh": "驱动执行动作响应的 MQTT 主题。", - "attribute": "required", - "type": "string", - "default": "/neuron/${random_str}/action/resp", - "valid": { - "length": 255 - } - }, "upload_drv_state": { "name": "Driver Status Report", "name_zh": "驱动状态上报", diff --git a/plugins/mqtt/schema.c b/plugins/mqtt/schema.c index ac4ff49a..e67f0189 100644 --- a/plugins/mqtt/schema.c +++ b/plugins/mqtt/schema.c @@ -2,6 +2,80 @@ #include "schema.h" +static int mqtt_schema_parse(json_t *root, mqtt_schema_vt_t **vts, + size_t *vts_len, int deep) +{ + *vts_len = 0; + *vts = NULL; + + const char *key = NULL; + json_t * value = NULL; + + json_object_foreach(root, key, value) + { + if (json_is_string(value)) { + const char *str_val = json_string_value(value); + if (str_val == NULL) { + return -1; + } + + *vts_len += 1; + *vts = realloc(*vts, *vts_len * sizeof(mqtt_schema_vt_t)); + mqtt_schema_vt_t *vt = &(*vts)[*vts_len - 1]; + + memset(vt, 0, sizeof(mqtt_schema_vt_t)); + strcpy(vt->name, key); + + if (strcmp(str_val, "${timestamp}") == 0) { + vt->vt = MQTT_SCHEMA_TIMESTAMP; + } else if (strcmp(str_val, "${node}") == 0) { + vt->vt = MQTT_SCHEMA_NODE_NAME; + } else if (strcmp(str_val, "${group}") == 0) { + vt->vt = MQTT_SCHEMA_GROUP_NAME; + } 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_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 { + if (str_val[0] == '$' && str_val[1] == '{' && + str_val[strlen(str_val) - 1] == '}') { + return -1; + } else { + vt->vt = MQTT_SCHEMA_UD; + strcpy(vt->ud, str_val); + } + } + } else if (json_is_object(value)) { + if (deep >= 3) { + return -1; + } + + *vts_len += 1; + *vts = realloc(*vts, *vts_len * sizeof(mqtt_schema_vt_t)); + mqtt_schema_vt_t *vt = &(*vts)[*vts_len - 1]; + + memset(vt, 0, sizeof(mqtt_schema_vt_t)); + strcpy(vt->name, key); + + vt->vt = MQTT_SCHEMA_OBJECT; + return mqtt_schema_parse(value, &vt->sub_vts, &vt->n_sub_vts, + deep + 1); + } else { + return -1; + } + } + + return 0; +} + int mqtt_schema_validate(const char *schema, mqtt_schema_vt_t **vts, size_t *vts_len) { @@ -10,71 +84,24 @@ int mqtt_schema_validate(const char *schema, mqtt_schema_vt_t **vts, return -1; } - *vts_len = 0; - *vts = NULL; - - const char *key = NULL; - json_t * value = NULL; - - json_object_foreach(root, key, value) - { - if (!json_is_string(value)) { - json_decref(root); - if (*vts_len > 0) { - free(*vts); - } - return -1; - } - - const char *str_val = json_string_value(value); - if (str_val == NULL) { - json_decref(root); - if (*vts_len > 0) { - free(*vts); - } - return -1; - } - - *vts_len += 1; - *vts = realloc(*vts, *vts_len * sizeof(mqtt_schema_vt_t)); - mqtt_schema_vt_t *vt = &(*vts)[*vts_len - 1]; - - memset(vt, 0, sizeof(mqtt_schema_vt_t)); - strcpy(vt->name, key); - - if (strcmp(str_val, "${timestamp}") == 0) { - vt->vt = MQTT_SCHEMA_TIMESTAMP; - } else if (strcmp(str_val, "${node}") == 0) { - 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, "${static_tags}") == 0) { - vt->vt = MQTT_SCHEMA_STATIC_TAG_VALUES; - } else if (strcmp(str_val, "${tag_errors}") == 0) { - vt->vt = MQTT_SCHEMA_TAG_ERRORS; - } else { - if (str_val[0] == '$' && str_val[1] == '{' && - str_val[strlen(str_val) - 1] == '}') { - json_decref(root); - free(*vts); - return -1; - } else { - vt->vt = MQTT_SCHEMA_UD; - strcpy(vt->ud, str_val); - } + int ret = mqtt_schema_parse(root, vts, vts_len, 1); + if (ret != 0) { + if (*vts != NULL) { + free(*vts); + *vts = NULL; } + json_decref(root); + return -1; } json_decref(root); return 0; } -int mqtt_schema_encode(char *driver, char *group, neu_json_read_resp_t *tags, - mqtt_schema_vt_t *vts, size_t n_vts, - mqtt_static_vt_t *s_tags, size_t n_s_tags, - char **result_str) +static void *schema_encode(char *driver, char *group, + neu_json_read_resp_t *tags, mqtt_schema_vt_t *vts, + size_t n_vts, mqtt_static_vt_t *s_tags, + size_t n_s_tags) { void *root = neu_json_encode_new(); @@ -95,7 +122,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 +157,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 +210,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,15 +256,56 @@ 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; break; + case MQTT_SCHEMA_OBJECT: { + void *sub_root = schema_encode(driver, group, tags, vts[i].sub_vts, + vts[i].n_sub_vts, s_tags, n_s_tags); + elem.t = NEU_JSON_OBJECT; + elem.v.val_object = sub_root; + break; + } } neu_json_encode_field(root, &elem, 1); } + return root; +} + +int mqtt_schema_encode(char *driver, char *group, neu_json_read_resp_t *tags, + mqtt_schema_vt_t *vts, size_t n_vts, + mqtt_static_vt_t *s_tags, size_t n_s_tags, + char **result_str) +{ + void *root = + schema_encode(driver, group, tags, vts, n_vts, s_tags, n_s_tags); + int ret = neu_json_encode(root, result_str); neu_json_decode_free(root); return ret; diff --git a/plugins/mqtt/schema.h b/plugins/mqtt/schema.h index 5fab6edb..ebeece9a 100644 --- a/plugins/mqtt/schema.h +++ b/plugins/mqtt/schema.h @@ -32,20 +32,29 @@ 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_OBJECT = 10, } mqtt_schema_vt_e; -typedef struct { +typedef struct mqtt_schema_vt mqtt_schema_vt_t; +typedef struct mqtt_schema_vt { char name[128]; mqtt_schema_vt_e vt; - char ud[128]; + + char ud[128]; + + mqtt_schema_vt_t *sub_vts; + size_t n_sub_vts; } mqtt_schema_vt_t; int mqtt_schema_validate(const char *schema, mqtt_schema_vt_t **vts, diff --git a/tests/ut/mqtt_schema_test.cc b/tests/ut/mqtt_schema_test.cc index a04f0b5f..5d94dda0 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); @@ -44,6 +44,51 @@ TEST(validate_success, schema) free(vts); } +TEST(validate_deep_success, schema) +{ + const char *schema = "{\"timestamp\": \"${timestamp}\", \"object\": " + "{\"group\":\"${group}\", \"sub_object\": " + "{\"tags\":\"${tags}\", \"cus\":\"aabb\"}}}"; + mqtt_schema_vt_t *vts; + size_t n_vts; + int ret = mqtt_schema_validate(schema, &vts, &n_vts); + EXPECT_EQ(ret, 0); + EXPECT_EQ(n_vts, 2); + + EXPECT_STREQ(vts[0].name, "timestamp"); + EXPECT_EQ(vts[0].vt, MQTT_SCHEMA_TIMESTAMP); + + EXPECT_STREQ(vts[1].name, "object"); + EXPECT_EQ(vts[1].vt, MQTT_SCHEMA_OBJECT); + EXPECT_EQ(vts[1].n_sub_vts, 2); + + EXPECT_STREQ(vts[1].sub_vts[0].name, "group"); + EXPECT_EQ(vts[1].sub_vts[0].vt, MQTT_SCHEMA_GROUP_NAME); + + EXPECT_STREQ(vts[1].sub_vts[1].name, "sub_object"); + EXPECT_EQ(vts[1].sub_vts[1].vt, MQTT_SCHEMA_OBJECT); + EXPECT_EQ(vts[1].sub_vts[1].n_sub_vts, 2); + + EXPECT_STREQ(vts[1].sub_vts[1].sub_vts[0].name, "tags"); + EXPECT_EQ(vts[1].sub_vts[1].sub_vts[0].vt, MQTT_SCHEMA_TAGS); + + EXPECT_STREQ(vts[1].sub_vts[1].sub_vts[1].name, "cus"); + EXPECT_EQ(vts[1].sub_vts[1].sub_vts[1].vt, MQTT_SCHEMA_UD); + + for (size_t i = 0; i < n_vts; i++) { + if (vts[i].vt == MQTT_SCHEMA_OBJECT) { + for (size_t k = 0; k < vts[i].n_sub_vts; k++) { + if (vts[i].sub_vts[k].vt == MQTT_SCHEMA_OBJECT) { + free(vts[i].sub_vts[k].sub_vts); + } + } + free(vts[i].sub_vts); + } + } + + free(vts); +} + TEST(validate_failed, schema) { const char *schema = @@ -100,6 +145,41 @@ TEST(validate_success, schema_static) mqtt_static_free(vts, n_vts); } +TEST(schema_encode_deep, schema_encode) +{ + const char *schema = "{\"timestamp\": \"${timestamp}\", \"object\": " + "{\"group\":\"${group}\", \"sub_object\": " + "{\"node\":\"${node}\", \"cus\":\"aabb\"}}}"; + mqtt_schema_vt_t *vts; + size_t n_vts; + int ret = mqtt_schema_validate(schema, &vts, &n_vts); + EXPECT_EQ(ret, 0); + EXPECT_EQ(n_vts, 2); + + char * result = NULL; + neu_json_read_resp_t tags = { 0 }; + ret = mqtt_schema_encode((char *) "driver", (char *) "group", &tags, vts, + n_vts, NULL, 0, &result); + EXPECT_EQ(ret, 0); + EXPECT_STREQ(result, + "{\"timestamp\": 0, \"object\": {\"group\": \"group\", " + "\"sub_object\": {\"node\": \"driver\", \"cus\": \"aabb\"}}}"); + free(result); + + for (size_t i = 0; i < n_vts; i++) { + if (vts[i].vt == MQTT_SCHEMA_OBJECT) { + for (size_t k = 0; k < vts[i].n_sub_vts; k++) { + if (vts[i].sub_vts[k].vt == MQTT_SCHEMA_OBJECT) { + free(vts[i].sub_vts[k].sub_vts); + } + } + free(vts[i].sub_vts); + } + } + + free(vts); +} + TEST(schema_encode, schema_static) { const char *schema = @@ -134,7 +214,7 @@ TEST(schema_encode, schema_static) EXPECT_STREQ( result, "{\"timestamp\": 0, \"node\": \"driver\", \"group\": \"group\", " - "\"tags\": [], \"static\": [{\"name\": \"static_tag1\", \"value\": " + "\"tags\": {}, \"static\": [{\"name\": \"static_tag1\", \"value\": " "\"string\"}, {\"name\": \"static_tag2\", \"value\": 1234}, {\"name\": " "\"static_tag3\", \"value\": 11.44}, {\"name\": \"static_tag4\", " "\"value\": true}, {\"name\": \"static_tag5\", \"value\": \"[1, 3, "