mirror of
https://github.com/emqx/neuron.git
synced 2026-10-25 11:26:32 +08:00
Merge pull request #646 from eeff/eeff/persist
Migrate persistence to sqlite
This commit is contained in:
+1
-1
@@ -121,7 +121,7 @@ set(CMAKE_BUILD_RPATH ./)
|
||||
add_executable(neuron)
|
||||
target_sources(neuron PRIVATE ${NEURON_SOURCES})
|
||||
target_include_directories(neuron PRIVATE include/neuron src plugins)
|
||||
target_link_libraries(neuron dl neuron-base)
|
||||
target_link_libraries(neuron dl neuron-base sqlite3 -lm)
|
||||
|
||||
#copy file for run
|
||||
file(COPY ${CMAKE_SOURCE_DIR}/zlog.conf DESTINATION ${CMAKE_BINARY_DIR}/config)
|
||||
|
||||
+13
-10
@@ -6,6 +6,12 @@ import shutil
|
||||
import time
|
||||
|
||||
|
||||
def prepare_persistence_dir():
|
||||
os.system("mkdir -p build/persistence")
|
||||
os.system("cp -r persistence build/")
|
||||
os.system("cp default_plugins.json build/persistence/plugins.json")
|
||||
|
||||
|
||||
class Neuron(object):
|
||||
process = 0
|
||||
profiler_process = None
|
||||
@@ -14,8 +20,7 @@ class Neuron(object):
|
||||
pass
|
||||
|
||||
def Start_Neuron(self):
|
||||
os.system("mkdir -p build/persistence")
|
||||
os.system("cp default_plugins.json build/persistence/plugins.json")
|
||||
prepare_persistence_dir()
|
||||
self.process = subprocess.Popen(['./neuron', '--log'], cwd='build/')
|
||||
return self.process
|
||||
|
||||
@@ -25,17 +30,15 @@ class Neuron(object):
|
||||
time.sleep(5)
|
||||
self.process.kill()
|
||||
if remove_persistence_data:
|
||||
shutil.rmtree("build/persistence", ignore_errors=True)
|
||||
os.system("mkdir -p build/persistence")
|
||||
os.system("cp default_plugins.json build/persistence/plugins.json")
|
||||
os.system("rm build/persistence/sqlite.db")
|
||||
prepare_persistence_dir()
|
||||
|
||||
def End_Neuron(self, process):
|
||||
process.kill()
|
||||
|
||||
def Remove_Persistence(self):
|
||||
shutil.rmtree("build/persistence", ignore_errors=True)
|
||||
os.system("mkdir -p build/persistence")
|
||||
os.system("cp default_plugins.json build/persistence/plugins.json")
|
||||
os.system("rm build/persistence/sqlite.db")
|
||||
prepare_persistence_dir()
|
||||
|
||||
def Profile_Neuron(self, interval, result_dir):
|
||||
cmd = [
|
||||
@@ -151,9 +154,9 @@ class Tag(object):
|
||||
def Tag_Find_By_Name(self, tags, name):
|
||||
for tag in tags:
|
||||
if tag['name'] == name:
|
||||
return tag['id']
|
||||
return tag
|
||||
|
||||
return -1
|
||||
return None
|
||||
|
||||
|
||||
class Subscribe(object):
|
||||
|
||||
+10
-8
@@ -237,15 +237,17 @@ Get the update tags after stopping ande restarting Neuron, it should return the
|
||||
${len} = Get Length ${res}[tags]
|
||||
Should Be Equal As Integers ${len} 2
|
||||
|
||||
Should Be Equal As Strings ${res}[tags][1][name] tag1
|
||||
Should Be Equal As Strings ${res}[tags][1][address] 1!400002
|
||||
Should Be Equal As Strings ${res}[tags][1][attribute] ${TAG_ATTRIBUTE_READ}
|
||||
Should Be Equal As Strings ${res}[tags][1][type] ${TAG_TYPE_INT16}
|
||||
${tag1} = Tag Find By Name ${res}[tags] tag1
|
||||
Should Be Equal As Strings ${tag1}[name] tag1
|
||||
Should Be Equal As Strings ${tag1}[address] 1!400002
|
||||
Should Be Equal As Strings ${tag1}[attribute] ${TAG_ATTRIBUTE_READ}
|
||||
Should Be Equal As Strings ${tag1}[type] ${TAG_TYPE_INT16}
|
||||
|
||||
Should Be Equal As Strings ${res}[tags][0][name] tag2
|
||||
Should Be Equal As Strings ${res}[tags][0][address] 1!00001
|
||||
Should Be Equal As Strings ${res}[tags][0][attribute] ${TAG_ATTRIBUTE_RW}
|
||||
Should Be Equal As Strings ${res}[tags][0][type] ${TAG_TYPE_BIT}
|
||||
${tag2} = Tag Find By Name ${res}[tags] tag2
|
||||
Should Be Equal As Strings ${tag2}[name] tag2
|
||||
Should Be Equal As Strings ${tag2}[address] 1!00001
|
||||
Should Be Equal As Strings ${tag2}[attribute] ${TAG_ATTRIBUTE_RW}
|
||||
Should Be Equal As Strings ${tag2}[type] ${TAG_TYPE_BIT}
|
||||
|
||||
[Teardown] Neuron Stop Running ${process}
|
||||
|
||||
|
||||
@@ -168,16 +168,6 @@ Del Tags
|
||||
|
||||
[RETURN] ${res}[0]
|
||||
|
||||
Get Tag ID
|
||||
[Arguments] ${node_id} ${group} ${tag_name}
|
||||
${res} = Get Tags ${node_id} ${group}
|
||||
Check Response Status ${res} 200
|
||||
|
||||
${id} = Tag Find By Name ${res}[tags] ${tag_name}
|
||||
Should Not Be Equal As Integers ${id} 0
|
||||
|
||||
[Return] ${id}
|
||||
|
||||
Read Tags
|
||||
[Arguments] ${node} ${group}
|
||||
POST /api/v2/read {"node": "${node}", "group": "${group}"}
|
||||
|
||||
+3
-1
@@ -83,7 +83,9 @@ function build_sqlite3() {
|
||||
tar xzf sqlite3.tar.gz --strip-components=1 -C sqlite3
|
||||
cd sqlite3
|
||||
./configure --prefix=${install_dir} \
|
||||
&& make CC=${compiler_prefix}-gcc -j4 \
|
||||
--disable-shared --disable-readline \
|
||||
--host ${arch} CC=${compiler_prefix}-gcc \
|
||||
&& make -j4 \
|
||||
&& sudo make install
|
||||
cd ../
|
||||
}
|
||||
|
||||
+12
-17
@@ -466,8 +466,12 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
|
||||
error.error = NEU_ERR_GROUP_NOT_ALLOW;
|
||||
} else {
|
||||
for (int i = 0; i < cmd->n_tag; i++) {
|
||||
neu_adapter_driver_del_tag((neu_adapter_driver_t *) adapter,
|
||||
cmd->group, cmd->tags[i]);
|
||||
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]);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -476,11 +480,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
|
||||
}
|
||||
free(cmd->tags);
|
||||
|
||||
if (error.error == NEU_ERR_SUCCESS) {
|
||||
adapter_storage_tag(adapter->persister, adapter, cmd->driver,
|
||||
cmd->group);
|
||||
}
|
||||
|
||||
neu_msg_exchange(header);
|
||||
header->type = NEU_RESP_ERROR;
|
||||
reply(adapter, header, &error);
|
||||
@@ -498,7 +497,10 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
|
||||
neu_adapter_driver_add_tag((neu_adapter_driver_t *) adapter,
|
||||
cmd->group, &cmd->tags[i]);
|
||||
if (ret == 0) {
|
||||
adapter_storage_add_tag(adapter->persister, cmd->driver,
|
||||
cmd->group, &cmd->tags[i]);
|
||||
resp.index += 1;
|
||||
|
||||
} else {
|
||||
resp.error = ret;
|
||||
break;
|
||||
@@ -512,11 +514,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
|
||||
}
|
||||
free(cmd->tags);
|
||||
|
||||
if (resp.index > 0) {
|
||||
adapter_storage_tag(adapter->persister, adapter, cmd->driver,
|
||||
cmd->group);
|
||||
}
|
||||
|
||||
neu_msg_exchange(header);
|
||||
header->type = NEU_RESP_ADD_TAG;
|
||||
reply(adapter, header, &resp);
|
||||
@@ -534,6 +531,9 @@ 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]);
|
||||
|
||||
resp.index += 1;
|
||||
} else {
|
||||
resp.error = ret;
|
||||
@@ -548,11 +548,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
|
||||
}
|
||||
free(cmd->tags);
|
||||
|
||||
if (resp.index > 0) {
|
||||
adapter_storage_tag(adapter->persister, adapter, cmd->driver,
|
||||
cmd->group);
|
||||
}
|
||||
|
||||
neu_msg_exchange(header);
|
||||
header->type = NEU_RESP_UPDATE_TAG;
|
||||
reply(adapter, header, &resp);
|
||||
|
||||
+39
-35
@@ -26,52 +26,60 @@
|
||||
void adapter_storage_setting(neu_persister_t *persister, const char *node,
|
||||
const char *setting)
|
||||
{
|
||||
neu_persister_store_adapter_setting(persister, node, setting);
|
||||
neu_persister_store_node_setting(persister, node, setting);
|
||||
}
|
||||
|
||||
void adapter_storage_add_group(neu_persister_t *persister, const char *node,
|
||||
const char *group, uint32_t interval)
|
||||
{
|
||||
neu_persist_group_config_info_t info = {
|
||||
.group_config_name = (char *) group,
|
||||
.read_interval = interval,
|
||||
.adapter_name = (char *) node,
|
||||
neu_persist_group_info_t info = {
|
||||
.name = (char *) group,
|
||||
.interval = interval,
|
||||
};
|
||||
|
||||
neu_persister_store_group_config(persister, node, &info);
|
||||
neu_persister_store_group(persister, node, &info);
|
||||
}
|
||||
|
||||
void adapter_storage_del_group(neu_persister_t *persister, const char *node,
|
||||
const char *group)
|
||||
{
|
||||
neu_persister_delete_group_config(persister, node, group);
|
||||
neu_persister_delete_group(persister, node, group);
|
||||
}
|
||||
|
||||
void adapter_storage_tag(neu_persister_t *persister, neu_adapter_t *adapter,
|
||||
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)
|
||||
{
|
||||
UT_array *tags = NULL;
|
||||
|
||||
int rv = neu_adapter_driver_get_tag((neu_adapter_driver_t *) adapter, group,
|
||||
&tags);
|
||||
if (rv != 0) {
|
||||
nlog_warn("fail get tag infos");
|
||||
return;
|
||||
}
|
||||
|
||||
rv = neu_persister_store_datatags(persister, node, group, tags);
|
||||
int rv = neu_persister_store_tag(persister, node, group, tag);
|
||||
if (0 != rv) {
|
||||
nlog_error("fail store adapter:%s grp:%s datatag infos", node, group);
|
||||
nlog_error("fail store tag:%s adapter:%s grp:%s", tag->name, node,
|
||||
group);
|
||||
}
|
||||
}
|
||||
|
||||
neu_persist_datatag_infos_free(tags);
|
||||
void adapter_storage_update_tag(neu_persister_t *persister, const char *node,
|
||||
const char *group, const neu_datatag_t *tag)
|
||||
{
|
||||
int rv = neu_persister_update_tag(persister, 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)
|
||||
{
|
||||
int rv = neu_persister_delete_tag(persister, 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 rv = neu_persister_load_adapter_setting(persister, node,
|
||||
(const char **) setting);
|
||||
int rv = neu_persister_load_node_setting(persister, node,
|
||||
(const char **) setting);
|
||||
if (0 != rv) {
|
||||
nlog_info("load %s setting fail", node);
|
||||
return -1;
|
||||
@@ -86,35 +94,31 @@ int adapter_load_group_and_tag(neu_persister_t * persister,
|
||||
UT_array * group_infos = NULL;
|
||||
neu_adapter_t *adapter = (neu_adapter_t *) driver;
|
||||
|
||||
int rv = neu_persister_load_group_configs(persister, adapter->name,
|
||||
&group_infos);
|
||||
int rv = neu_persister_load_groups(persister, adapter->name, &group_infos);
|
||||
if (0 != rv) {
|
||||
nlog_warn("load %s group fail", adapter->name);
|
||||
return rv;
|
||||
}
|
||||
|
||||
utarray_foreach(group_infos, neu_persist_group_config_info_t *, p)
|
||||
utarray_foreach(group_infos, neu_persist_group_info_t *, p)
|
||||
{
|
||||
UT_array *tags = NULL;
|
||||
neu_adapter_driver_add_group(driver, p->group_config_name,
|
||||
p->read_interval);
|
||||
neu_adapter_driver_add_group(driver, p->name, p->interval);
|
||||
|
||||
rv = neu_persister_load_datatags(persister, adapter->name,
|
||||
p->group_config_name, &tags);
|
||||
rv = neu_persister_load_tags(persister, adapter->name, p->name, &tags);
|
||||
if (0 != rv) {
|
||||
nlog_warn("load %s:%s tags fail", adapter->name,
|
||||
p->group_config_name);
|
||||
nlog_warn("load %s:%s tags fail", adapter->name, p->name);
|
||||
continue;
|
||||
}
|
||||
|
||||
utarray_foreach(tags, neu_datatag_t *, tag)
|
||||
{
|
||||
neu_adapter_driver_add_tag(driver, p->group_config_name, tag);
|
||||
neu_adapter_driver_add_tag(driver, p->name, tag);
|
||||
}
|
||||
|
||||
neu_persist_datatag_infos_free(tags);
|
||||
utarray_free(tags);
|
||||
}
|
||||
|
||||
neu_persist_group_config_infos_free(group_infos);
|
||||
utarray_free(group_infos);
|
||||
return rv;
|
||||
}
|
||||
@@ -30,8 +30,12 @@ void adapter_storage_add_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_tag(neu_persister_t *persister, neu_adapter_t *adapter,
|
||||
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_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);
|
||||
|
||||
int adapter_load_setting(neu_persister_t *persister, const char *node,
|
||||
char **setting);
|
||||
|
||||
+4
-2
@@ -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);
|
||||
}
|
||||
|
||||
@@ -137,8 +137,8 @@ UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app)
|
||||
return neu_subscribe_manager_get(manager->subscribe_manager, app);
|
||||
}
|
||||
|
||||
int neu_manager_get_adapter_info(neu_manager_t *manager, const char *name,
|
||||
neu_persist_adapter_info_t *info)
|
||||
int neu_manager_get_node_info(neu_manager_t *manager, const char *name,
|
||||
neu_persist_node_info_t *info)
|
||||
{
|
||||
neu_adapter_t *adapter = neu_node_manager_find(manager->node_manager, name);
|
||||
|
||||
|
||||
@@ -57,8 +57,8 @@ int neu_manager_unsubscribe(neu_manager_t *manager, const char *app,
|
||||
const char *driver, const char *group);
|
||||
UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app);
|
||||
|
||||
int neu_manager_get_adapter_info(neu_manager_t *manager, const char *name,
|
||||
neu_persist_adapter_info_t *info);
|
||||
int neu_manager_get_node_info(neu_manager_t *manager, const char *name,
|
||||
neu_persist_node_info_t *info);
|
||||
|
||||
void neu_manager_notify_app_sub(neu_manager_t *manager, const char *app,
|
||||
const char *driver, const char *group);
|
||||
|
||||
+38
-32
@@ -37,40 +37,48 @@ void manager_strorage_plugin(neu_manager_t *manager)
|
||||
|
||||
void manager_storage_add_node(neu_manager_t *manager, const char *node)
|
||||
{
|
||||
int rv = 0;
|
||||
neu_persist_adapter_info_t adapter_info = {};
|
||||
int rv = 0;
|
||||
neu_persist_node_info_t node_info = {};
|
||||
|
||||
rv = neu_manager_get_adapter_info(manager, node, &adapter_info);
|
||||
rv = neu_manager_get_node_info(manager, node, &node_info);
|
||||
if (0 != rv) {
|
||||
nlog_error("unable to get adapter:%s info", node);
|
||||
return;
|
||||
}
|
||||
|
||||
rv = neu_persister_store_adapter(manager->persister, &adapter_info);
|
||||
rv = neu_persister_store_node(manager->persister, &node_info);
|
||||
if (0 != rv) {
|
||||
nlog_error("failed to store adapter info");
|
||||
}
|
||||
|
||||
neu_persist_adapter_info_fini(&adapter_info);
|
||||
neu_persist_node_info_fini(&node_info);
|
||||
}
|
||||
|
||||
void manager_storage_del_node(neu_manager_t *manager, const char *node)
|
||||
{
|
||||
neu_persister_delete_adapter(manager->persister, 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)
|
||||
@@ -97,26 +105,26 @@ int manager_load_plugin(neu_manager_t *manager)
|
||||
|
||||
int manager_load_node(neu_manager_t *manager)
|
||||
{
|
||||
UT_array *adapter_infos = NULL;
|
||||
int rv = 0;
|
||||
UT_array *node_infos = NULL;
|
||||
int rv = 0;
|
||||
|
||||
rv = neu_persister_load_adapters(manager->persister, &adapter_infos);
|
||||
rv = neu_persister_load_nodes(manager->persister, &node_infos);
|
||||
if (0 != rv) {
|
||||
nlog_error("failed to load adapter infos");
|
||||
return -1;
|
||||
}
|
||||
|
||||
utarray_foreach(adapter_infos, neu_persist_adapter_info_t *, adapter_info)
|
||||
utarray_foreach(node_infos, neu_persist_node_info_t *, node_info)
|
||||
{
|
||||
rv = neu_manager_add_node(manager, adapter_info->name,
|
||||
adapter_info->plugin_name);
|
||||
rv = neu_manager_add_node(manager, node_info->name,
|
||||
node_info->plugin_name);
|
||||
const char *ok_or_err = (0 == rv) ? "success" : "fail";
|
||||
nlog_info("load adapter %s type:%" PRId64 ", name:%s plugin:%s",
|
||||
ok_or_err, adapter_info->type, adapter_info->name,
|
||||
adapter_info->plugin_name);
|
||||
ok_or_err, node_info->type, node_info->name,
|
||||
node_info->plugin_name);
|
||||
}
|
||||
|
||||
neu_persist_adapter_infos_free(adapter_infos);
|
||||
utarray_free(node_infos);
|
||||
return rv;
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+4
-1
@@ -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);
|
||||
|
||||
+559
-609
File diff suppressed because it is too large
Load Diff
+135
-129
@@ -25,69 +25,48 @@ extern "C" {
|
||||
#endif
|
||||
|
||||
#include "adapter/adapter_info.h"
|
||||
#include "persist/json/persist_json_adapter.h"
|
||||
#include "persist/json/persist_json_datatags.h"
|
||||
#include "persist/json/persist_json_group_configs.h"
|
||||
#include "persist/json/persist_json_plugin.h"
|
||||
#include "persist/json/persist_json_subs.h"
|
||||
|
||||
typedef neu_json_node_req_node_t neu_persist_adapter_info_t;
|
||||
typedef neu_json_plugin_req_plugin_t neu_persist_plugin_info_t;
|
||||
typedef neu_json_datatag_req_tag_t neu_persist_datatag_info_t;
|
||||
typedef neu_json_group_configs_req_t neu_persist_group_config_info_t;
|
||||
typedef neu_json_subscriptions_req_subscription_t
|
||||
neu_persist_subscription_info_t;
|
||||
|
||||
static inline void
|
||||
neu_persist_adapter_info_fini(neu_persist_adapter_info_t *adapter_info)
|
||||
{
|
||||
free(adapter_info->name);
|
||||
free(adapter_info->plugin_name);
|
||||
}
|
||||
typedef struct {
|
||||
char * name;
|
||||
int64_t type;
|
||||
char * plugin_name;
|
||||
int64_t state;
|
||||
} neu_persist_node_info_t;
|
||||
|
||||
static inline void neu_persist_adapter_infos_free(UT_array *adapter_infos)
|
||||
{
|
||||
utarray_foreach(adapter_infos, neu_persist_adapter_info_t *, p)
|
||||
{
|
||||
free(p->name);
|
||||
free(p->plugin_name);
|
||||
}
|
||||
typedef struct {
|
||||
uint32_t interval;
|
||||
char * name;
|
||||
} neu_persist_group_info_t;
|
||||
|
||||
utarray_free(adapter_infos);
|
||||
}
|
||||
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)
|
||||
{
|
||||
utarray_free(plugin_infos);
|
||||
}
|
||||
|
||||
static inline void
|
||||
neu_persist_group_config_infos_free(UT_array *group_config_infos)
|
||||
static inline void neu_persist_node_info_fini(neu_persist_node_info_t *info)
|
||||
{
|
||||
utarray_foreach(group_config_infos, neu_persist_group_config_info_t *, p)
|
||||
{
|
||||
free(p->group_config_name);
|
||||
free(p->adapter_name);
|
||||
}
|
||||
|
||||
utarray_free(group_config_infos);
|
||||
free(info->name);
|
||||
free(info->plugin_name);
|
||||
}
|
||||
|
||||
static inline void neu_persist_datatag_infos_free(UT_array *datatag_infos)
|
||||
static inline void neu_persist_group_info_fini(neu_persist_group_info_t *info)
|
||||
{
|
||||
utarray_free(datatag_infos);
|
||||
free(info->name);
|
||||
}
|
||||
|
||||
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);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -95,7 +74,6 @@ neu_persist_subscription_infos_free(UT_array *subscription_infos)
|
||||
typedef struct neu_persister neu_persister_t;
|
||||
|
||||
const char *neu_persister_get_persist_dir(neu_persister_t *persister);
|
||||
const char *neu_persister_get_adapters_fname(neu_persister_t *persister);
|
||||
const char *neu_persister_get_plugins_fname(neu_persister_t *persister);
|
||||
|
||||
/**
|
||||
@@ -110,41 +88,39 @@ neu_persister_t *neu_persister_create(const char *dir_name);
|
||||
void neu_persister_destroy(neu_persister_t *persiter);
|
||||
|
||||
/**
|
||||
* Persist adapters.
|
||||
* Persist nodes.
|
||||
* @param persister persiter object.
|
||||
* @param adapter_infos neu_persist_adapter_info_t.
|
||||
* @param node_info neu_persist_node_info_t.
|
||||
* @return 0 on success, non-zero on failure
|
||||
*/
|
||||
int neu_persister_store_adapter(neu_persister_t * persister,
|
||||
neu_persist_adapter_info_t *info);
|
||||
int neu_persister_store_node(neu_persister_t * persister,
|
||||
neu_persist_node_info_t *info);
|
||||
/**
|
||||
* Load adapter infos.
|
||||
* Load node infos.
|
||||
* @param persister persiter object.
|
||||
* @param[out] adapter_infos used to return pointer to heap allocated
|
||||
* vector of neu_persist_adapter_info_t.
|
||||
* @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_adapters(neu_persister_t *persister,
|
||||
UT_array ** adapter_infos);
|
||||
int neu_persister_load_nodes(neu_persister_t *persister, UT_array **node_infos);
|
||||
/**
|
||||
* Delete adapter.
|
||||
* Delete node.
|
||||
* @param persister persiter object.
|
||||
* @param adapter_name name of the adapter to delete.
|
||||
* @param node_name name of the node to delete.
|
||||
* @return 0 on success, none-zero on failure
|
||||
*/
|
||||
int neu_persister_delete_adapter(neu_persister_t *persister,
|
||||
const char * adapter_name);
|
||||
int neu_persister_delete_node(neu_persister_t *persister,
|
||||
const char * node_name);
|
||||
|
||||
/**
|
||||
* Update adapter.
|
||||
* Update node.
|
||||
* @param persister persiter object.
|
||||
* @param adapter_name name of the adapter to update.
|
||||
* @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_adapter(neu_persister_t *persister,
|
||||
const char * adapter_name,
|
||||
const char * new_name);
|
||||
int neu_persister_update_node(neu_persister_t *persister, const char *node_name,
|
||||
const char *new_name);
|
||||
|
||||
/**
|
||||
* Persist plugins.
|
||||
@@ -165,118 +141,148 @@ int neu_persister_load_plugins(neu_persister_t *persister,
|
||||
UT_array ** plugin_infos);
|
||||
|
||||
/**
|
||||
* Persist adapter datatags.
|
||||
* Persist node tags.
|
||||
* @param persister persiter object.
|
||||
* @param adapter_name name of the adapter who owns the datatags.
|
||||
* @param group_config_name name of the group config.
|
||||
* @param datatag_infos vector of neu_persist_datatag_info_t.
|
||||
* @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_datatags(neu_persister_t *persister,
|
||||
const char * adapter_name,
|
||||
const char * group_config_name,
|
||||
UT_array * datatag_infos);
|
||||
/**
|
||||
* Load adapter datatag infos.
|
||||
* @param persister persiter object.
|
||||
* @param adapter_name name of the adapter who owns the datatags.
|
||||
* @param group_config_name name of the group config.
|
||||
* @param[out] datatag_infos used to return pointer to heap allocated
|
||||
* vector of neu_persist_datatag_info_t.
|
||||
* @return 0 on success, non-zero otherwise
|
||||
*/
|
||||
int neu_persister_load_datatags(neu_persister_t *persister,
|
||||
const char * adapter_name,
|
||||
const char * group_config_name,
|
||||
UT_array ** datatag_infos);
|
||||
int neu_persister_store_tag(neu_persister_t *persister, const char *driver_name,
|
||||
const char *group_name, const neu_datatag_t *tag);
|
||||
|
||||
/**
|
||||
* Persist adapter subscriptions.
|
||||
* Load node tag infos.
|
||||
* @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 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_store_subscriptions(neu_persister_t *persister,
|
||||
const char * adapter_name,
|
||||
UT_array * subscription_infos);
|
||||
int neu_persister_load_tags(neu_persister_t *persister, const char *driver_name,
|
||||
const char *group_name, UT_array **tag_infos);
|
||||
|
||||
/**
|
||||
* Load adapter subscription infos.
|
||||
* Update node tags.
|
||||
* @param persister persiter object.
|
||||
* @param adapter_name name of the adapter who owns the
|
||||
* subscriptions.
|
||||
* @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,
|
||||
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,
|
||||
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);
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* 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.
|
||||
* @param adapter_name name of the adapter who owns the group
|
||||
* config.
|
||||
* @param group_config_info group config info to persist.
|
||||
* @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_config(
|
||||
neu_persister_t *persister, const char *adapter_name,
|
||||
neu_persist_group_config_info_t *group_config_info);
|
||||
int neu_persister_store_group(neu_persister_t * persister,
|
||||
const char * driver_name,
|
||||
neu_persist_group_info_t *group_info);
|
||||
/**
|
||||
* Load all group config infos under an adapter.
|
||||
* @param persister persiter object.
|
||||
* @param adapter_name name of the adapter who owns the group
|
||||
* configs.
|
||||
* @param[out] group_config_infos used to return pointer to heap allocated
|
||||
* vector of neu_persist_group_config_info_t.
|
||||
* @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_group_configs(neu_persister_t *persister,
|
||||
const char * adapter_name,
|
||||
UT_array ** group_config_infos);
|
||||
int neu_persister_load_groups(neu_persister_t *persister,
|
||||
const char *driver_name, UT_array **group_infos);
|
||||
/**
|
||||
* Delete group config.
|
||||
* @param persister persiter object.
|
||||
* @param adapter_name name of the adapter who owns the group
|
||||
* config.
|
||||
* @param group_config_name name of the group config.
|
||||
* @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_config(neu_persister_t *persister,
|
||||
const char * adapter_name,
|
||||
const char * group_config_name);
|
||||
int neu_persister_delete_group(neu_persister_t *persister,
|
||||
const char *driver_name, const char *group_name);
|
||||
|
||||
/**
|
||||
* Persist adapter setting.
|
||||
* Persist node setting.
|
||||
* @param persister persiter object.
|
||||
* @param adapter_name name of the adapter who owns the setting.
|
||||
* @param setting adapter setting string.
|
||||
* @param setting node setting string.
|
||||
* @return 0 on success, non-zero otherwise
|
||||
*/
|
||||
int neu_persister_store_adapter_setting(neu_persister_t *persister,
|
||||
const char * adapter_name,
|
||||
const char * setting);
|
||||
int neu_persister_store_node_setting(neu_persister_t *persister,
|
||||
const char * node_name,
|
||||
const char * setting);
|
||||
/**
|
||||
* Load adapter setting.
|
||||
* Load node setting.
|
||||
* @param persister persiter object.
|
||||
* @param adapter_name name of the adapter.
|
||||
* @param[out] setting used to return adapter setting string.
|
||||
* @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_adapter_setting(neu_persister_t * persister,
|
||||
const char * adapter_name,
|
||||
const char **const setting);
|
||||
int neu_persister_load_node_setting(neu_persister_t * persister,
|
||||
const char * node_name,
|
||||
const char **const setting);
|
||||
/**
|
||||
* Delete adapter setting.
|
||||
* Delete node setting.
|
||||
* @param persister persiter object.
|
||||
* @param adapter_name name of the adapter.
|
||||
* @param node_name name of the node.
|
||||
* @return 0 on success, none-zero on failure
|
||||
*/
|
||||
int neu_persister_delete_adapter_setting(neu_persister_t *persister,
|
||||
const char * adapter_name);
|
||||
int neu_persister_delete_node_setting(neu_persister_t *persister,
|
||||
const char * node_name);
|
||||
|
||||
#ifdef __cplusplus
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user