Merge pull request #1022 from eeff/eeff/dev

feat: update subscription persistence with params string

persistence/0002_2.4.0_subscription_params.sql

@@ -0,0 +1,52 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2023 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ **/
+BEGIN TRANSACTION;
+
+DROP TABLE
+  old_subscriptions;
+
+ALTER TABLE
+  subscriptions RENAME TO old_subscriptions;
+
+CREATE TABLE
+  IF NOT EXISTS subscriptions (
+    app_name TEXT NOT NULL,
+    driver_name TEXT NOT NULL,
+    group_name TEXT NOT NULL,
+    params TEXT DEFAULT NULL,
+    CHECK (app_name != driver_name),
+    UNIQUE (app_name, driver_name, group_name),
+    FOREIGN KEY (app_name) REFERENCES nodes (name) ON UPDATE CASCADE ON DELETE CASCADE,
+    FOREIGN KEY (driver_name, group_name) REFERENCES groups (driver_name, name) ON UPDATE CASCADE ON DELETE CASCADE
+  );
+
+INSERT INTO
+  subscriptions (app_name, driver_name, group_name, params)
+SELECT
+  app_name,
+  driver_name,
+  group_name,
+  NULL
+FROM
+  old_subscriptions;
+
+DROP TABLE
+  old_subscriptions;
+
+COMMIT;

src/core/manager.c

@@ -406,7 +406,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         if (error.error == NEU_ERR_SUCCESS) {
             forward_msg(manager, msg, cmd->app);
             manager_storage_subscribe(manager, cmd->app, cmd->driver,
-                                      cmd->group);
+                                      cmd->group, cmd->params);
         }
 
         header->type = NEU_RESP_ERROR;

src/core/storage.c

@@ -61,10 +61,11 @@ void manager_storage_del_node(neu_manager_t *manager, const char *node)
 }
 
 void manager_storage_subscribe(neu_manager_t *manager, const char *app,
-                               const char *driver, const char *group)
+                               const char *driver, const char *group,
+                               const char *params)
 {
     (void) manager;
-    int rv = neu_persister_store_subscription(app, driver, group);
+    int rv = neu_persister_store_subscription(app, driver, group, params);
     if (0 != rv) {
         nlog_error("fail store subscription app:%s driver:%s group:%s", app,
                    driver, group);
@@ -148,7 +149,7 @@ int manager_load_subscribe(neu_manager_t *manager)
             {
                 rv = neu_manager_subscribe(manager, node->node,
                                            info->driver_name, info->group_name,
-                                           NULL);
+                                           info->params);
                 const char *ok_or_err = (0 == rv) ? "success" : "fail";
                 nlog_info("%s load subscription app:%s driver:%s grp:%s",
                           ok_or_err, node->node, info->driver_name,

src/core/storage.h

@@ -28,7 +28,8 @@ 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 *app,
-                               const char *driver, const char *group);
+                               const char *driver, const char *group,
+                               const char *params);
 void manager_storage_unsubscribe(neu_manager_t *manager, const char *app,
                                  const char *driver, const char *group);
 

src/persist/persist.c

@@ -847,13 +847,13 @@ int neu_persister_delete_tag(const char *driver_name, const char *group_name,
 
 int neu_persister_store_subscription(const char *app_name,
                                      const char *driver_name,
-                                     const char *group_name)
+                                     const char *group_name, const char *params)
 {
     return execute_sql(
         global_db,
-        "INSERT INTO subscriptions (app_name, driver_name, group_name) "
-        "VALUES (%Q, %Q, %Q)",
-        app_name, driver_name, group_name);
+        "INSERT INTO subscriptions (app_name, driver_name, group_name, params) "
+        "VALUES (%Q, %Q, %Q, %Q)",
+        app_name, driver_name, group_name, params);
 }
 
 static UT_icd subscription_info_icd = {
@@ -866,9 +866,9 @@ static UT_icd subscription_info_icd = {
 int neu_persister_load_subscriptions(const char *app_name,
                                      UT_array ** subscription_infos)
 {
-    sqlite3_stmt *stmt = NULL;
-    const char *  query =
-        "SELECT driver_name, group_name FROM subscriptions WHERE app_name=?";
+    sqlite3_stmt *stmt  = NULL;
+    const char *  query = "SELECT driver_name, group_name, params "
+                        "FROM subscriptions WHERE app_name=?";
 
     utarray_new(*subscription_infos, &subscription_info_icd);
 
@@ -896,9 +896,18 @@ int neu_persister_load_subscriptions(const char *app_name,
             break;
         }
 
+        char *params = (char *) sqlite3_column_text(stmt, 2);
+        // copy if params not NULL
+        if (NULL != params && NULL == (params = strdup(params))) {
+            free(group_name);
+            free(driver_name);
+            break;
+        }
+
         neu_persist_subscription_info_t info = {
             .driver_name = driver_name,
             .group_name  = group_name,
+            .params      = params,
         };
         utarray_push_back(*subscription_infos, &info);
 

src/persist/persist.h

@@ -44,6 +44,7 @@ typedef struct {
 typedef struct {
     char *driver_name;
     char *group_name;
+    char *params;
 } neu_persist_subscription_info_t;
 
 typedef struct {
@@ -72,6 +73,7 @@ neu_persist_subscription_info_fini(neu_persist_subscription_info_t *info)
 {
     free(info->driver_name);
     free(info->group_name);
+    free(info->params);
 }
 
 static inline void neu_persist_user_info_fini(neu_persist_user_info_t *info)
@@ -189,11 +191,13 @@ int neu_persister_delete_tag(const char *driver_name, const char *group_name,
  * @param app_name                  name of the app node
  * @param driver_name               name of the driver node
  * @param group_name                name of the group
+ * @param params                    subscription params
  * @return 0 on success, non-zero otherwise
  */
 int neu_persister_store_subscription(const char *app_name,
                                      const char *driver_name,
-                                     const char *group_name);
+                                     const char *group_name,
+                                     const char *params);
 
 /**
  * Load adapter subscriptions.