diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 5e594f26..f270e72a 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -374,6 +374,48 @@ static int persister_singleton_handle_plugins(neu_adapter_t *adapter, return rv; } +static int persister_singleton_handle_grp_config(neu_adapter_t *adapter, + msg_type_e event, + neu_node_id_t node_id, + void * arg) +{ + neu_persister_t *persister = persister_singleton_get(); + char * adapter_name = NULL; + + log_info("%s handling grp config event %d", adapter->name, event); + + int rv = neu_manager_get_node_name_by_id(adapter->manager, node_id, + &adapter_name); + if (0 != rv) { + return rv; + } + + if (MSG_EVENT_DEL_GRP_CONFIG != event) { + neu_taggrp_config_t * grp_config = arg; + neu_persist_group_config_info_t info = { + .group_config_name = (char *) neu_taggrp_cfg_get_name(grp_config), + .read_interval = neu_taggrp_cfg_get_interval(grp_config), + .adapter_name = adapter_name, // TODO: remove + .datatag_names = NULL, // we don't need these fields + .n_datatag_name = 0, // + }; + rv = neu_persister_store_group_config(persister, adapter_name, &info); + if (0 != rv) { + log_error("% failed to store group config infos", adapter->name); + } + } else { + const char *group_config_name = arg; + rv = neu_persister_delete_group_config(persister, adapter_name, + group_config_name); + if (0 != rv) { + log_error("% failed to del group config infos", adapter->name); + } + } + + free(adapter_name); + return rv; +} + neu_plugin_running_state_e neu_adapter_state_to_plugin_state(neu_adapter_t *adapter) { @@ -549,6 +591,30 @@ static void adapter_loop(void *arg) break; } + case MSG_EVENT_ADD_GRP_CONFIG: { + neu_cmd_add_grp_config_t *cmd = msg_get_buf_ptr(pay_msg); + persister_singleton_handle_grp_config( + adapter, pay_msg_type, cmd->node_id, cmd->grp_config); + neu_taggrp_cfg_free(cmd->grp_config); + break; + } + + case MSG_EVENT_UPDATE_GRP_CONFIG: { + neu_cmd_update_grp_config_t *cmd = msg_get_buf_ptr(pay_msg); + persister_singleton_handle_grp_config( + adapter, pay_msg_type, cmd->node_id, cmd->grp_config); + neu_taggrp_cfg_free(cmd->grp_config); + break; + } + + case MSG_EVENT_DEL_GRP_CONFIG: { + neu_cmd_del_grp_config_t *cmd = msg_get_buf_ptr(pay_msg); + persister_singleton_handle_grp_config( + adapter, pay_msg_type, cmd->node_id, cmd->config_name); + free(cmd->config_name); + break; + } + case MSG_EVENT_ADD_PLUGIN: // fall through diff --git a/src/core/neu_manager.c b/src/core/neu_manager.c index 79c4ef62..43d6400d 100644 --- a/src/core/neu_manager.c +++ b/src/core/neu_manager.c @@ -1327,6 +1327,14 @@ static void manager_loop(void *arg) // fall through case MSG_EVENT_SET_NODE_SETTING: + + case MSG_EVENT_ADD_GRP_CONFIG: + // fall through + + case MSG_EVENT_UPDATE_GRP_CONFIG: + // fall through + + case MSG_EVENT_DEL_GRP_CONFIG: // fall through case MSG_EVENT_ADD_PLUGIN: diff --git a/src/persist/persist.c b/src/persist/persist.c index 5a649832..187990cf 100644 --- a/src/persist/persist.c +++ b/src/persist/persist.c @@ -17,10 +17,10 @@ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ -#include #include #include +#include #include #include #include @@ -48,6 +48,13 @@ #define PATH_MAX_SIZE 128 +static inline bool ends_with(const char *str, const char *suffix) +{ + size_t m = strlen(str); + size_t n = strlen(suffix); + return m >= n && !strncmp(str + m - n, suffix, n); +} + /** * Escape special characters in path string, * '%' -> '%%', '/' => '%-', '.' -> '%.' @@ -151,13 +158,7 @@ static int path_cat_escaped(char *dst, size_t len, size_t size, const char *src) return i + n; } -typedef struct neu_persister { - const char *persist_dir; - const char *adapters_fname; - const char *plugins_fname; -} neu_persister_t; - -static int create_dir(char *dir_name) +static inline int create_dir(char *dir_name) { int rv = mkdir(dir_name, 0777); if (0 != rv && EEXIST == errno) { @@ -166,6 +167,149 @@ static int create_dir(char *dir_name) return rv; } +/** + * Create a directory and any intermediate directory if necessary. + */ +static int create_dir_recursive(const char *path) +{ + int rv = 0; + char p[PATH_MAX_SIZE] = { 0 }; + + rv = snprintf(p, sizeof(p), "%s", path); + if (sizeof(p) == rv) { + return -1; + } + + int i = 0; + while (1) { + while (p[i] && PATH_SEP_CHAR == p[i]) { + ++i; + } + if ('\0' == p[i]) { + break; + } + while (p[i] && PATH_SEP_CHAR != p[i]) { + ++i; + } + + char c = p[i]; + p[i] = '\0'; + rv = create_dir(p); + if (0 != rv) { + break; + } + p[i] = c; + } + return rv; +} + +/** + * Type of callback invoked in file tree walking. + */ +typedef int (*file_tree_walk_cb_t)(const char *fpath, bool is_dir, void *arg); + +// depth first file tree walking implementation. +// `path_buf` is buffer for accumulating file path, +// `len` is the length of the directory path of the current invocation, +// `size` is the path buffer size. +static int file_tree_walk_(char *path_buf, size_t len, size_t size, + file_tree_walk_cb_t cb, void *ctx) +{ + int rv = 0; + DIR * dirp; + struct dirent *dent; + + // if not possible to read the directory for this user + if ((dirp = opendir(path_buf)) == NULL) { + return -1; + } + + while (NULL != (dent = readdir(dirp))) { + if (DT_DIR == dent->d_type) { + // skip entries "." and ".." + if (!strcmp(dent->d_name, ".") || !strcmp(dent->d_name, "..")) { + continue; + } + } else if (DT_REG != dent->d_type) { + continue; + } + + // determinate a full path of an entry + size_t n = path_cat(path_buf, len, size, dent->d_name); + if (size == n) { + rv = -1; + break; + } + + if (DT_DIR == dent->d_type) { + rv = file_tree_walk_(path_buf, n, size, cb, ctx); + } else { + rv = cb(path_buf, false, ctx); + } + + path_buf[len] = '\0'; // restore current directory path + + if (0 != rv) { + break; + } + } + + if (0 == rv) { + rv = cb(path_buf, true, ctx); + } + + closedir(dirp); + return rv; +} + +/** + * Depth first traversal of file tree. + * + * @param dir path of directory + * @param cb callback to invoke for each directory entry + * @param ctx argument to callback + * + * @return If `cb` returns nonzero, then the tree walk is terminated and the + * value returned by `cb` is returned as the result + */ +static int file_tree_walk(const char *dir, file_tree_walk_cb_t cb, void *ctx) +{ + char buf[PATH_MAX_SIZE] = { 0 }; + int n = path_cat(buf, 0, sizeof(buf), dir); + if (sizeof(buf) == n) { + return -1; + } + return file_tree_walk_(buf, n, sizeof(buf), cb, ctx); +} + +// file tree walking callback for `rmdir_recursive` +static int remove_cb(const char *fpath, bool is_dir, void *arg) +{ + (void) arg; + + if (!is_dir && !ends_with(fpath, ".json")) { + // we are being defensive here, ignoring non-json files + return 0; + } + log_info("remove %s", fpath); + return remove(fpath); +} + +/** + * Remove directory recursively. + * @param dir path of directory to remove + */ +static int rmdir_recursive(const char *dir) +{ + return file_tree_walk(dir, remove_cb, NULL); +} + +/** + * Ensure that a file with the given name exists. + * + * @param name file name + * @param default_content default file content if file not exists + */ static inline int ensure_file_exist(const char *name, const char *default_content) { @@ -186,75 +330,6 @@ static inline int ensure_file_exist(const char *name, return 0; } -neu_persister_t *neu_persister_create(char *dir_name) -{ - char dir_path[128] = { 0 }; - int rv = snprintf(dir_path, 128, "%s", dir_name); - if (sizeof(dir_path) == rv) { - log_error("dir_path exceeds maximum value"); - goto error_dir_name; - } - - rv = create_dir(dir_path); - if (rv != 0) { - log_error("failed to create directory"); - goto error_dir_name; - } - - char adapter_path[128] = { 0 }; - rv = snprintf(adapter_path, 128, "%s/adapters.json", dir_path); - if (sizeof(adapter_path) == rv) { - log_error("adapter_path exceeds maximun value"); - goto error_dir_name; - } - - char plugin_path[128] = { 0 }; - rv = snprintf(plugin_path, 128, "%s/plugins.json", dir_path); - if (sizeof(plugin_path) == rv) { - log_error("plugin_path exceeds maximun value"); - goto error_dir_name; - } - - neu_persister_t *persister = malloc(sizeof(neu_persister_t)); - if (persister == NULL) { - log_error("failed to alloc memory for persister struct"); - goto error_dir_name; - } - - if (0 != ensure_file_exist(adapter_path, "{\"nodes\": []}")) { - log_error("persister failed to ensure file exist: %s", adapter_path); - goto error_io; - } - if (0 != ensure_file_exist(plugin_path, "{\"plugins\": []}")) { - log_error("persister failed to ensure file exist: %s", plugin_path); - goto error_io; - } - - persister->adapters_fname = strdup(adapter_path); - persister->plugins_fname = strdup(plugin_path); - persister->persist_dir = strdup(dir_name); - - return persister; - -error_io: - free(persister); -error_dir_name: - return NULL; -} - -static inline int persister_adapter_dir(char *buf, size_t size, - neu_persister_t *persister, - const char * adapter_name) -{ - size_t n = path_cat(buf, 0, size, persister->persist_dir); - if (size == n) { - return n; - } - - n = path_cat_escaped(buf, n, size, adapter_name); - return n; -} - static int write_file_string(const char *fn, const char *s) { char tmp[PATH_MAX_SIZE] = { 0 }; @@ -335,6 +410,108 @@ error_open: return NEU_ERR_FAILURE; } +typedef struct neu_persister { + const char *persist_dir; + const char *adapters_fname; + const char *plugins_fname; +} neu_persister_t; + +neu_persister_t *neu_persister_create(char *dir_name) +{ + char dir_path[128] = { 0 }; + int rv = snprintf(dir_path, 128, "%s", dir_name); + if (sizeof(dir_path) == rv) { + log_error("dir_path exceeds maximum value"); + goto error_dir_name; + } + + rv = create_dir(dir_path); + if (rv != 0) { + log_error("failed to create directory"); + goto error_dir_name; + } + + char adapter_path[128] = { 0 }; + rv = snprintf(adapter_path, 128, "%s/adapters.json", dir_path); + if (sizeof(adapter_path) == rv) { + log_error("adapter_path exceeds maximun value"); + goto error_dir_name; + } + + char plugin_path[128] = { 0 }; + rv = snprintf(plugin_path, 128, "%s/plugins.json", dir_path); + if (sizeof(plugin_path) == rv) { + log_error("plugin_path exceeds maximun value"); + goto error_dir_name; + } + + neu_persister_t *persister = malloc(sizeof(neu_persister_t)); + if (persister == NULL) { + log_error("failed to alloc memory for persister struct"); + goto error_dir_name; + } + + if (0 != ensure_file_exist(adapter_path, "{\"nodes\": []}")) { + log_error("persister failed to ensure file exist: %s", adapter_path); + goto error_io; + } + if (0 != ensure_file_exist(plugin_path, "{\"plugins\": []}")) { + log_error("persister failed to ensure file exist: %s", plugin_path); + goto error_io; + } + + persister->adapters_fname = strdup(adapter_path); + persister->plugins_fname = strdup(plugin_path); + persister->persist_dir = strdup(dir_name); + + return persister; + +error_io: + free(persister); +error_dir_name: + return NULL; +} + +static inline int persister_adapter_dir(char *buf, size_t size, + neu_persister_t *persister, + const char * adapter_name) +{ + size_t n = path_cat(buf, 0, size, persister->persist_dir); + if (size == n) { + return n; + } + + n = path_cat_escaped(buf, n, size, adapter_name); + return n; +} + +static inline int persister_group_configs_dir(char *buf, size_t size, + neu_persister_t *persister, + const char * adapter_name) +{ + size_t n = persister_adapter_dir(buf, size, persister, adapter_name); + if (size == n) { + return n; + } + + n = path_cat(buf, n, size, "groups"); + return n; +} + +static inline int persister_group_config_dir(char *buf, size_t size, + neu_persister_t *persister, + const char * adapter_name, + const char * group_config_name) +{ + size_t n = persister_group_configs_dir(buf, size, persister, adapter_name); + if (size == n) { + return n; + } + + n = path_cat_escaped(buf, n, size, group_config_name); + return n; +} + void neu_persister_destroy(neu_persister_t *persister) { free((char *) persister->adapters_fname); @@ -625,43 +802,38 @@ int neu_persister_load_subscriptions(neu_persister_t *persister, return 0; } -int neu_persister_store_group_configs( - neu_persister_t * persister, - neu_persist_group_config_info_t *group_config_infos) +int neu_persister_store_group_config( + neu_persister_t *persister, const char *adapter_name, + neu_persist_group_config_info_t *group_config_info) { - char group_config_path[128] = { 0 }; + char path[PATH_MAX_SIZE] = { 0 }; - int rv = snprintf(group_config_path, 128, "%s/persist/%s/group_configs", - persister->persist_dir, group_config_infos->adapter_name); - if (sizeof(group_config_path) == rv) { - log_error("group_config_path exceeds maximum value"); - return -1; - } - create_dir(group_config_path); - - char group_config_file[128] = { 0 }; - rv = snprintf(group_config_file, 128, "%s/%s", group_config_path, - group_config_infos->group_config_name); - if (sizeof(group_config_file) == rv) { - log_error("group_config_file exceeds maximum value"); + int n = + persister_group_config_dir(path, sizeof(path), persister, adapter_name, + group_config_info->group_config_name); + if (sizeof(path) == n) { + log_error("persister path too long: %s", path); return -1; } - neu_json_group_configs_resp_t group_config_resp = { - .adapter_name = group_config_infos->adapter_name, - .group_config_name = group_config_infos->group_config_name, - .read_interval = group_config_infos->read_interval, - .datatag_names = group_config_infos->datatag_names, - }; - - char *result = NULL; - rv = neu_json_encode_by_fn(&group_config_resp, - neu_json_encode_group_configs_resp, &result); - if (rv != 0) { + int rv = create_dir_recursive(path); + if (0 != rv) { + log_error("persister failed to create dir: %s", path); return rv; } - rv = write_file_string(group_config_file, result); + n = path_cat(path, n, sizeof(path), "config.json"); + if (sizeof(path) == n) { + log_error("persister path too long: %s", path); + return -1; + } + + char *result = NULL; + rv = neu_json_encode_by_fn(group_config_info, + neu_json_encode_group_configs_resp, &result); + if (rv == 0) { + rv = write_file_string(path, result); + } free(result); return rv; @@ -713,14 +885,21 @@ int neu_persister_load_group_configs(neu_persister_t *persister, return 0; } -int neu_persister_delete_group_configs(neu_persister_t *persister, - const char * adapter_name, - vector_t ** group_config_infos) +int neu_persister_delete_group_config(neu_persister_t *persister, + const char * adapter_name, + const char * group_config_name) { - (void *) persister; - (void *) adapter_name; - (void **) group_config_infos; - return 0; + char path[PATH_MAX_SIZE] = { 0 }; + + int n = persister_group_config_dir(path, sizeof(path), persister, + adapter_name, group_config_name); + if (sizeof(path) == n) { + log_error("persister path too long: %s", path); + return -1; + } + + int rv = rmdir_recursive(path); + return rv; } int neu_persister_store_adapter_setting(neu_persister_t *persister, diff --git a/src/persist/persist.h b/src/persist/persist.h index d7bb55e1..c5e6f2b0 100644 --- a/src/persist/persist.h +++ b/src/persist/persist.h @@ -175,7 +175,7 @@ int neu_persister_load_subscriptions(neu_persister_t *persister, * @return 0 on success, non-zero otherwise */ int neu_persister_store_group_config( - neu_persister_t * persister, + neu_persister_t *persister, const char *adapter_name, neu_persist_group_config_info_t *group_config_info); /** * Load all group config infos under an adapter.