diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index f270e72a..c0352044 100644 --- a/src/adapter/adapter.c +++ b/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 } diff --git a/src/persist/persist.c b/src/persist/persist.c index 187990cf..4afa4e8a 100644 --- a/src/persist/persist.c +++ b/src/persist/persist.c @@ -410,6 +410,43 @@ error_open: 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, diff --git a/src/persist/persist.h b/src/persist/persist.h index c5e6f2b0..c52c1cb2 100644 --- a/src/persist/persist.h +++ b/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.