feat(persistence): migrate subscription persistence to sqlite

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);
         }

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);
         }
     }
 

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);

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,

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.