From ff5ab12d475389d8ba08bb47c4f26b5b0b87fd41 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Mon, 12 Jun 2023 14:35:39 +0800 Subject: [PATCH 1/2] fix: neu_json_encode_ndriver_tag --- src/parser/neu_json_ndriver.c | 12 +++--------- 1 file changed, 3 insertions(+), 9 deletions(-) diff --git a/src/parser/neu_json_ndriver.c b/src/parser/neu_json_ndriver.c index f6bea70a..bb16252a 100644 --- a/src/parser/neu_json_ndriver.c +++ b/src/parser/neu_json_ndriver.c @@ -198,15 +198,9 @@ int neu_json_encode_ndriver_tag(void *json_obj, void *param) return -1; } - neu_json_elem_t params_elem[] = { - { - .name = "params", - .t = NEU_JSON_STR, - .v.val_str = tag->params, - }, - }; - ret = neu_json_encode_field(json_obj, params_elem, - NEU_JSON_ELEM_SIZE(params_elem)); + if (tag->params) { + ret = neu_json_load_key(json_obj, "params", tag->params, true); + } return ret; } From ded6a537d1d28b5edbde62b463f2666831d2319e Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Mon, 12 Jun 2023 14:34:54 +0800 Subject: [PATCH 2/2] feat(core): forward add/del ndriver map message to ndriver node --- include/neuron/errcodes.h | 1 + src/adapter/adapter.c | 6 +++ src/core/manager.c | 26 ++++++++----- src/core/manager_internal.c | 76 +++++++++++++++++++++++++++++++------ src/core/manager_internal.h | 6 +++ src/core/storage.c | 14 +++++++ src/core/storage.h | 6 +++ src/core/subscribe.c | 24 ++++++++++++ src/core/subscribe.h | 2 + src/utils/http.c | 1 + 10 files changed, 140 insertions(+), 22 deletions(-) diff --git a/include/neuron/errcodes.h b/include/neuron/errcodes.h index 30f37cfe..0e0106dd 100644 --- a/include/neuron/errcodes.h +++ b/include/neuron/errcodes.h @@ -58,6 +58,7 @@ typedef enum { NEU_ERR_NODE_NOT_ALLOW_DELETE = 2011, NEU_ERR_NODE_NOT_ALLOW_SUBSCRIBE = 2012, NEU_ERR_NODE_NOT_ALLOW_UPDATE = 2013, + NEU_ERR_NODE_NOT_ALLOW_MAP = 2014, NEU_ERR_GROUP_ALREADY_SUBSCRIBED = 2101, NEU_ERR_GROUP_NOT_SUBSCRIBE = 2102, diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 71d556f0..1c6a82e8 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -844,6 +844,12 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) } break; } + case NEU_REQ_ADD_NDRIVER_MAP: { + break; + } + case NEU_REQ_DEL_NDRIVER_MAP: { + break; + } case NEU_REQ_UPDATE_NDRIVER_TAG_PARAM: { neu_req_update_ndriver_tag_param_t *cmd = (neu_req_update_ndriver_tag_param_t *) &header[1]; diff --git a/src/core/manager.c b/src/core/manager.c index e358a294..6e0c69dc 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -925,9 +925,13 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_req_ndriver_map_t *cmd = (neu_req_ndriver_map_t *) &header[1]; neu_resp_error_t error = { 0 }; - // TODO - nlog_info("add map ndriver:`%s` driver:`%s` group:`%s`", cmd->ndriver, - cmd->driver, cmd->group); + error.error = neu_manager_add_ndriver_map(manager, cmd->ndriver, + cmd->driver, cmd->group); + if (error.error == NEU_ERR_SUCCESS) { + forward_msg(manager, msg, cmd->ndriver); + manager_storage_add_ndriver_map(manager, cmd->ndriver, cmd->driver, + cmd->group); + } header->type = NEU_RESP_ERROR; strcpy(header->receiver, header->sender); @@ -938,9 +942,13 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_req_ndriver_map_t *cmd = (neu_req_ndriver_map_t *) &header[1]; neu_resp_error_t error = { 0 }; - // TODO - nlog_info("del map ndriver:`%s` driver:`%s` group:`%s`", cmd->ndriver, - cmd->driver, cmd->group); + error.error = neu_manager_del_ndriver_map(manager, cmd->ndriver, + cmd->driver, cmd->group); + if (error.error == NEU_ERR_SUCCESS) { + forward_msg(manager, msg, cmd->ndriver); + manager_storage_del_ndriver_map(manager, cmd->ndriver, cmd->driver, + cmd->group); + } header->type = NEU_RESP_ERROR; strcpy(header->receiver, header->sender); @@ -953,10 +961,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_resp_error_t e = { 0 }; neu_resp_get_ndriver_maps_t resp = { 0 }; - // TODO - (void) cmd; - UT_icd icd = { sizeof(neu_resp_get_ndriver_maps_t), NULL, NULL, NULL }; - utarray_new(resp.groups, &icd); + e.error = + neu_manager_get_ndriver_maps(manager, cmd->ndriver, &resp.groups); strcpy(header->receiver, header->sender); if (0 == e.error) { diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c index 225934ae..e4710dcc 100644 --- a/src/core/manager_internal.c +++ b/src/core/manager_internal.c @@ -572,9 +572,9 @@ UT_array *neu_manager_get_driver_group(neu_manager_t *manager) return driver_groups; } -int neu_manager_subscribe(neu_manager_t *manager, const char *app, - const char *driver, const char *group, - const char *params) +static inline int manager_subscribe(neu_manager_t *manager, const char *app, + const char *driver, const char *group, + const char *params) { int ret = NEU_ERR_SUCCESS; nng_pipe pipe = { 0 }; @@ -585,25 +585,37 @@ int neu_manager_subscribe(neu_manager_t *manager, const char *app, return NEU_ERR_NODE_NOT_EXIST; } - if (0 == strcmp(app, "monitor")) { - // filter out monitor node - return NEU_ERR_NODE_NOT_ALLOW_SUBSCRIBE; - } - ret = neu_adapter_driver_group_exist((neu_adapter_driver_t *) adapter, group); if (ret != NEU_ERR_SUCCESS) { return ret; } - adapter = neu_node_manager_find(manager->node_manager, app); + pipe = neu_node_manager_get_pipe(manager->node_manager, app); + return neu_subscribe_manager_sub(manager->subscribe_manager, driver, app, + group, params, pipe); +} + +int neu_manager_subscribe(neu_manager_t *manager, const char *app, + const char *driver, const char *group, + const char *params) +{ + if (0 == strcmp(app, "monitor")) { + // filter out monitor node + return NEU_ERR_NODE_NOT_ALLOW_SUBSCRIBE; + } + + neu_adapter_t *adapter = neu_node_manager_find(manager->node_manager, app); + if (adapter == NULL) { return NEU_ERR_NODE_NOT_EXIST; } - pipe = neu_node_manager_get_pipe(manager->node_manager, app); - return neu_subscribe_manager_sub(manager->subscribe_manager, driver, app, - group, params, pipe); + if (NEU_NA_TYPE_APP != neu_adapter_get_type(adapter)) { + return NEU_ERR_NODE_NOT_ALLOW_SUBSCRIBE; + } + + return manager_subscribe(manager, app, driver, group, params); } int neu_manager_send_subscribe(neu_manager_t *manager, const char *app, @@ -672,6 +684,46 @@ UT_array *neu_manager_get_sub_group_deep_copy(neu_manager_t *manager, return subs; } +int neu_manager_add_ndriver_map(neu_manager_t *manager, const char *ndriver, + const char *driver, const char *group) +{ + neu_adapter_t *adapter = + neu_node_manager_find(manager->node_manager, ndriver); + + if (adapter == NULL) { + return NEU_ERR_NODE_NOT_EXIST; + } + + if (NEU_NA_TYPE_NDRIVER != neu_adapter_get_type(adapter)) { + return NEU_ERR_NODE_NOT_ALLOW_MAP; + } + + return manager_subscribe(manager, ndriver, driver, group, NULL); +} + +int neu_manager_del_ndriver_map(neu_manager_t *manager, const char *ndriver, + const char *driver, const char *group) +{ + return neu_manager_unsubscribe(manager, ndriver, driver, group); +} + +int neu_manager_get_ndriver_maps(neu_manager_t *manager, const char *ndriver, + UT_array **result) +{ + neu_adapter_t *adapter = + neu_node_manager_find(manager->node_manager, ndriver); + + if (adapter == NULL) { + return NEU_ERR_NODE_NOT_EXIST; + } else if (NEU_NA_TYPE_NDRIVER != neu_adapter_get_type(adapter)) { + return NEU_ERR_NODE_NOT_ALLOW_MAP; + } + + *result = neu_subscribe_manager_get_ndriver_maps(manager->subscribe_manager, + ndriver); + return 0; +} + int neu_manager_get_node_info(neu_manager_t *manager, const char *name, neu_persist_node_info_t *info) { diff --git a/src/core/manager_internal.h b/src/core/manager_internal.h index 1d40220d..4e3de318 100644 --- a/src/core/manager_internal.h +++ b/src/core/manager_internal.h @@ -105,6 +105,12 @@ int neu_manager_unsubscribe(neu_manager_t *manager, const char *app, UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app); UT_array *neu_manager_get_sub_group_deep_copy(neu_manager_t *manager, const char * app); +int neu_manager_add_ndriver_map(neu_manager_t *manager, const char *ndriver, + const char *driver, const char *group); +int neu_manager_del_ndriver_map(neu_manager_t *manager, const char *ndriver, + const char *driver, const char *group); +int neu_manager_get_ndriver_maps(neu_manager_t *manager, const char *ndriver, + UT_array **result); int neu_manager_get_node_info(neu_manager_t *manager, const char *name, neu_persist_node_info_t *info); diff --git a/src/core/storage.c b/src/core/storage.c index bba1f51c..f0967a52 100644 --- a/src/core/storage.c +++ b/src/core/storage.c @@ -92,6 +92,20 @@ void manager_storage_unsubscribe(neu_manager_t *manager, const char *app, } } +void manager_storage_add_ndriver_map(neu_manager_t *manager, + const char *ndriver, const char *driver, + const char *group) +{ + return manager_storage_subscribe(manager, ndriver, driver, group, NULL); +} + +void manager_storage_del_ndriver_map(neu_manager_t *manager, + const char *ndriver, const char *driver, + const char *group) +{ + return manager_storage_unsubscribe(manager, ndriver, driver, group); +} + static int save_node_group_and_tags(neu_group_t *grp, void *data) { const char *node = data; diff --git a/src/core/storage.h b/src/core/storage.h index f28c7245..e3669b58 100644 --- a/src/core/storage.h +++ b/src/core/storage.h @@ -34,6 +34,12 @@ void manager_storage_subscribe(neu_manager_t *manager, const char *app, const char *params); void manager_storage_unsubscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group); +void manager_storage_add_ndriver_map(neu_manager_t *manager, + const char *ndriver, const char *driver, + const char *group); +void manager_storage_del_ndriver_map(neu_manager_t *manager, + const char *ndriver, const char *driver, + const char *group); void manager_storage_inst_node(neu_manager_t *manager, const char *tmpl_name, const char *node); diff --git a/src/core/subscribe.c b/src/core/subscribe.c index 1dd55f2b..a8e439fd 100644 --- a/src/core/subscribe.c +++ b/src/core/subscribe.c @@ -132,6 +132,30 @@ UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app) return groups; } +UT_array *neu_subscribe_manager_get_ndriver_maps(neu_subscribe_mgr_t *mgr, + const char * ndriver) +{ + UT_array *groups = NULL; + UT_icd icd = { sizeof(neu_resp_ndriver_map_info_t), NULL, NULL, NULL }; + utarray_new(groups, &icd); + + sub_elem_t *el = NULL, *tmp = NULL; + HASH_ITER(hh, mgr->ss, el, tmp) + { + utarray_foreach(el->apps, neu_app_subscribe_t *, sub_app) + { + if (strcmp(sub_app->app_name, ndriver) == 0) { + neu_resp_ndriver_map_info_t info = { 0 }; + strncpy(info.driver, el->key.driver, sizeof(info.driver)); + strncpy(info.group, el->key.group, sizeof(info.group)); + utarray_push_back(groups, &info); + } + } + } + + return groups; +} + void neu_subscribe_manager_unsub_all(neu_subscribe_mgr_t *mgr, const char *app) { sub_elem_t *el = NULL, *tmp = NULL; diff --git a/src/core/subscribe.h b/src/core/subscribe.h index 07c2cab3..cdd0535e 100644 --- a/src/core/subscribe.h +++ b/src/core/subscribe.h @@ -47,6 +47,8 @@ UT_array *neu_subscribe_manager_find(neu_subscribe_mgr_t *mgr, UT_array *neu_subscribe_manager_find_by_driver(neu_subscribe_mgr_t *mgr, const char * driver); UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app); +UT_array *neu_subscribe_manager_get_ndriver_maps(neu_subscribe_mgr_t *mgr, + const char * ndriver); void neu_subscribe_manager_unsub_all(neu_subscribe_mgr_t *mgr, const char *app); int neu_subscribe_manager_sub(neu_subscribe_mgr_t *mgr, const char *driver, const char *app, const char *group, diff --git a/src/utils/http.c b/src/utils/http.c index d60a6d39..58da0797 100644 --- a/src/utils/http.c +++ b/src/utils/http.c @@ -322,6 +322,7 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content) case NEU_ERR_NODE_SETTING_INVALID: case NEU_ERR_NODE_NOT_ALLOW_UPDATE: case NEU_ERR_NODE_NOT_ALLOW_SUBSCRIBE: + case NEU_ERR_NODE_NOT_ALLOW_MAP: case NEU_ERR_LIBRARY_INFO_INVALID: case NEU_ERR_LICENSE_INVALID: case NEU_ERR_GROUP_PARAMETER_INVALID: