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

Update persist adapter to persist/load with adapter id

Browse Source
This commit is contained in:
eeff
2022-03-28 17:19:43 +08:00
parent d4b9220d78
commit b58ffcd8a3
2 changed files with 77 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapter/adapter.c
+34
View File
@@ -443,11 +443,22 @@ static int persister_singleton_load_subscriptions(neu_adapter_t *adapter,
return rv;
}
static int persist_adapter_info_cmp(const void *a, const void *b)
{
return ((neu_persist_adapter_info_t *) a)->id -
((neu_persist_adapter_info_t *) b)->id;
}
static int persister_singleton_load_data(neu_adapter_t *adapter)
{
vector_t * adapter_infos = NULL;
neu_persister_t *persister = persister_singleton_get();
// declaration for neu_manager_update_node_id since it is not exposed
int neu_manager_update_node_id(neu_manager_t * manager,
neu_node_id_t node_id,
neu_node_id_t new_node_id);
log_info("%s start persistence loading", adapter->name);
int rv = persister_singleton_load_plugins(adapter);
@@ -461,6 +472,10 @@ static int persister_singleton_load_data(neu_adapter_t *adapter)
goto error_load_adapters;
}
// sort by adapter id
qsort(adapter_infos->data, adapter_infos->size,
sizeof(neu_persist_adapter_info_t), persist_adapter_info_cmp);
VECTOR_FOR_EACH(adapter_infos, iter)
{
neu_persist_adapter_info_t *adapter_info = iterator_get(&iter);
@@ -480,6 +495,13 @@ static int persister_singleton_load_data(neu_adapter_t *adapter)
goto error_add_adapters;
}
if (node_id != adapter_info->id) {
// set to persisted adapter id
neu_manager_update_node_id(adapter->manager, node_id,
adapter_info->id);
node_id = adapter_info->id;
}
rv = persister_singleton_load_setting(adapter, adapter_info->name,
node_id);
if (0 != rv) {
@@ -2058,6 +2080,18 @@ adapter_id_t neu_adapter_get_id(neu_adapter_t *adapter)
return adapter->id;
}
// NOTE: Do NOT expose this function in header files, use for persistence only!
adapter_id_t neu_adapter_set_id(neu_adapter_t *adapter, adapter_id_t id)
{
if (adapter == NULL) {
return 0;
}
adapter_id_t old = adapter->id;
adapter->id = id;
return old;
}
adapter_type_e neu_adapter_get_type(neu_adapter_t *adapter)
{
if (adapter == NULL) {
src/core/neu_manager.c
+43
View File
@@ -1386,6 +1386,47 @@ int neu_manager_add_node(neu_manager_t *manager, neu_cmd_add_node_t *cmd,
return rv;
}
// NOTE: Do NOT expose this function in header files, use for persistence only!
int neu_manager_update_node_id(neu_manager_t *manager, neu_node_id_t node_id,
neu_node_id_t new_node_id)
{
int rv = 0;
adapter_id_t adapter_id;
adapter_id_t new_adapter_id;
adapter_reg_entity_t *reg_entity;
// declaration for neu_adapter_set_id since it is not exposed
adapter_id_t neu_adapter_set_id(neu_adapter_t * adapter, adapter_id_t id);
if (0 == new_node_id) {
return NEU_ERR_EINVAL;
}
adapter_id = neu_manager_adapter_id_from_node_id(manager, node_id);
new_adapter_id = neu_manager_adapter_id_from_node_id(manager, new_node_id);
nng_mtx_lock(manager->adapters_mtx);
reg_entity = find_reg_adapter_by_id(&manager->reg_adapters, new_adapter_id);
if (NULL != reg_entity) {
rv = NEU_ERR_NODE_EXIST;
} else {
reg_entity = find_reg_adapter_by_id(&manager->reg_adapters, adapter_id);
if (reg_entity != NULL) {
reg_entity->adapter_id = new_adapter_id;
neu_adapter_set_id(reg_entity->adapter, new_adapter_id);
if (new_adapter_id >= manager->new_adapter_id) {
// update new available adapter id
manager->new_adapter_id = new_adapter_id + 1;
}
} else {
rv = NEU_ERR_NODE_NOT_EXIST;
}
}
nng_mtx_unlock(manager->adapters_mtx);
return rv;
}
int neu_manager_del_node(neu_manager_t *manager, neu_node_id_t node_id)
{
int rv = 0;
@@ -2107,6 +2148,8 @@ int neu_manager_get_persist_adapter_infos(neu_manager_t *manager,
}
neu_persist_adapter_info_t adapter_info = {};
adapter_info.id =
neu_manager_adapter_id_to_node_id(manager, reg_entity->adapter_id);
adapter_info.type = neu_adapter_get_type(reg_entity->adapter);
adapter_info.state = neu_adapter_get_state(reg_entity->adapter).running;
adapter_info.name = strdup(adapter_name);