GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-22 01:05:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #842 from fengzeroz/v2.2

Browse Source 
use a globally unique db instance
This commit is contained in:
fengzeroz
2022-10-20 14:31:11 +08:00
committed by GitHub
parent f071f60321 0e938f3f63
commit 0030ecc234
11 changed files with 229 additions and 574 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapter/adapter.c
+16 -26
View File
@@ -81,8 +81,6 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
adapter->cb_funs.command = callback_funs.command;
adapter->cb_funs.response = callback_funs.response;
adapter->module = info->module;
adapter->persister = neu_persister_create("persistence");
assert(adapter->persister != NULL);
rv = nng_pair1_open(&adapter->sock);
assert(rv == 0);
@@ -110,8 +108,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
strcpy(common->name, adapter->name);
adapter->module->intf_funs->init(adapter->plugin);
if (adapter_load_setting(adapter->persister, adapter->name,
&adapter->setting) == 0) {
if (adapter_load_setting(adapter->name, &adapter->setting) == 0) {
if (adapter->module->intf_funs->setting(adapter->plugin,
adapter->setting) == 0) {
adapter->state = NEU_NODE_RUNNING_STATE_READY;
@@ -122,8 +119,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
}
if (info->module->type == NEU_NA_TYPE_DRIVER) {
adapter_load_group_and_tag(adapter->persister,
(neu_adapter_driver_t *) adapter);
adapter_load_group_and_tag((neu_adapter_driver_t *) adapter);
}
param.fd = adapter->recv_fd;
@@ -139,7 +135,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
nlog_info("Success to create adapter: %s", adapter->name);
adapter_storage_state(adapter->persister, adapter->name, adapter->state);
adapter_storage_state(adapter->name, adapter->state);
return adapter;
}
@@ -334,8 +330,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
error.error = neu_adapter_set_setting(adapter, cmd->setting);
if (error.error == NEU_ERR_SUCCESS) {
adapter_storage_setting(adapter->persister, adapter->name,
cmd->setting);
adapter_storage_setting(adapter->name, cmd->setting);
}
free(cmd->setting);
@@ -429,8 +424,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
if (error.error == NEU_ERR_SUCCESS) {
adapter_storage_add_group(adapter->persister, adapter->name,
cmd->group, cmd->interval);
adapter_storage_add_group(adapter->name, cmd->group, cmd->interval);
}
neu_msg_exchange(header);
@@ -455,8 +449,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
if (error.error == NEU_ERR_SUCCESS) {
adapter_storage_update_group(adapter->persister, adapter->name,
cmd->group, cmd->interval);
adapter_storage_update_group(adapter->name, cmd->group,
cmd->interval);
}
neu_msg_exchange(header);
@@ -476,8 +470,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
if (error.error == NEU_ERR_SUCCESS) {
adapter_storage_del_group(adapter->persister, cmd->driver,
cmd->group);
adapter_storage_del_group(cmd->driver, cmd->group);
}
neu_msg_exchange(header);
@@ -514,8 +507,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
int ret = neu_adapter_driver_del_tag(
(neu_adapter_driver_t *) adapter, cmd->group, cmd->tags[i]);
if (0 == ret) {
adapter_storage_del_tag(adapter->persister, cmd->driver,
cmd->group, cmd->tags[i]);
adapter_storage_del_tag(cmd->driver, cmd->group,
cmd->tags[i]);
}
}
}
@@ -552,8 +545,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
if (resp.index) {
// we have added some tags, try to persist
adapter_storage_add_tags(adapter->persister, cmd->driver,
cmd->group, cmd->tags, resp.index);
adapter_storage_add_tags(cmd->driver, cmd->group, cmd->tags,
resp.index);
}
for (int i = 0; i < cmd->n_tag; i++) {
@@ -580,8 +573,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
(neu_adapter_driver_t *) adapter, cmd->group,
&cmd->tags[i]);
if (ret == 0) {
adapter_storage_update_tag(adapter->persister, cmd->driver,
cmd->group, &cmd->tags[i]);
adapter_storage_update_tag(cmd->driver, cmd->group,
&cmd->tags[i]);
resp.index += 1;
} else {
@@ -641,7 +634,6 @@ void neu_adapter_destroy(neu_adapter_t *adapter)
nng_dialer_close(adapter->dialer);
nng_close(adapter->sock);
neu_persister_destroy(adapter->persister);
if (adapter->name != NULL) {
free(adapter->name);
@@ -697,8 +689,7 @@ int neu_adapter_start(neu_adapter_t *adapter)
error = intf_funs->start(adapter->plugin);
if (error == NEU_ERR_SUCCESS) {
adapter->state = NEU_NODE_RUNNING_STATE_RUNNING;
adapter_storage_state(adapter->persister, adapter->name,
adapter->state);
adapter_storage_state(adapter->name, adapter->state);
}
return error;
@@ -736,8 +727,7 @@ int neu_adapter_stop(neu_adapter_t *adapter)
error = intf_funs->stop(adapter->plugin);
if (error == NEU_ERR_SUCCESS) {
adapter->state = NEU_NODE_RUNNING_STATE_STOPPED;
adapter_storage_state(adapter->persister, adapter->name,
adapter->state);
adapter_storage_state(adapter->name, adapter->state);
}
return error;
src/adapter/adapter_internal.h
-1
View File
@@ -38,7 +38,6 @@ struct neu_adapter {
neu_node_running_state_e state;
neu_persister_t * persister;
adapter_callbacks_t cb_funs;
void * handle;
src/adapter/driver/driver.c
+1 -2
View File
@@ -529,8 +529,7 @@ int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
HASH_FIND_STR(driver->groups, group, find);
if (find == NULL) {
neu_adapter_driver_add_group(driver, group, 3000);
adapter_storage_add_group(driver->adapter.persister,
driver->adapter.name, group, 3000);
adapter_storage_add_group(driver->adapter.name, group, 3000);
}
HASH_FIND_STR(driver->groups, group, find);
assert(find != NULL);
src/adapter/storage.c
+30 -37
View File
@@ -23,103 +23,97 @@
#include "driver/driver_internal.h"
#include "storage.h"
void adapter_storage_state(neu_persister_t *persister, const char *node,
neu_node_running_state_e state)
void adapter_storage_state(const char *node, neu_node_running_state_e state)
{
neu_persister_update_node_state(persister, node, state);
neu_persister_update_node_state(node, state);
}
void adapter_storage_setting(neu_persister_t *persister, const char *node,
const char *setting)
void adapter_storage_setting(const char *node, const char *setting)
{
neu_persister_store_node_setting(persister, node, setting);
neu_persister_store_node_setting(node, setting);
}
void adapter_storage_add_group(neu_persister_t *persister, const char *node,
const char *group, uint32_t interval)
void adapter_storage_add_group(const char *node, const char *group,
uint32_t interval)
{
neu_persist_group_info_t info = {
.name = (char *) group,
.interval = interval,
};
neu_persister_store_group(persister, node, &info);
neu_persister_store_group(node, &info);
}
void adapter_storage_update_group(neu_persister_t *persister, const char *node,
const char *group, uint32_t interval)
void adapter_storage_update_group(const char *node, const char *group,
uint32_t interval)
{
neu_persist_group_info_t info = {
.name = (char *) group,
.interval = interval,
};
int rv = neu_persister_update_group(persister, node, &info);
int rv = neu_persister_update_group(node, &info);
if (0 != rv) {
nlog_error("fail update adapter:%s group:%s, interval: %u", node, group,
interval);
}
}
void adapter_storage_del_group(neu_persister_t *persister, const char *node,
const char *group)
void adapter_storage_del_group(const char *node, const char *group)
{
neu_persister_delete_group(persister, node, group);
neu_persister_delete_group(node, group);
}
void adapter_storage_add_tag(neu_persister_t *persister, const char *node,
const char *group, const neu_datatag_t *tag)
void adapter_storage_add_tag(const char *node, const char *group,
const neu_datatag_t *tag)
{
int rv = neu_persister_store_tag(persister, node, group, tag);
int rv = neu_persister_store_tag(node, group, tag);
if (0 != rv) {
nlog_error("fail store tag:%s adapter:%s grp:%s", tag->name, node,
group);
}
}
void adapter_storage_add_tags(neu_persister_t *persister, const char *node,
const char *group, const neu_datatag_t *tags,
size_t n)
void adapter_storage_add_tags(const char *node, const char *group,
const neu_datatag_t *tags, size_t n)
{
if (0 == n) {
return;
}
if (1 == n) {
return adapter_storage_add_tag(persister, node, group, &tags[0]);
return adapter_storage_add_tag(node, group, &tags[0]);
}
int rv = neu_persister_store_tags(persister, node, group, tags, n);
int rv = neu_persister_store_tags(node, group, tags, n);
if (0 != rv) {
nlog_error("fail store %zu tags:[%s ... %s] adapter:%s grp:%s", n,
tags->name, tags[n - 1].name, node, group);
}
}
void adapter_storage_update_tag(neu_persister_t *persister, const char *node,
const char *group, const neu_datatag_t *tag)
void adapter_storage_update_tag(const char *node, const char *group,
const neu_datatag_t *tag)
{
int rv = neu_persister_update_tag(persister, node, group, tag);
int rv = neu_persister_update_tag(node, group, tag);
if (0 != rv) {
nlog_error("fail update tag:%s adapter:%s grp:%s", tag->name, node,
group);
}
}
void adapter_storage_del_tag(neu_persister_t *persister, const char *node,
const char *group, const char *name)
void adapter_storage_del_tag(const char *node, const char *group,
const char *name)
{
int rv = neu_persister_delete_tag(persister, node, group, name);
int rv = neu_persister_delete_tag(node, group, name);
if (0 != rv) {
nlog_error("fail del tag:%s adapter:%s grp:%s", name, node, group);
}
}
int adapter_load_setting(neu_persister_t *persister, const char *node,
char **setting)
int adapter_load_setting(const char *node, char **setting)
{
int rv = neu_persister_load_node_setting(persister, node,
(const char **) setting);
int rv = neu_persister_load_node_setting(node, (const char **) setting);
if (0 != rv) {
nlog_info("load %s setting fail", node);
return -1;
@@ -128,13 +122,12 @@ int adapter_load_setting(neu_persister_t *persister, const char *node,
return 0;
}
int adapter_load_group_and_tag(neu_persister_t * persister,
neu_adapter_driver_t *driver)
int adapter_load_group_and_tag(neu_adapter_driver_t *driver)
{
UT_array * group_infos = NULL;
neu_adapter_t *adapter = (neu_adapter_t *) driver;
int rv = neu_persister_load_groups(persister, adapter->name, &group_infos);
int rv = neu_persister_load_groups(adapter->name, &group_infos);
if (0 != rv) {
nlog_warn("load %s group fail", adapter->name);
return rv;
@@ -145,7 +138,7 @@ int adapter_load_group_and_tag(neu_persister_t * persister,
UT_array *tags = NULL;
neu_adapter_driver_add_group(driver, p->name, p->interval);
rv = neu_persister_load_tags(persister, adapter->name, p->name, &tags);
rv = neu_persister_load_tags(adapter->name, p->name, &tags);
if (0 != rv) {
nlog_warn("load %s:%s tags fail", adapter->name, p->name);
continue;
src/adapter/storage.h
+17 -23
View File
@@ -24,29 +24,23 @@
#include "adapter_internal.h"
void adapter_storage_state(neu_persister_t *persister, const char *node,
neu_node_running_state_e state);
void adapter_storage_setting(neu_persister_t *persister, const char *node,
const char *setting);
void adapter_storage_add_group(neu_persister_t *persister, const char *node,
const char *group, uint32_t interval);
void adapter_storage_update_group(neu_persister_t *persister, const char *node,
const char *group, uint32_t interval);
void adapter_storage_del_group(neu_persister_t *persister, const char *node,
const char *group);
void adapter_storage_add_tag(neu_persister_t *persister, const char *node,
const char *group, const neu_datatag_t *tag);
void adapter_storage_add_tags(neu_persister_t *persister, const char *node,
const char *group, const neu_datatag_t *tags,
size_t n);
void adapter_storage_update_tag(neu_persister_t *persister, const char *node,
const char *group, const neu_datatag_t *tag);
void adapter_storage_del_tag(neu_persister_t *persister, const char *node,
const char *group, const char *name);
void adapter_storage_state(const char *node, neu_node_running_state_e state);
void adapter_storage_setting(const char *node, const char *setting);
void adapter_storage_add_group(const char *node, const char *group,
uint32_t interval);
void adapter_storage_update_group(const char *node, const char *group,
uint32_t interval);
void adapter_storage_del_group(const char *node, const char *group);
void adapter_storage_add_tag(const char *node, const char *group,
const neu_datatag_t *tag);
void adapter_storage_add_tags(const char *node, const char *group,
const neu_datatag_t *tags, size_t n);
void adapter_storage_update_tag(const char *node, const char *group,
const neu_datatag_t *tag);
void adapter_storage_del_tag(const char *node, const char *group,
const char *name);
int adapter_load_setting(neu_persister_t *persister, const char *node,
char **setting);
int adapter_load_group_and_tag(neu_persister_t * persister,
neu_adapter_driver_t *driver);
int adapter_load_setting(const char *node, char **setting);
int adapter_load_group_and_tag(neu_adapter_driver_t *driver);
#endif
src/core/manager.c
-3
View File
@@ -77,8 +77,6 @@ neu_manager_t *neu_manager_create()
manager->node_manager = neu_node_manager_create();
manager->subscribe_manager = neu_subscribe_manager_create();
manager->sub_msg_manager = neu_sub_msg_manager_create();
manager->persister = neu_persister_create("persistence");
assert(manager->persister != NULL);
rv = nng_pair1_open_poly(&manager->socket);
assert(rv == 0);
@@ -149,7 +147,6 @@ void neu_manager_destroy(neu_manager_t *manager)
}
neu_sub_msg_manager_destroy(manager->sub_msg_manager);
neu_persister_destroy(manager->persister);
neu_subscribe_manager_destroy(manager->subscribe_manager);
neu_node_manager_destroy(manager->node_manager);
neu_plugin_manager_destroy(manager->plugin_manager);
src/core/manager_internal.h
-2
View File
@@ -41,8 +41,6 @@ typedef struct neu_manager {
neu_node_manager_t * node_manager;
neu_subscribe_mgr_t * subscribe_manager;
neu_sub_msg_mgr_t * sub_msg_manager;
neu_persister_t *persister;
} neu_manager_t;
int neu_manager_add_plugin(neu_manager_t *manager, const char *library);
src/core/storage.c
+11 -11
View File
@@ -26,7 +26,7 @@ void manager_strorage_plugin(neu_manager_t *manager)
plugin_infos = neu_manager_get_plugins(manager);
int rv = neu_persister_store_plugins(manager->persister, plugin_infos);
int rv = neu_persister_store_plugins(plugin_infos);
if (0 != rv) {
nlog_error("failed to store plugins infos");
}
@@ -46,7 +46,7 @@ void manager_storage_add_node(neu_manager_t *manager, const char *node)
return;
}
rv = neu_persister_store_node(manager->persister, &node_info);
rv = neu_persister_store_node(&node_info);
if (0 != rv) {
nlog_error("failed to store adapter info");
}
@@ -56,14 +56,15 @@ void manager_storage_add_node(neu_manager_t *manager, const char *node)
void manager_storage_del_node(neu_manager_t *manager, const char *node)
{
neu_persister_delete_node(manager->persister, node);
(void) manager;
neu_persister_delete_node(node);
}
void manager_storage_subscribe(neu_manager_t *manager, const char *app,
const char *driver, const char *group)
{
int rv = neu_persister_store_subscription(manager->persister, app, driver,
group);
(void) manager;
int rv = neu_persister_store_subscription(app, driver, group);
if (0 != rv) {
nlog_error("fail store subscription app:%s driver:%s group:%s", app,
driver, group);
@@ -73,8 +74,8 @@ void manager_storage_subscribe(neu_manager_t *manager, const char *app,
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);
(void) manager;
int rv = neu_persister_delete_subscription(app, driver, group);
if (0 != rv) {
nlog_error("fail delete subscription app:%s driver:%s group:%s", app,
driver, group);
@@ -85,7 +86,7 @@ int manager_load_plugin(neu_manager_t *manager)
{
UT_array *plugin_infos = NULL;
int rv = neu_persister_load_plugins(manager->persister, &plugin_infos);
int rv = neu_persister_load_plugins(&plugin_infos);
if (rv != 0) {
return rv;
}
@@ -108,7 +109,7 @@ int manager_load_node(neu_manager_t *manager)
UT_array *node_infos = NULL;
int rv = 0;
rv = neu_persister_load_nodes(manager->persister, &node_infos);
rv = neu_persister_load_nodes(&node_infos);
if (0 != rv) {
nlog_error("failed to load adapter infos");
return -1;
@@ -139,8 +140,7 @@ int manager_load_subscribe(neu_manager_t *manager)
int rv = 0;
UT_array *sub_infos = NULL;
rv = neu_persister_load_subscriptions(manager->persister, node->node,
&sub_infos);
rv = neu_persister_load_subscriptions(node->node, &sub_infos);
if (0 != rv) {
nlog_warn("load %s subscribetion infos error", node->node);
} else {
src/main.c
+6 -1
View File
@@ -40,6 +40,7 @@ static void sig_handler(int sig)
if (sig != SIGABRT) {
neu_manager_destroy(g_manager);
neu_persister_destroy();
}
zlog_fini();
exit_flag = true;
@@ -47,7 +48,8 @@ static void sig_handler(int sig)
static int neuron_run(const neu_cli_args_t *args)
{
struct rlimit rl;
struct rlimit rl = { 0 };
int rv = 0;
signal(SIGINT, sig_handler);
signal(SIGTERM, sig_handler);
@@ -59,6 +61,9 @@ static int neuron_run(const neu_cli_args_t *args)
nlog_warn("neuron process failed enable core dump, ignore");
}
rv = neu_persister_create();
assert(rv == 0);
zlog_notice(neuron, "neuron process, daemon: %d, version: %s (%s %s)",
args->daemonized, NEURON_VERSION,
NEURON_GIT_REV NEURON_GIT_DIFF, NEURON_BUILD_DATE);
src/persist/persist.c
+115 -387
View File
@@ -49,6 +49,10 @@
#define PATH_MAX_SIZE 128
static const char *plugin_file = "persistence/plugins.json";
static const char *db_file = "persistence/sqlite.db";
static sqlite3 * global_db = NULL;
static inline bool ends_with(const char *str, const char *suffix)
{
size_t m = strlen(str);
@@ -56,74 +60,6 @@ static inline bool ends_with(const char *str, const char *suffix)
return m >= n && !strcmp(str + m - n, suffix);
}
typedef struct neu_persister {
const char *persist_dir;
const char *plugins_fname;
sqlite3 * db;
} neu_persister_t;
const char *neu_persister_get_persist_dir(neu_persister_t *persister)
{
if (persister == NULL) {
return NULL;
}
return persister->persist_dir;
}
const char *neu_persister_get_plugins_fname(neu_persister_t *persister)
{
if (persister == NULL) {
return NULL;
}
return persister->plugins_fname;
}
/**
* Escape special characters in path string,
* '%' -> '%%', '/' => '%-', '.' -> '%.'
*
* @param buf output buffer,
* @param size size of output buffer
* @param path path string
*
* @return if `buf` is NULL, return number of bytes needed to store the result.
* otherwise, return the number of bytes written excluding the
* terminating null byte, `size` indicates overflow.
*/
static int path_escape(char *buf, size_t size, const char *path)
{
size_t i = 0;
char c;
if (NULL != buf) {
for (const char *s = path; (c = *s++) && i < size; ++i) {
if (PATH_SEP_CHAR == c || '.' == c || '%' == c) {
c = (PATH_SEP_CHAR == c) ? '-' : c;
buf[i++] = '%';
if (size == i) {
break;
}
}
buf[i] = c;
}
if (i < size) {
buf[i] = '\0';
} else if (i > 0) {
buf[i - 1] = '\0';
}
} else {
for (const char *s = path; (c = *s++); ++i) {
if (PATH_SEP_CHAR == c || '.' == c || '%' == c) {
++i;
}
}
}
return i;
}
/**
* Concatenate a path string to another.
*
@@ -158,66 +94,6 @@ static int path_cat(char *dst, size_t len, size_t size, const char *src)
return i;
}
/**
* Escape and concatenate a path string to another.
*
* @param dst destination path string buffer
* @param len destination path string len, not greater than size
* @param size destination path buffer size
* @param src path string
*
* @return length of the result path string excluding the terminating NULL
* byte, `size` indicates overflow.
*/
static int path_cat_escaped(char *dst, size_t len, size_t size, const char *src)
{
size_t i = len;
if (0 < i && i < size && (PATH_SEP_CHAR != dst[i - 1])) {
dst[i++] = PATH_SEP_CHAR;
dst[i] = '\0';
}
size_t n = path_escape(dst + i, size - i, src);
return i + n;
}
static inline int create_dir(char *dir_name)
{
int rv = mkdir(dir_name, 0777);
if (0 != rv && EEXIST == errno) {
rv = 0;
}
return rv;
}
/**
* 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)
{
int fd = open(name, O_CREAT | O_WRONLY, S_IRUSR | S_IWUSR);
if (-1 == fd) {
return -1;
}
struct stat statbuf;
if (-1 == fstat(fd, &statbuf)) {
return -1;
}
if (0 == statbuf.st_size) {
ssize_t size = strlen(default_content);
if (size != write(fd, default_content, size)) {
return -1;
}
}
return 0;
}
static int write_file_string(const char *fn, const char *s)
{
char *tmp = NULL;
@@ -554,167 +430,49 @@ static int apply_schemas(sqlite3 *db, const char *dir)
return rv;
}
neu_persister_t *neu_persister_create(const char *dir_name)
int neu_persister_create()
{
int rv = 0;
char *persist_dir = NULL;
char *plugins_fname = NULL;
char *path = NULL;
int rv = sqlite3_open(db_file, &global_db);
path = calloc(PATH_MAX_SIZE, sizeof(char));
if (NULL == path) {
nlog_error("malloc fail");
goto error;
}
int dir_len = path_cat(path, 0, PATH_MAX_SIZE, dir_name);
if (PATH_MAX_SIZE <= dir_len) {
nlog_error("path too long: %s", dir_name);
goto error;
}
persist_dir = strdup(dir_name);
if (NULL == persist_dir) {
nlog_error("fail to strdup: %s", dir_name);
goto error;
}
rv = create_dir(persist_dir);
if (rv != 0) {
nlog_error("failed to create directory: %s", persist_dir);
goto error;
}
int n = path_cat(path, dir_len, PATH_MAX_SIZE, "plugins.json");
if (PATH_MAX_SIZE <= n) {
nlog_error("path too long: %s", path);
goto error;
}
plugins_fname = strdup(path);
if (NULL == plugins_fname) {
nlog_error("fail to strdup: %s", path);
goto error;
}
if (0 != ensure_file_exist(plugins_fname, "{\"plugins\": []}")) {
nlog_error("persister failed to ensure file exist: %s", plugins_fname);
goto error;
}
neu_persister_t *persister = malloc(sizeof(neu_persister_t));
if (NULL == persister) {
nlog_error("failed to alloc memory for persister struct");
goto error;
}
persister->plugins_fname = plugins_fname;
persister->persist_dir = persist_dir;
n = path_cat(path, dir_len, PATH_MAX_SIZE, "sqlite.db");
if (PATH_MAX_SIZE <= n) {
nlog_error("path too long: %s", path);
goto error;
}
rv = sqlite3_open(path, &persister->db);
if (SQLITE_OK != rv) {
nlog_error("db `%s` fail: %s", path, sqlite3_errstr(rv));
goto error;
nlog_fatal("db `%s` fail: %s", db_file, sqlite3_errstr(rv));
return -1;
}
sqlite3_busy_timeout(persister->db, 100 * 1000);
// we rely on sqlite foreign key support
if (SQLITE_OK !=
sqlite3_exec(persister->db, "PRAGMA foreign_keys=ON", NULL, NULL,
NULL)) {
nlog_error("db foreign key support fail: %s",
sqlite3_errmsg(persister->db));
sqlite3_close(persister->db);
goto error;
}
if (SQLITE_OK !=
sqlite3_exec(persister->db, "PRAGMA journal_mode=WAL", NULL, NULL,
NULL)) {
nlog_error("db journal_mode WAL fail: %s",
sqlite3_errmsg(persister->db));
sqlite3_close(persister->db);
goto error;
}
if (0 != apply_schemas(persister->db, g_config_dir)) {
goto error;
sqlite3_busy_timeout(global_db, 100 * 1000);
rv = sqlite3_exec(global_db, "PRAGMA foreign_keys=ON", NULL, NULL, NULL);
if (rv != SQLITE_OK) {
nlog_fatal("db foreign key support fail: %s",
sqlite3_errmsg(global_db));
sqlite3_close(global_db);
return -1;
}
free(path);
return persister;
rv = sqlite3_exec(global_db, "PRAGMA journal_mode=WAL", NULL, NULL, NULL);
if (rv != SQLITE_OK) {
nlog_fatal("db journal_mode WAL fail: %s", sqlite3_errmsg(global_db));
sqlite3_close(global_db);
return -1;
}
error:
free(path);
free(plugins_fname);
free(persist_dir);
return NULL;
rv = apply_schemas(global_db, g_config_dir);
if (rv != 0) {
nlog_fatal("db apply schemas fail");
sqlite3_close(global_db);
return -1;
}
return 0;
}
static inline int persister_adapters_dir(char *buf, size_t size,
neu_persister_t *persister)
void neu_persister_destroy()
{
size_t n = path_cat(buf, 0, size, persister->persist_dir);
if (size == n) {
return n;
}
n = path_cat(buf, n, size, "adapters");
return n;
sqlite3_close(global_db);
}
static inline int persister_adapter_dir(char *buf, size_t size,
neu_persister_t *persister,
const char * adapter_name)
int neu_persister_store_node(neu_persist_node_info_t *info)
{
size_t n = persister_adapters_dir(buf, size, persister);
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)
{
sqlite3_close(persister->db);
free((char *) persister->plugins_fname);
free((char *) persister->persist_dir);
free(persister);
}
int neu_persister_store_node(neu_persister_t * persister,
neu_persist_node_info_t *info)
{
return execute_sql(persister->db,
return execute_sql(global_db,
"INSERT INTO nodes (name, type, state, plugin_name) "
"VALUES (%Q, %i, %i, %Q)",
info->name, info->type, info->state, info->plugin_name);
@@ -727,7 +485,7 @@ static UT_icd node_info_icd = {
(dtor_f *) neu_persist_node_info_fini,
};
int neu_persister_load_nodes(neu_persister_t *persister, UT_array **node_infos)
int neu_persister_load_nodes(UT_array **node_infos)
{
int rv = 0;
sqlite3_stmt *stmt = NULL;
@@ -735,10 +493,8 @@ int neu_persister_load_nodes(neu_persister_t *persister, UT_array **node_infos)
utarray_new(*node_infos, &node_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));
if (SQLITE_OK != sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL)) {
nlog_error("prepare `%s` fail: %s", query, sqlite3_errmsg(global_db));
utarray_free(*node_infos);
*node_infos = NULL;
return NEU_ERR_EINTERNAL;
@@ -768,7 +524,7 @@ int neu_persister_load_nodes(neu_persister_t *persister, UT_array **node_infos)
}
if (SQLITE_DONE != step) {
nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(persister->db));
nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(global_db));
// do not set return code, return partial or empty result
}
@@ -776,31 +532,27 @@ int neu_persister_load_nodes(neu_persister_t *persister, UT_array **node_infos)
return rv;
}
int neu_persister_delete_node(neu_persister_t *persister, const char *node_name)
int neu_persister_delete_node(const char *node_name)
{
// rely on foreign key constraints to remove settings, groups, tags and
// subscriptions
return execute_sql(persister->db, "DELETE FROM nodes WHERE name=%Q;",
return execute_sql(global_db, "DELETE FROM nodes WHERE name=%Q;",
node_name);
}
int neu_persister_update_node(neu_persister_t *persister, const char *node_name,
const char *new_name)
int neu_persister_update_node(const char *node_name, const char *new_name)
{
return execute_sql(persister->db, "UPDATE nodes SET name=%Q WHERE name=%Q;",
return execute_sql(global_db, "UPDATE nodes SET name=%Q WHERE name=%Q;",
new_name, node_name);
}
int neu_persister_update_node_state(neu_persister_t *persister,
const char *node_name, int state)
int neu_persister_update_node_state(const char *node_name, int state)
{
return execute_sql(persister->db,
"UPDATE nodes SET state=%i WHERE name=%Q;", state,
node_name);
return execute_sql(global_db, "UPDATE nodes SET state=%i WHERE name=%Q;",
state, node_name);
}
int neu_persister_store_plugins(neu_persister_t *persister,
UT_array * plugin_infos)
int neu_persister_store_plugins(UT_array *plugin_infos)
{
int index = 0;
neu_json_plugin_resp_t plugin_resp = {
@@ -828,7 +580,7 @@ int neu_persister_store_plugins(neu_persister_t *persister,
return rv;
}
rv = write_file_string(persister->plugins_fname, result);
rv = write_file_string(plugin_file, result);
free(result);
return rv;
@@ -865,8 +617,7 @@ static int ut_str_cmp(const void *a, const void *b)
return strcmp(*(char **) a, *(char **) b);
}
int neu_persister_load_plugins(neu_persister_t *persister,
UT_array ** plugin_infos)
int neu_persister_load_plugins(UT_array **plugin_infos)
{
UT_array *default_plugins = NULL;
UT_array *user_plugins = NULL;
@@ -879,7 +630,7 @@ int neu_persister_load_plugins(neu_persister_t *persister,
nlog_warn("cannot load default plugins");
}
// user plugins
if (0 != load_plugins_file(persister->plugins_fname, user_plugins)) {
if (0 != load_plugins_file(plugin_file, user_plugins)) {
nlog_warn("cannot load user plugins");
} else {
// the following operation needs sorting
@@ -903,10 +654,10 @@ int neu_persister_load_plugins(neu_persister_t *persister,
return 0;
}
int neu_persister_store_tag(neu_persister_t *persister, const char *driver_name,
const char *group_name, const neu_datatag_t *tag)
int neu_persister_store_tag(const char *driver_name, const char *group_name,
const neu_datatag_t *tag)
{
return execute_sql(persister->db,
return execute_sql(global_db,
"INSERT INTO tags (driver_name, group_name, name, "
"address, attribute, precision, "
"type, decimal, description) VALUES(%Q, %Q, %Q, %Q, %i, "
@@ -916,8 +667,7 @@ int neu_persister_store_tag(neu_persister_t *persister, const char *driver_name,
tag->description);
}
int neu_persister_store_tags(neu_persister_t *persister,
const char *driver_name, const char *group_name,
int neu_persister_store_tags(const char *driver_name, const char *group_name,
const neu_datatag_t *tags, size_t n)
{
sqlite3_stmt *stmt = NULL;
@@ -925,15 +675,13 @@ int neu_persister_store_tags(neu_persister_t *persister,
"address, attribute, precision, type, decimal, "
"description) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)";
if (SQLITE_OK != sqlite3_exec(persister->db, "BEGIN", NULL, NULL, NULL)) {
nlog_error("begin transaction fail: %s", sqlite3_errmsg(persister->db));
if (SQLITE_OK != sqlite3_exec(global_db, "BEGIN", NULL, NULL, NULL)) {
nlog_error("begin transaction fail: %s", sqlite3_errmsg(global_db));
return NEU_ERR_EINTERNAL;
}
if (SQLITE_OK !=
sqlite3_prepare_v2(persister->db, query, -1, &stmt, NULL)) {
nlog_error("prepare `%s` fail: %s", query,
sqlite3_errmsg(persister->db));
if (SQLITE_OK != sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL)) {
nlog_error("prepare `%s` fail: %s", query, sqlite3_errmsg(global_db));
goto error;
}
@@ -944,68 +692,67 @@ int neu_persister_store_tags(neu_persister_t *persister,
if (SQLITE_OK != sqlite3_bind_text(stmt, 1, driver_name, -1, NULL)) {
nlog_error("bind `%s` with driver_name=`%s` fail: %s", query,
driver_name, sqlite3_errmsg(persister->db));
driver_name, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 2, group_name, -1, NULL)) {
nlog_error("bind `%s` with group_name=`%s` fail: %s", query,
group_name, sqlite3_errmsg(persister->db));
group_name, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 3, tag->name, -1, NULL)) {
nlog_error("bind `%s` with name=`%s` fail: %s", query, tag->name,
sqlite3_errmsg(persister->db));
sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 4, tag->address, -1, NULL)) {
nlog_error("bind `%s` with address=`%s` fail: %s", query,
tag->address, sqlite3_errmsg(persister->db));
tag->address, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_int(stmt, 5, tag->attribute)) {
nlog_error("bind `%s` with attribute=`%i` fail: %s", query,
tag->attribute, sqlite3_errmsg(persister->db));
tag->attribute, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_int(stmt, 6, tag->precision)) {
nlog_error("bind `%s` with precision=`%i` fail: %s", query,
tag->precision, sqlite3_errmsg(persister->db));
tag->precision, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_int(stmt, 7, tag->type)) {
nlog_error("bind `%s` with type=`%i` fail: %s", query, tag->type,
sqlite3_errmsg(persister->db));
sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_double(stmt, 8, tag->decimal)) {
nlog_error("bind `%s` with decimal=`%f` fail: %s", query,
tag->decimal, sqlite3_errmsg(persister->db));
tag->decimal, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK !=
sqlite3_bind_text(stmt, 9, tag->description, -1, NULL)) {
nlog_error("bind `%s` with description=`%s` fail: %s", query,
tag->description, sqlite3_errmsg(persister->db));
tag->description, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_DONE != sqlite3_step(stmt)) {
nlog_error("sqlite3_step fail: %s", sqlite3_errmsg(persister->db));
nlog_error("sqlite3_step fail: %s", sqlite3_errmsg(global_db));
goto error;
}
}
if (SQLITE_OK != sqlite3_exec(persister->db, "COMMIT", NULL, NULL, NULL)) {
nlog_error("commit transaction fail: %s",
sqlite3_errmsg(persister->db));
if (SQLITE_OK != sqlite3_exec(global_db, "COMMIT", NULL, NULL, NULL)) {
nlog_error("commit transaction fail: %s", sqlite3_errmsg(global_db));
goto error;
}
@@ -1014,13 +761,13 @@ int neu_persister_store_tags(neu_persister_t *persister,
error:
nlog_info("rollback transaction");
sqlite3_exec(persister->db, "ROLLBACK", NULL, NULL, NULL);
sqlite3_exec(global_db, "ROLLBACK", NULL, NULL, NULL);
sqlite3_finalize(stmt);
return NEU_ERR_EINTERNAL;
}
int neu_persister_load_tags(neu_persister_t *persister, const char *driver_name,
const char *group_name, UT_array **tags)
int neu_persister_load_tags(const char *driver_name, const char *group_name,
UT_array **tags)
{
sqlite3_stmt *stmt = NULL;
const char * query = "SELECT name, address, attribute, precision, type, "
@@ -1029,22 +776,20 @@ int neu_persister_load_tags(neu_persister_t *persister, const char *driver_name,
utarray_new(*tags, neu_tag_get_icd());
if (SQLITE_OK !=
sqlite3_prepare_v2(persister->db, query, -1, &stmt, NULL)) {
nlog_error("prepare `%s` fail: %s", query,
sqlite3_errmsg(persister->db));
if (SQLITE_OK != sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL)) {
nlog_error("prepare `%s` fail: %s", query, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 1, driver_name, -1, NULL)) {
nlog_error("bind `%s` with `%s` fail: %s", query, driver_name,
sqlite3_errmsg(persister->db));
sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 2, group_name, -1, NULL)) {
nlog_error("bind `%s` with `%s` fail: %s", query, group_name,
sqlite3_errmsg(persister->db));
sqlite3_errmsg(global_db));
goto error;
}
@@ -1065,7 +810,7 @@ int neu_persister_load_tags(neu_persister_t *persister, const char *driver_name,
}
if (SQLITE_DONE != step) {
nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(persister->db));
nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(global_db));
// do not set return code, return partial or empty result
}
@@ -1078,11 +823,10 @@ error:
return NEU_ERR_EINTERNAL;
}
int neu_persister_update_tag(neu_persister_t *persister,
const char *driver_name, const char *group_name,
int neu_persister_update_tag(const char *driver_name, const char *group_name,
const neu_datatag_t *tag)
{
return execute_sql(persister->db,
return execute_sql(global_db,
"UPDATE tags SET address=%Q, attribute=%i, "
"precision=%i, type=%i, decimal=%lf, description=%Q "
"WHERE driver_name=%Q AND group_name=%Q AND name=%Q",
@@ -1091,23 +835,21 @@ int neu_persister_update_tag(neu_persister_t *persister,
tag->name);
}
int neu_persister_delete_tag(neu_persister_t *persister,
const char *driver_name, const char *group_name,
int neu_persister_delete_tag(const char *driver_name, const char *group_name,
const char *tag_name)
{
return execute_sql(
persister->db,
global_db,
"DELETE FROM tags WHERE driver_name=%Q AND group_name=%Q AND name=%Q",
driver_name, group_name, tag_name);
}
int neu_persister_store_subscription(neu_persister_t *persister,
const char * app_name,
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)
{
return execute_sql(
persister->db,
global_db,
"INSERT INTO subscriptions (app_name, driver_name, group_name) "
"VALUES (%Q, %Q, %Q)",
app_name, driver_name, group_name);
@@ -1120,9 +862,8 @@ static UT_icd subscription_info_icd = {
(dtor_f *) neu_persist_subscription_info_fini,
};
int neu_persister_load_subscriptions(neu_persister_t *persister,
const char * app_name,
UT_array ** subscription_infos)
int neu_persister_load_subscriptions(const char *app_name,
UT_array ** subscription_infos)
{
sqlite3_stmt *stmt = NULL;
const char * query =
@@ -1130,16 +871,14 @@ int neu_persister_load_subscriptions(neu_persister_t *persister,
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));
if (SQLITE_OK != sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL)) {
nlog_error("prepare `%s` fail: %s", query, sqlite3_errmsg(global_db));
goto error;
}
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));
sqlite3_errmsg(global_db));
goto error;
}
@@ -1166,7 +905,7 @@ int neu_persister_load_subscriptions(neu_persister_t *persister,
}
if (SQLITE_DONE != step) {
nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(persister->db));
nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(global_db));
// do not set return code, return partial or empty result
}
@@ -1179,32 +918,29 @@ error:
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)
int neu_persister_delete_subscription(const char *app_name,
const char *driver_name,
const char *group_name)
{
return execute_sql(persister->db,
return execute_sql(global_db,
"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,
const char * driver_name,
int neu_persister_store_group(const char * driver_name,
neu_persist_group_info_t *group_info)
{
return execute_sql(
persister->db,
global_db,
"INSERT INTO groups (driver_name, name, interval) VALUES (%Q, %Q, %u)",
driver_name, group_info->name, (unsigned) group_info->interval);
}
int neu_persister_update_group(neu_persister_t * persister,
const char * driver_name,
int neu_persister_update_group(const char * driver_name,
neu_persist_group_info_t *group_info)
{
return execute_sql(persister->db,
return execute_sql(global_db,
"UPDATE groups SET driver_name=%Q, name=%Q, "
"interval=%i WHERE driver_name=%Q AND name=%Q",
driver_name, group_info->name, group_info->interval,
@@ -1218,24 +954,21 @@ static UT_icd group_info_icd = {
(dtor_f *) neu_persist_group_info_fini,
};
int neu_persister_load_groups(neu_persister_t *persister,
const char *driver_name, UT_array **group_infos)
int neu_persister_load_groups(const char *driver_name, UT_array **group_infos)
{
sqlite3_stmt *stmt = NULL;
const char *query = "SELECT name, interval FROM groups WHERE driver_name=?";
utarray_new(*group_infos, &group_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));
if (SQLITE_OK != sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL)) {
nlog_error("prepare `%s` fail: %s", query, sqlite3_errmsg(global_db));
goto error;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 1, driver_name, -1, NULL)) {
nlog_error("bind `%s` with `%s` fail: %s", query, driver_name,
sqlite3_errmsg(persister->db));
sqlite3_errmsg(global_db));
goto error;
}
@@ -1255,7 +988,7 @@ int neu_persister_load_groups(neu_persister_t *persister,
}
if (SQLITE_DONE != step) {
nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(persister->db));
nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(global_db));
// do not set return code, return partial or empty result
}
@@ -1268,49 +1001,45 @@ error:
return NEU_ERR_EINTERNAL;
}
int neu_persister_delete_group(neu_persister_t *persister,
const char *driver_name, const char *group_name)
int neu_persister_delete_group(const char *driver_name, const char *group_name)
{
// rely on foreign key constraints to delete tags and subscriptions
return execute_sql(persister->db,
return execute_sql(global_db,
"DELETE FROM groups WHERE driver_name=%Q AND name=%Q",
driver_name, group_name);
}
int neu_persister_store_node_setting(neu_persister_t *persister,
const char *node_name, const char *setting)
int neu_persister_store_node_setting(const char *node_name, const char *setting)
{
return execute_sql(
persister->db,
global_db,
"INSERT OR REPLACE INTO settings (node_name, setting) VALUES (%Q, %Q)",
node_name, setting);
}
int neu_persister_load_node_setting(neu_persister_t * persister,
const char * node_name,
int neu_persister_load_node_setting(const char * node_name,
const char **const setting)
{
int rv = 0;
sqlite3_stmt *stmt = NULL;
const char * query = "SELECT setting FROM settings WHERE node_name=?";
if (SQLITE_OK !=
sqlite3_prepare_v2(persister->db, query, -1, &stmt, NULL)) {
if (SQLITE_OK != sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL)) {
nlog_error("prepare `%s` with `%s` fail: %s", query, node_name,
sqlite3_errmsg(persister->db));
sqlite3_errmsg(global_db));
return NEU_ERR_EINTERNAL;
}
if (SQLITE_OK != sqlite3_bind_text(stmt, 1, node_name, -1, NULL)) {
nlog_error("bind `%s` with `%s` fail: %s", query, node_name,
sqlite3_errmsg(persister->db));
sqlite3_errmsg(global_db));
rv = NEU_ERR_EINTERNAL;
goto end;
}
if (SQLITE_ROW != sqlite3_step(stmt)) {
nlog_warn("SQL `%s` with `%s` fail: %s", query, node_name,
sqlite3_errmsg(persister->db));
sqlite3_errmsg(global_db));
rv = NEU_ERR_EINTERNAL;
goto end;
}
@@ -1329,9 +1058,8 @@ end:
return rv;
}
int neu_persister_delete_node_setting(neu_persister_t *persister,
const char * node_name)
int neu_persister_delete_node_setting(const char *node_name)
{
return execute_sql(persister->db, "DELETE FROM settings WHERE node_name=%Q",
return execute_sql(global_db, "DELETE FROM settings WHERE node_name=%Q",
node_name);
}
src/persist/persist.h
+33 -81
View File
@@ -69,245 +69,197 @@ neu_persist_subscription_info_fini(neu_persist_subscription_info_t *info)
free(info->group_name);
}
/**
* Persister, provide methods to persist data */
typedef struct neu_persister neu_persister_t;
const char *neu_persister_get_persist_dir(neu_persister_t *persister);
const char *neu_persister_get_plugins_fname(neu_persister_t *persister);
/**
* Create persister.
* @return Pointer to heap allocated neu_persister_t on success, NULL otherwise.
* @return 0 on success, -1 otherwise.
*/
neu_persister_t *neu_persister_create(const char *dir_name);
int neu_persister_create();
/**
* Destroy perister.
* @param persister persiter object to destroy.
*/
void neu_persister_destroy(neu_persister_t *persiter);
void neu_persister_destroy();
/**
* Persist nodes.
* @param persister persiter object.
* @param node_info neu_persist_node_info_t.
* @return 0 on success, non-zero on failure
*/
int neu_persister_store_node(neu_persister_t * persister,
neu_persist_node_info_t *info);
int neu_persister_store_node(neu_persist_node_info_t *info);
/**
* Load node infos.
* @param persister persiter object.
* @param[out] node_infos used to return pointer to heap allocated
* vector of neu_persist_node_info_t.
* @return 0 on success, none-zero on failure
*/
int neu_persister_load_nodes(neu_persister_t *persister, UT_array **node_infos);
int neu_persister_load_nodes(UT_array **node_infos);
/**
* Delete node.
* @param persister persiter object.
* @param node_name name of the node to delete.
* @return 0 on success, none-zero on failure
*/
int neu_persister_delete_node(neu_persister_t *persister,
const char * node_name);
int neu_persister_delete_node(const char *node_name);
/**
* Update node name.
* @param persister persiter object.
* @param node_name name of the node to update.
* @param new_name new name of the adapter.
* @return 0 on success, none-zero on failure
*/
int neu_persister_update_node(neu_persister_t *persister, const char *node_name,
const char *new_name);
int neu_persister_update_node(const char *node_name, const char *new_name);
/**
* Update node state.
* @param persister persiter object.
* @param node_name name of the node to update.
* @param state state of the adapter.
* @return 0 on success, none-zero on failure
*/
int neu_persister_update_node_state(neu_persister_t *persister,
const char *node_name, int state);
int neu_persister_update_node_state(const char *node_name, int state);
/**
* Persist plugins.
* @param persister persiter object.
* @param plugin_infos vector of neu_persist_plugin_info_t.
* @return 0 on success, non-zero otherwise
*/
int neu_persister_store_plugins(neu_persister_t *persister,
UT_array * plugin_infos);
int neu_persister_store_plugins(UT_array *plugin_infos);
/**
* Load plugin infos.
* @param persister persiter object.
* @param[out] plugin_infos used to return pointer to heap allocated
* vector of neu_persist_plugin_info_t.
* @return 0 on success, none-zero on failure
*/
int neu_persister_load_plugins(neu_persister_t *persister,
UT_array ** plugin_infos);
int neu_persister_load_plugins(UT_array **plugin_infos);
/**
* Persist node tags.
* @param persister persiter object.
* @param driver_name name of the driver who owns the tags
* @param group_name name of the group
* @param tag the tag to store
* @return 0 on success, non-zero otherwise
*/
int neu_persister_store_tag(neu_persister_t *persister, const char *driver_name,
const char *group_name, const neu_datatag_t *tag);
int neu_persister_store_tag(const char *driver_name, const char *group_name,
const neu_datatag_t *tag);
int neu_persister_store_tags(neu_persister_t *persister,
const char *driver_name, const char *group_name,
int neu_persister_store_tags(const char *driver_name, const char *group_name,
const neu_datatag_t *tags, size_t n);
/**
* Load node tag infos.
* @param persister persiter object.
* @param node_name name of the node who owns the tags
* @param group_name name of the group
* @param[out] tag_infos used to return pointer to heap allocated
* vector of neu_datatag_t
* @return 0 on success, non-zero otherwise
*/
int neu_persister_load_tags(neu_persister_t *persister, const char *driver_name,
const char *group_name, UT_array **tag_infos);
int neu_persister_load_tags(const char *driver_name, const char *group_name,
UT_array **tag_infos);
/**
* Update node tags.
* @param persister persiter object.
* @param driver_name name of the driver who owns the tags
* @param group_name name of the group
* @param tag the tag to update
* @return 0 on success, non-zero otherwise
*/
int neu_persister_update_tag(neu_persister_t *persister,
const char *driver_name, const char *group_name,
int neu_persister_update_tag(const char *driver_name, const char *group_name,
const neu_datatag_t *tag);
/**
* Delete node tags.
* @param persister persiter object.
* @param driver_name name of the driver who owns the tags
* @param group_name name of the group
* @param tag_name name of the tag
* @return 0 on success, non-zero otherwise
*/
int neu_persister_delete_tag(neu_persister_t *persister,
const char *driver_name, const char *group_name,
int neu_persister_delete_tag(const char *driver_name, const char *group_name,
const char *tag_name);
/**
* 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_store_subscription(neu_persister_t *persister,
const char * app_name,
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);
/**
* Load adapter subscriptions.
* @param persister persiter object.
* @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.
* @return 0 on success, non-zero otherwise
*/
int neu_persister_load_subscriptions(neu_persister_t *persister,
const char * app_name,
UT_array ** subscription_infos);
int neu_persister_load_subscriptions(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);
int neu_persister_delete_subscription(const char *app_name,
const char *driver_name,
const char *group_name);
/**
* Persist group config.
* @param persister persiter object.
* @param driver_name name of the driver who owns the group
* @param group_info group info to persist.
* @return 0 on success, non-zero otherwise
*/
int neu_persister_store_group(neu_persister_t * persister,
const char * driver_name,
int neu_persister_store_group(const char * driver_name,
neu_persist_group_info_t *group_info);
/**
* Update group config.
* @param persister persiter object.
* @param driver_name name of the driver who owns the group
* @param group_info group info to persist.
* @return 0 on success, non-zero otherwise
*/
int neu_persister_update_group(neu_persister_t * persister,
const char * driver_name,
int neu_persister_update_group(const char * driver_name,
neu_persist_group_info_t *group_info);
/**
* Load all group config infos under an adapter.
* @param persister persiter object.
* @param driver_name name of the driver who owns the group
* @param[out] group_infos used to return pointer to heap allocated
* vector of neu_persist_group_info_t.
* @return 0 on success, non-zero otherwise
*/
int neu_persister_load_groups(neu_persister_t *persister,
const char *driver_name, UT_array **group_infos);
int neu_persister_load_groups(const char *driver_name, UT_array **group_infos);
/**
* Delete group config.
* @param persister persiter object.
* @param driver_name name of the driver who owns the group
* @param group_name name of the group.
* @return 0 on success, none-zero on failure
*/
int neu_persister_delete_group(neu_persister_t *persister,
const char *driver_name, const char *group_name);
int neu_persister_delete_group(const char *driver_name, const char *group_name);
/**
* Persist node setting.
* @param persister persiter object.
* @param adapter_name name of the adapter who owns the setting.
* @param setting node setting string.
* @return 0 on success, non-zero otherwise
*/
int neu_persister_store_node_setting(neu_persister_t *persister,
const char * node_name,
const char * setting);
int neu_persister_store_node_setting(const char *node_name,
const char *setting);
/**
* Load node setting.
* @param persister persiter object.
* @param node_name name of the node.
* @param[out] setting used to return node setting string.
* @return 0 on success, non-zero otherwise
*/
int neu_persister_load_node_setting(neu_persister_t * persister,
const char * node_name,
int neu_persister_load_node_setting(const char * node_name,
const char **const setting);
/**
* Delete node setting.
* @param persister persiter object.
* @param node_name name of the node.
* @return 0 on success, none-zero on failure
*/
int neu_persister_delete_node_setting(neu_persister_t *persister,
const char * node_name);
int neu_persister_delete_node_setting(const char *node_name);
#ifdef __cplusplus
}