diff --git a/persistence/0002_2.4.0_subscription_params.sql b/persistence/0002_2.4.0_subscription_params.sql new file mode 100644 index 00000000..40a383ea --- /dev/null +++ b/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; diff --git a/src/core/manager.c b/src/core/manager.c index 103c3260..95219b64 100644 --- a/src/core/manager.c +++ b/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; diff --git a/src/core/storage.c b/src/core/storage.c index c706b7d7..8a9b0619 100644 --- a/src/core/storage.c +++ b/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, diff --git a/src/core/storage.h b/src/core/storage.h index 5151898f..99aae4d5 100644 --- a/src/core/storage.h +++ b/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); diff --git a/src/persist/persist.c b/src/persist/persist.c index babf43ff..f1dacdf6 100644 --- a/src/persist/persist.c +++ b/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); diff --git a/src/persist/persist.h b/src/persist/persist.h index b145e5af..09cef08f 100644 --- a/src/persist/persist.h +++ b/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.