Make default persist adapter load group configs

Author: eeff

src/adapter/adapter.c

@@ -260,6 +260,53 @@ static int persister_singleton_load_setting(neu_adapter_t *adapter,
     return rv;
 }
 
+static int persister_singleton_load_grp_and_tags(neu_adapter_t *adapter,
+                                                 const char *   adapter_name,
+                                                 neu_node_id_t  node_id)
+{
+    vector_t *       group_config_infos = NULL;
+    neu_persister_t *persister          = persister_singleton_get();
+
+    int rv = neu_persister_load_group_configs(persister, adapter_name,
+                                              &group_config_infos);
+    if (0 != rv) {
+        log_error("%s fail load grp config of %s", adapter->name, adapter_name);
+        return rv;
+    }
+
+    VECTOR_FOR_EACH(group_config_infos, iter)
+    {
+        neu_persist_group_config_info_t *p = iterator_get(&iter);
+
+        neu_taggrp_config_t *grp_config =
+            neu_taggrp_cfg_new(p->group_config_name);
+        if (NULL == grp_config) {
+            rv = NEU_ERR_ENOMEM;
+            break;
+        }
+        neu_taggrp_cfg_set_interval(grp_config, p->read_interval);
+
+        neu_cmd_add_grp_config_t cmd = {
+            .node_id    = node_id,
+            .grp_config = grp_config,
+        };
+
+        rv = neu_manager_add_grp_config(adapter->manager, &cmd);
+        const char *ok_or_err = (0 == rv) ? "success" : "fail";
+        log_info("%s load group config %s interval:%d", adapter->name,
+                 p->group_config_name, p->read_interval);
+        if (0 != rv) {
+            neu_taggrp_cfg_free(grp_config);
+            break;
+        }
+
+        // TODO: add tags
+    }
+
+    neu_persist_group_config_infos_free(group_config_infos);
+    return rv;
+}
+
 static int persister_singleton_load_data(neu_adapter_t *adapter)
 {
     vector_t *       adapter_infos = NULL;
@@ -303,6 +350,12 @@ static int persister_singleton_load_data(neu_adapter_t *adapter)
             goto error_add_adapters;
         }
 
+        rv = persister_singleton_load_grp_and_tags(adapter, adapter_info->name,
+                                                   node_id);
+        if (0 != rv) {
+            goto error_add_adapters;
+        }
+
         // TODO: start node according to state
     }
 

src/persist/persist.c

@@ -410,6 +410,43 @@ static int read_file_string(const char *fn, char **out)
     return NEU_ERR_FAILURE;
 }
 
+// file tree walking callback for collecting adapter group config infos
+static int read_group_config_cb(const char *fpath, bool is_dir, void *arg)
+{
+    vector_t *group_config_infos = arg;
+    int       rv                 = 0;
+
+    if (is_dir) {
+        return 0;
+    }
+
+    char *json_str = NULL;
+    rv             = read_file_string(fpath, &json_str);
+    if (0 != rv) {
+        return rv;
+    }
+
+    neu_json_group_configs_req_t *group_config_req = NULL;
+    rv = neu_json_decode_group_configs_req(json_str, &group_config_req);
+    free(json_str);
+    if (0 != rv) {
+        return rv;
+    }
+
+    log_info("read %s", fpath);
+
+    if (0 == vector_push_back(group_config_infos, group_config_req)) {
+        // NOTE: do not call neu_json_decode_group_configs_req_free,
+        //       since member ownership was transferred to vector
+        free(group_config_req);
+    } else {
+        neu_json_decode_group_configs_req_free(group_config_req);
+        rv = NEU_ERR_ENOMEM;
+    }
+
+    return rv;
+}
+
 typedef struct neu_persister {
     const char *persist_dir;
     const char *adapters_fname;
@@ -841,48 +878,30 @@ int neu_persister_store_group_config(
 
 int neu_persister_load_group_configs(neu_persister_t *persister,
                                      const char *     adapter_name,
-                                     const char *     group_config_name,
                                      vector_t **      group_config_infos)
 {
-    char *group_configs = NULL;
+    char path[PATH_MAX_SIZE] = { 0 };
 
-    char group_config_file[128] = { 0 };
-    int  rv = snprintf(group_config_file, 128, "%s/persist/%s/group_configs/%s",
-                      persister->persist_dir, adapter_name, group_config_name);
-    if (sizeof(group_config_file) == rv) {
-        log_error("group_config_file exceeds maximum value");
+    int n = persister_group_configs_dir(path, sizeof(path), persister,
+                                        adapter_name);
+    if (sizeof(path) == n) {
+        log_error("persister path too long: %s", path);
         return -1;
     }
 
-    rv = read_file_string(group_config_file, &group_configs);
-    if (rv != 0) {
-        return rv;
+    vector_t *result = vector_new(0, sizeof(neu_persist_group_config_info_t));
+    if (NULL == result) {
+        return NEU_ERR_ENOMEM;
     }
 
-    neu_json_group_configs_req_t *group_config_req = NULL;
-    rv = neu_json_decode_group_configs_req(group_configs, &group_config_req);
-    if (rv != 0) {
-        return rv;
+    int rv = file_tree_walk(path, read_group_config_cb, result);
+    if (0 == rv) {
+        *group_config_infos = result;
+    } else {
+        neu_persist_group_config_infos_free(result);
     }
 
-    // vector_t *vec = vector_new_move_from_buf(
-    //     group_config_req->datatag_names, group_config_req->datatag_names,
-    //     group_config_req->group_config_name,
-    //     group_config_req->n_datatag_name, group_config_req->read_interval,
-    //     sizeof(neu_persist_group_config_info_t));
-    // if (vec == NULL) {
-    //     return -1;
-    // }
-
-    // *group_config_infos                 = vec;
-    // group_config_req->adapter_name      = NULL;
-    // group_config_req->datatag_names     = NULL;
-    // group_config_req->group_config_name = NULL;
-    // group_config_req->n_datatag_name    = 0;
-    // group_config_req->read_interval     = 0;
-
-    neu_json_decode_group_configs_req_free(group_config_req);
-    return 0;
+    return rv;
 }
 
 int neu_persister_delete_group_config(neu_persister_t *persister,

src/persist/persist.h

@@ -61,6 +61,23 @@ static inline void neu_persist_plugin_infos_free(vector_t *plugin_infos)
     vector_free(plugin_infos);
 }
 
+static inline void
+neu_persist_group_config_infos_free(vector_t *group_config_infos)
+{
+    VECTOR_FOR_EACH(group_config_infos, iter)
+    {
+        neu_persist_group_config_info_t *p         = iterator_get(&iter);
+        char **                          tag_names = p->datatag_names;
+        for (int i = 0; i < p->n_datatag_name; ++i) {
+            free(tag_names[i]);
+        }
+        free(tag_names);
+        free(p->group_config_name);
+        free(p->adapter_name);
+    }
+    vector_free(group_config_infos);
+}
+
 /**
  * Persister, provide methods to persist data */
 typedef struct neu_persister neu_persister_t;
@@ -188,7 +205,6 @@ int neu_persister_store_group_config(
  */
 int neu_persister_load_group_configs(neu_persister_t *persister,
                                      const char *     adapter_name,
-                                     const char *     group_config_name,
                                      vector_t **      group_config_infos);
 /**
  * Delete group config.