GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-17 04:40:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #367 from eeff/eeff/fix

Browse Source 
Add persistence with adapter id
This commit is contained in:
fengzeroz
2022-03-29 17:08:05 +08:00
committed by GitHub
parent 2ccab9b7ef b58ffcd8a3
commit d01e8c90f3
7 changed files with 119 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/json-persistence/json_adapter.json
+6
View File
@@ -2,18 +2,21 @@
"node_req": {
"nodes": [
{
"id": 1,
"state": 1,
"type": 1,
"name": "sample-driver-adapter",
"plugin_name": "sample-drv-plugin"
},
{
"id": 2,
"state": 2,
"type": 5,
"name": "sample-app-adapter",
"plugin_name": "sample-app-plugin"
},
{
"id": 3,
"state": 3,
"type": 3,
"name": "mqtt-adapter",
@@ -24,18 +27,21 @@
"node_resp": {
"nodes": [
{
"id": 1,
"state": 1,
"type": 1,
"name": "sample-driver-adapter",
"plugin_name": "sample-drv-plugin"
},
{
"id": 2,
"state": 2,
"type": 5,
"name": "sample-app-adapter",
"plugin_name": "sample-app-plugin"
},
{
"id": 3,
"state": 3,
"type": 3,
"name": "mqtt-adapter",
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);
src/persist/json/persist_json_adapter.c
+19 -25
View File
@@ -1,6 +1,6 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2021 EMQ Technologies Co., Ltd All rights reserved.
* Copyright (C) 2020-2022 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
@@ -56,6 +56,11 @@ int neu_json_encode_node_resp(void *json_object, void *param)
.name = "name",
.t = NEU_JSON_STR,
.v.val_str = p_node->name,
},
{
.name = "id",
.t = NEU_JSON_INT,
.v.val_int = p_node->id,
} };
node_array = neu_json_encode_array(node_array, node_elems,
NEU_JSON_ELEM_SIZE(node_elems));
@@ -75,21 +80,11 @@ int neu_json_encode_node_resp(void *json_object, void *param)
int neu_json_decode_node_req(char *buf, neu_json_node_req_t **result)
{
int ret = 0;
void * json_obj = NULL;
neu_json_node_req_t *req = calloc(1, sizeof(neu_json_node_req_t));
if (req == NULL) {
return -1;
}
int ret = 0;
neu_json_node_req_t *req = calloc(1, sizeof(neu_json_node_req_t));
json_obj = neu_json_decode_new(buf);
req->n_node = neu_json_decode_array_size_by_json(json_obj, "nodes");
if (req->n_node < 0) {
goto decode_fail;
}
req->nodes = calloc(req->n_node, sizeof(neu_json_node_req_node_t));
req->n_node = neu_json_decode_array_size(buf, "nodes");
req->nodes = calloc(req->n_node, sizeof(neu_json_node_req_node_t));
neu_json_node_req_node_t *p_node = req->nodes;
for (int i = 0; i < req->n_node; i++) {
neu_json_elem_t node_elems[] = { {
@@ -107,9 +102,13 @@ int neu_json_decode_node_req(char *buf, neu_json_node_req_t **result)
{
.name = "name",
.t = NEU_JSON_STR,
},
{
.name = "id",
.t = NEU_JSON_INT,
} };
ret = neu_json_decode_array_by_json(
json_obj, "nodes", i, NEU_JSON_ELEM_SIZE(node_elems), node_elems);
ret = neu_json_decode_array(buf, "nodes", i,
NEU_JSON_ELEM_SIZE(node_elems), node_elems);
if (ret != 0) {
goto decode_fail;
}
@@ -118,11 +117,12 @@ int neu_json_decode_node_req(char *buf, neu_json_node_req_t **result)
p_node->state = node_elems[1].v.val_int;
p_node->plugin_name = node_elems[2].v.val_str;
p_node->name = node_elems[3].v.val_str;
p_node->id = node_elems[4].v.val_int;
p_node++;
}
*result = req;
goto decode_exit;
return ret;
decode_fail:
if (req->nodes != NULL) {
@@ -131,13 +131,7 @@ decode_fail:
if (req != NULL) {
free(req);
}
ret = -1;
decode_exit:
if (json_obj != NULL) {
neu_json_decode_free(json_obj);
}
return ret;
return -1;
}
void neu_json_decode_node_req_free(neu_json_node_req_t *req)
src/persist/json/persist_json_adapter.h
+3 -1
View File
@@ -1,6 +1,6 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2021 EMQ Technologies Co., Ltd All rights reserved.
* Copyright (C) 2020-2022 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
@@ -32,6 +32,7 @@ extern "C" {
#endif
typedef struct {
int64_t id;
char * name;
int64_t type;
char * plugin_name;
@@ -46,6 +47,7 @@ typedef struct {
int neu_json_encode_node_resp(void *json_object, void *param);
typedef struct {
int64_t id;
char * name;
int64_t type;
char * plugin_name;
tests/json_persistence_test.cc
+11 -8
View File
@@ -13,13 +13,13 @@ TEST(JsonAdapter, AdapterPersistenceDecode)
{
char *buf =
(char *) "{\"nodes\": "
"[{\"type\":1, \"state\": 1, \"name\": "
"[{\"id\":1, \"type\":1, \"state\": 1, \"name\": "
"\"sample-driver-adapter\", \"plugin_name\": "
"\"sample-drv-plugin\"}, "
"{\"type\":5, \"state\": 2, \"name\": "
"{\"id\":2, \"type\":5, \"state\": 2, \"name\": "
"\"sample-app-adapter\", \"plugin_name\": "
"\"sample-app-plugin\"}, "
"{\"type\": 3, \"state\": 3, \"name\": "
"{\"id\":3, \"type\": 3, \"state\": 3, \"name\": "
"\"mqtt-adapter\", \"plugin_name\": \"mqtt-plugin\"}]}";
neu_json_node_req_t *req = NULL;
@@ -160,11 +160,11 @@ TEST(JsonAdapterTest, AdapterPersistenceEncode)
{
char *buf =
(char *) "{\"nodes\": [{\"type\": 1, \"state\": 1, \"plugin_name\": "
"\"sample-drv-plugin\", \"name\": \"sample-driver-adapter\"}, "
"{\"type\": 5, \"state\": 2, \"plugin_name\": "
"\"sample-app-plugin\", \"name\": \"sample-app-adapter\"}, "
"{\"type\": 3, \"state\": 3, \"plugin_name\": "
"\"mqtt-plugin\", \"name\": \"mqtt-adapter\"}]}";
"\"sample-drv-plugin\", \"name\": \"sample-driver-adapter\", "
"\"id\": 1}, {\"type\": 5, \"state\": 2, \"plugin_name\": "
"\"sample-app-plugin\", \"name\": \"sample-app-adapter\", "
"\"id\": 2}, {\"type\": 3, \"state\": 3, \"plugin_name\": "
"\"mqtt-plugin\", \"name\": \"mqtt-adapter\", \"id\": 3}]}";
char *buf2 = (char *) "{\"nodes\": []}";
char *result = NULL;
@@ -179,16 +179,19 @@ TEST(JsonAdapterTest, AdapterPersistenceEncode)
resp.nodes = (neu_json_node_resp_node_t *) calloc(
3, sizeof(neu_json_node_resp_node_t));
resp.nodes[0].id = 1;
resp.nodes[0].state = 1;
resp.nodes[0].type = 1;
resp.nodes[0].name = strdup((char *) "sample-driver-adapter");
resp.nodes[0].plugin_name = strdup((char *) "sample-drv-plugin");
resp.nodes[1].id = 2;
resp.nodes[1].state = 2;
resp.nodes[1].type = 5;
resp.nodes[1].name = strdup((char *) "sample-app-adapter");
resp.nodes[1].plugin_name = strdup((char *) "sample-app-plugin");
resp.nodes[2].id = 3;
resp.nodes[2].state = 3;
resp.nodes[2].type = 3;
resp.nodes[2].name = strdup((char *) "mqtt-adapter");
tests/persistence_test.cc
+3 -3
View File
@@ -10,11 +10,11 @@ TEST(PersistenceAdapterTest, Adapter)
const char *adapters_fname = neu_persister_get_adapters_fname(persister);
const char *plugins_fname = neu_persister_get_plugins_fname(persister);
neu_json_node_req_node_t node1 = { (char *) "sample-driver-adapter", 1,
neu_json_node_req_node_t node1 = { 1, (char *) "sample-driver-adapter", 1,
(char *) "sample-drv-plugin", 1 };
neu_json_node_req_node_t node2 = { (char *) "sample-app-adapter", 5,
neu_json_node_req_node_t node2 = { 2, (char *) "sample-app-adapter", 5,
(char *) "sample-app-plugin", 2 };
neu_json_node_req_node_t node3 = { (char *) "mqtt-adapter", 3,
neu_json_node_req_node_t node3 = { 3, (char *) "mqtt-adapter", 3,
(char *) "mqtt-plugin", 3 };
neu_json_node_req_t *req =