mqtt schema deep 3

2026-10-17 04:40:42 +08:00 · 2025-02-06 02:45:18 +00:00
parent 11d716eb9b
commit 9ba80017b3
3 changed files with 192 additions and 67 deletions
@@ -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;
@@ -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,
@@ -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, "