From 6b6ed493de2b8205603dbd7781c043bec3fe78f6 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Thu, 14 Jul 2022 16:16:27 +0800 Subject: [PATCH] feat(persistence): migrate subscription persistence to sqlite --- src/core/manager.c | 6 +- src/core/storage.c | 40 ++++++----- src/core/storage.h | 5 +- src/persist/persist.c | 152 +++++++++++++++++++----------------------- src/persist/persist.h | 56 +++++++++------- 5 files changed, 134 insertions(+), 125 deletions(-) diff --git a/src/core/manager.c b/src/core/manager.c index 62835ea7..f59bc449 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -297,7 +297,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_manager_subscribe(manager, cmd->app, cmd->driver, cmd->group); if (error.error == NEU_ERR_SUCCESS) { - manager_storage_subscribe(manager, cmd->app); + manager_storage_subscribe(manager, cmd->app, cmd->driver, + cmd->group); neu_manager_notify_app_sub(manager, cmd->app, cmd->driver, cmd->group); } @@ -315,7 +316,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_manager_unsubscribe(manager, cmd->app, cmd->driver, cmd->group); if (error.error == NEU_ERR_SUCCESS) { - manager_storage_subscribe(manager, cmd->app); + manager_storage_unsubscribe(manager, cmd->app, cmd->driver, + cmd->group); neu_manager_notify_app_unsub(manager, cmd->app, cmd->driver, cmd->group); } diff --git a/src/core/storage.c b/src/core/storage.c index 53f66422..54b22673 100644 --- a/src/core/storage.c +++ b/src/core/storage.c @@ -59,18 +59,26 @@ void manager_storage_del_node(neu_manager_t *manager, const char *node) neu_persister_delete_node(manager->persister, node); } -void manager_storage_subscribe(neu_manager_t *manager, const char *app) +void manager_storage_subscribe(neu_manager_t *manager, const char *app, + const char *driver, const char *group) { - - UT_array *sub_infos = neu_manager_get_sub_group(manager, app); - - int rv = - neu_persister_store_subscriptions(manager->persister, app, sub_infos); + int rv = neu_persister_store_subscription(manager->persister, app, driver, + group); if (0 != rv) { - nlog_error("fail store adapter:%s subscription infos", app); + nlog_error("fail store subscription app:%s driver:%s group:%s", app, + driver, group); } +} - utarray_free(sub_infos); +void manager_storage_unsubscribe(neu_manager_t *manager, const char *app, + const char *driver, const char *group) +{ + int rv = neu_persister_delete_subscription(manager->persister, app, driver, + group); + if (0 != rv) { + nlog_error("fail delete subscription app:%s driver:%s group:%s", app, + driver, group); + } } int manager_load_plugin(neu_manager_t *manager) @@ -137,21 +145,19 @@ int manager_load_subscribe(neu_manager_t *manager) } else { utarray_foreach(sub_infos, neu_persist_subscription_info_t *, info) { - rv = neu_manager_subscribe(manager, info->sub_adapter_name, - info->src_adapter_name, - info->group_config_name); + rv = neu_manager_subscribe(manager, node->node, + info->driver_name, info->group_name); const char *ok_or_err = (0 == rv) ? "success" : "fail"; nlog_info("%s load subscription app:%s driver:%s grp:%s", - ok_or_err, info->sub_adapter_name, - info->src_adapter_name, info->group_config_name); + ok_or_err, node->node, info->driver_name, + info->group_name); if (rv == 0) { - neu_manager_notify_app_sub_update(manager, - info->src_adapter_name, - info->group_config_name); + neu_manager_notify_app_sub_update( + manager, info->driver_name, info->group_name); } } - neu_persist_subscription_infos_free(sub_infos); + utarray_free(sub_infos); } } diff --git a/src/core/storage.h b/src/core/storage.h index db8ecd65..5151898f 100644 --- a/src/core/storage.h +++ b/src/core/storage.h @@ -27,7 +27,10 @@ void manager_strorage_plugin(neu_manager_t *manager); void manager_storage_del_node(neu_manager_t *manager, const char *node); void manager_storage_add_node(neu_manager_t *manager, const char *node); -void manager_storage_subscribe(neu_manager_t *manager, const char *node); +void manager_storage_subscribe(neu_manager_t *manager, const char *app, + const char *driver, const char *group); +void manager_storage_unsubscribe(neu_manager_t *manager, const char *app, + const char *driver, const char *group); int manager_load_plugin(neu_manager_t *manager); int manager_load_node(neu_manager_t *manager); diff --git a/src/persist/persist.c b/src/persist/persist.c index 46aa8474..c1e96a9c 100644 --- a/src/persist/persist.c +++ b/src/persist/persist.c @@ -710,105 +710,93 @@ int neu_persister_delete_tag(neu_persister_t *persister, driver_name, group_name, tag_name); } -int neu_persister_store_subscriptions(neu_persister_t *persister, - const char * adapter_name, - UT_array * subscription_infos) +int neu_persister_store_subscription(neu_persister_t *persister, + const char * app_name, + const char * driver_name, + const char * group_name) { - char path[PATH_MAX_SIZE] = { 0 }; - - int n = persister_adapter_dir(path, sizeof(path), persister, adapter_name); - if (sizeof(path) == n) { - nlog_error("path too long: %s", path); - return -1; - } - - int rv = create_dir(path); - if (0 != rv) { - nlog_error("fail to create dir: %s", path); - return rv; - } - - n = path_cat(path, n, sizeof(path), "subscriptions.json"); - if (sizeof(path) == n) { - nlog_error("path too long: %s", path); - return -1; - } - - neu_json_subscriptions_req_t subs_resp = { 0 }; - int index = 0; - subs_resp.n_subscription = utarray_len(subscription_infos); - subs_resp.subscriptions = - calloc(subs_resp.n_subscription, - sizeof(neu_json_subscriptions_req_subscription_t)); - utarray_foreach(subscription_infos, neu_resp_subscribe_info_t *, info) - { - subs_resp.subscriptions[index].group_config_name = info->group; - subs_resp.subscriptions[index].src_adapter_name = info->driver; - subs_resp.subscriptions[index].sub_adapter_name = info->app; - index += 1; - } - - char *result = NULL; - rv = neu_json_encode_by_fn(&subs_resp, neu_json_encode_subscriptions_resp, - &result); - if (rv == 0) { - rv = write_file_string(path, result); - } - - free(result); - free(subs_resp.subscriptions); - - return rv; + return neu_persister_sql_exec( + persister, + "INSERT INTO subscriptions (app_name, driver_name, group_name) VALUES " + "(%Q, %Q, %Q)", + app_name, driver_name, group_name); } +static UT_icd subscription_info_icd = { + sizeof(neu_persist_subscription_info_t), + NULL, + NULL, + (dtor_f *) neu_persist_subscription_info_fini, +}; + int neu_persister_load_subscriptions(neu_persister_t *persister, - const char * adapter_name, + const char * app_name, UT_array ** subscription_infos) { - char path[PATH_MAX_SIZE] = { 0 }; + sqlite3_stmt *stmt = NULL; + const char * query = + "SELECT driver_name, group_name FROM subscriptions WHERE app_name=?"; - int n = persister_adapter_dir(path, sizeof(path), persister, adapter_name); - if (sizeof(path) == n) { - nlog_error("path too long: %s", path); - return -1; + utarray_new(*subscription_infos, &subscription_info_icd); + + if (SQLITE_OK != + sqlite3_prepare_v2(persister->db, query, -1, &stmt, NULL)) { + nlog_error("prepare `%s` fail: %s", query, + sqlite3_errmsg(persister->db)); + goto error; } - n = path_cat(path, n, sizeof(path), "subscriptions.json"); - if (sizeof(path) == n) { - nlog_error("path too long: %s", path); - return -1; + if (SQLITE_OK != sqlite3_bind_text(stmt, 1, app_name, -1, NULL)) { + nlog_error("bind `%s` with `%s` fail: %s", query, app_name, + sqlite3_errmsg(persister->db)); + goto error; } - char *json_str = NULL; - int rv = read_file_string(path, &json_str); - if (rv != 0) { - return rv; - } + int step = sqlite3_step(stmt); + while (SQLITE_ROW == step) { + char *driver_name = strdup((char *) sqlite3_column_text(stmt, 0)); + if (NULL == driver_name) { + break; + } - neu_json_subscriptions_req_t *subs_req = NULL; - rv = neu_json_decode_subscriptions_req(json_str, &subs_req); - free(json_str); - if (0 != rv) { - return rv; - } - - UT_icd icd = { sizeof(neu_persist_subscription_info_t), NULL, NULL, NULL }; - utarray_new(*subscription_infos, &icd); - - for (int i = 0; i < subs_req->n_subscription; i++) { - neu_persist_subscription_info_t info = { 0 }; - info.group_config_name = - strdup(subs_req->subscriptions[i].group_config_name); - info.src_adapter_name = - strdup(subs_req->subscriptions[i].src_adapter_name); - info.sub_adapter_name = - strdup(subs_req->subscriptions[i].sub_adapter_name); + char *group_name = strdup((char *) sqlite3_column_text(stmt, 1)); + if (NULL == group_name) { + free(driver_name); + break; + } + neu_persist_subscription_info_t info = { + .driver_name = driver_name, + .group_name = group_name, + }; utarray_push_back(*subscription_infos, &info); + + step = sqlite3_step(stmt); } - neu_json_decode_subscriptions_req_free(subs_req); + if (SQLITE_DONE != step) { + nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(persister->db)); + // do not set return code, return partial or empty result + } + + sqlite3_finalize(stmt); return 0; + +error: + utarray_free(*subscription_infos); + *subscription_infos = NULL; + return NEU_ERR_EINTERNAL; +} + +int neu_persister_delete_subscription(neu_persister_t *persister, + const char * app_name, + const char * driver_name, + const char * group_name) +{ + return neu_persister_sql_exec(persister, + "DELETE FROM subscriptions WHERE app_name=%Q " + "AND driver_name=%Q AND group_name=%Q", + app_name, driver_name, group_name); } int neu_persister_store_group(neu_persister_t * persister, diff --git a/src/persist/persist.h b/src/persist/persist.h index a033033b..35de0551 100644 --- a/src/persist/persist.h +++ b/src/persist/persist.h @@ -26,7 +26,6 @@ extern "C" { #include "adapter/adapter_info.h" #include "persist/json/persist_json_plugin.h" -#include "persist/json/persist_json_subs.h" typedef neu_json_plugin_req_plugin_t neu_persist_plugin_info_t; @@ -42,8 +41,10 @@ typedef struct { char * name; } neu_persist_group_info_t; -typedef neu_json_subscriptions_req_subscription_t - neu_persist_subscription_info_t; +typedef struct { + char *driver_name; + char *group_name; +} neu_persist_subscription_info_t; static inline void neu_persist_plugin_infos_free(UT_array *plugin_infos) { @@ -62,15 +63,10 @@ static inline void neu_persist_group_info_fini(neu_persist_group_info_t *info) } static inline void -neu_persist_subscription_infos_free(UT_array *subscription_infos) +neu_persist_subscription_info_fini(neu_persist_subscription_info_t *info) { - utarray_foreach(subscription_infos, neu_persist_subscription_info_t *, info) - { - free(info->group_config_name); - free(info->src_adapter_name); - free(info->sub_adapter_name); - } - utarray_free(subscription_infos); + free(info->driver_name); + free(info->group_name); } /** @@ -192,29 +188,43 @@ int neu_persister_delete_tag(neu_persister_t *persister, const char *tag_name); /** - * Persist adapter subscriptions. + * Persist subscriptions. * @param persister persiter object. - * @param adapter_name name of the adapter who owns the - * subscriptions. - * @param subscription_infos vector of neu_persist_subscription_info_t. + * @param app_name name of the app node + * @param driver_name name of the driver node + * @param group_name name of the group * @return 0 on success, non-zero otherwise */ -int neu_persister_store_subscriptions(neu_persister_t *persister, - const char * adapter_name, - UT_array * subscription_infos); +int neu_persister_store_subscription(neu_persister_t *persister, + const char * app_name, + const char * driver_name, + const char * group_name); + /** - * Load adapter subscription infos. + * Load adapter subscriptions. * @param persister persiter object. - * @param adapter_name name of the adapter who owns the - * subscriptions. + * @param app_name name of the app node * @param[out] subscription_infos used to return pointer to heap allocated - * vector of neu_persist_subscription_info_t. + * vector of neu_persist_subscription_info_t. * @return 0 on success, non-zero otherwise */ int neu_persister_load_subscriptions(neu_persister_t *persister, - const char * adapter_name, + const char * app_name, UT_array ** subscription_infos); +/** + * Persist subscriptions. + * @param persister persiter object. + * @param app_name name of the app node + * @param driver_name name of the driver node + * @param group_name name of the group + * @return 0 on success, non-zero otherwise + */ +int neu_persister_delete_subscription(neu_persister_t *persister, + const char * app_name, + const char * driver_name, + const char * group_name); + /** * Persist group config. * @param persister persiter object.