diff --git a/plugins/mqtt/schema.c b/plugins/mqtt/schema.c index ced0c7a0..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,77 +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, "${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] == '}') { - 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(); @@ -262,11 +283,29 @@ int mqtt_schema_encode(char *driver, char *group, neu_json_read_resp_t *tags, 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 e0195412..ebeece9a 100644 --- a/plugins/mqtt/schema.h +++ b/plugins/mqtt/schema.h @@ -42,13 +42,19 @@ typedef enum { 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 5c7ee01e..5d94dda0 100644 --- a/tests/ut/mqtt_schema_test.cc +++ b/tests/ut/mqtt_schema_test.cc @@ -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, "