diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 56b672f8..d0fce842 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -294,22 +294,35 @@ typedef struct neu_resp_get_tag { UT_array *tags; // array neu_datatag_t } neu_resp_get_tag_t; +typedef struct { + char app[NEU_NODE_NAME_LEN]; + char driver[NEU_NODE_NAME_LEN]; + char group[NEU_GROUP_NAME_LEN]; + char *params; +} neu_req_subscribe_t; + typedef struct { char app[NEU_NODE_NAME_LEN]; char driver[NEU_NODE_NAME_LEN]; char group[NEU_GROUP_NAME_LEN]; -} neu_req_subscribe_t, neu_req_unsubscribe_t; +} neu_req_unsubscribe_t; typedef struct neu_req_get_subscribe_group { char app[NEU_NODE_NAME_LEN]; } neu_req_get_subscribe_group_t, neu_req_get_sub_driver_tags_t; typedef struct neu_resp_subscribe_info { - char app[NEU_NODE_NAME_LEN]; - char driver[NEU_NODE_NAME_LEN]; - char group[NEU_GROUP_NAME_LEN]; + char app[NEU_NODE_NAME_LEN]; + char driver[NEU_NODE_NAME_LEN]; + char group[NEU_GROUP_NAME_LEN]; + char *params; } neu_resp_subscribe_info_t; +static inline void neu_resp_subscribe_info_fini(neu_resp_subscribe_info_t *info) +{ + free(info->params); +} + typedef struct { char driver[NEU_NODE_NAME_LEN]; char group[NEU_GROUP_NAME_LEN]; diff --git a/include/neuron/json/json.h b/include/neuron/json/json.h index 03e9b074..af2a22d5 100644 --- a/include/neuron/json/json.h +++ b/include/neuron/json/json.h @@ -88,6 +88,9 @@ void *neu_json_encode_array_value(void *array, neu_json_elem_t *t, int n); int neu_json_encode_field(void *json_object, neu_json_elem_t *elem, int n); int neu_json_encode(void *json_object, char **str); +int neu_json_dump_key(void *object, const char *key, char **const result, + bool must_exist); + #ifdef __cplusplus } #endif diff --git a/plugins/restful/group_config_handle.c b/plugins/restful/group_config_handle.c index c020bebb..7d010934 100644 --- a/plugins/restful/group_config_handle.c +++ b/plugins/restful/group_config_handle.c @@ -208,7 +208,9 @@ void handle_grp_subscribe(nng_aio *aio) strcpy(cmd.app, req->app); strcpy(cmd.driver, req->driver); strcpy(cmd.group, req->group); - ret = neu_plugin_op(plugin, header, &cmd); + cmd.params = req->params; // ownership moved + req->params = NULL; + ret = neu_plugin_op(plugin, header, &cmd); if (ret != 0) { NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, { neu_http_response(aio, NEU_ERR_IS_BUSY, result_error); @@ -284,6 +286,7 @@ void handle_grp_get_subscribe_resp(nng_aio * aio, int index = utarray_eltidx(groups->groups, group); sub_grp_configs.groups[index].driver = group->driver; sub_grp_configs.groups[index].group = group->group; + sub_grp_configs.groups[index].params = group->params; } neu_json_encode_by_fn(&sub_grp_configs, neu_json_encode_get_subscribe_resp, diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 180291b0..51e751ab 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -385,6 +385,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_RESP_GET_NODE_STATE: case NEU_RESP_GET_NODES_STATE: case NEU_RESP_GET_NODE_SETTING: + case NEU_REQ_SUBSCRIBE_GROUP: + case NEU_REQ_UNSUBSCRIBE_GROUP: case NEU_RESP_READ_GROUP: case NEU_RESP_GET_SUBSCRIBE_GROUP: case NEU_RESP_ADD_TAG: diff --git a/src/core/manager.c b/src/core/manager.c index acdc9778..103c3260 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -400,10 +400,11 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_req_subscribe_t *cmd = (neu_req_subscribe_t *) &header[1]; neu_resp_error_t error = { 0 }; - error.error = - neu_manager_subscribe(manager, cmd->app, cmd->driver, cmd->group); + error.error = neu_manager_subscribe(manager, cmd->app, cmd->driver, + cmd->group, cmd->params); if (error.error == NEU_ERR_SUCCESS) { + forward_msg(manager, msg, cmd->app); manager_storage_subscribe(manager, cmd->app, cmd->driver, cmd->group); } @@ -421,6 +422,7 @@ 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) { + forward_msg(manager, msg, cmd->app); manager_storage_unsubscribe(manager, cmd->app, cmd->driver, cmd->group); } @@ -433,7 +435,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_REQ_GET_SUBSCRIBE_GROUP: { neu_req_get_subscribe_group_t *cmd = (neu_req_get_subscribe_group_t *) &header[1]; - UT_array *groups = neu_manager_get_sub_group(manager, cmd->app); + UT_array *groups = + neu_manager_get_sub_group_deep_copy(manager, cmd->app); neu_resp_get_subscribe_group_t resp = { .groups = groups }; strcpy(header->receiver, header->sender); diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c index 0835393a..8845cec3 100644 --- a/src/core/manager_internal.c +++ b/src/core/manager_internal.c @@ -139,7 +139,8 @@ UT_array *neu_manager_get_driver_group(neu_manager_t *manager) } int neu_manager_subscribe(neu_manager_t *manager, const char *app, - const char *driver, const char *group) + const char *driver, const char *group, + const char *params) { int ret = NEU_ERR_SUCCESS; nng_pipe pipe = { 0 }; @@ -168,7 +169,7 @@ int neu_manager_subscribe(neu_manager_t *manager, const char *app, pipe = neu_node_manager_get_pipe(manager->node_manager, app); return neu_subscribe_manager_sub(manager->subscribe_manager, driver, app, - group, pipe); + group, params, pipe); } int neu_manager_unsubscribe(neu_manager_t *manager, const char *app, @@ -183,6 +184,24 @@ UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app) return neu_subscribe_manager_get(manager->subscribe_manager, app); } +UT_array *neu_manager_get_sub_group_deep_copy(neu_manager_t *manager, + const char * app) +{ + UT_array *subs = neu_subscribe_manager_get(manager->subscribe_manager, app); + + utarray_foreach(subs, neu_resp_subscribe_info_t *, sub) + { + if (sub->params) { + sub->params = strdup(sub->params); + } + } + + // set vector element destructor + subs->icd.dtor = (void (*)(void *)) neu_resp_subscribe_info_fini; + + return subs; +} + 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 bda25c49..9669d0b0 100644 --- a/src/core/manager_internal.h +++ b/src/core/manager_internal.h @@ -58,10 +58,13 @@ UT_array *neu_manager_get_nodes(neu_manager_t *manager, neu_node_type_e type, UT_array *neu_manager_get_driver_group(neu_manager_t *manager); int neu_manager_subscribe(neu_manager_t *manager, const char *app, - const char *driver, const char *group); + const char *driver, const char *group, + const char *params); 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); +UT_array *neu_manager_get_sub_group_deep_copy(neu_manager_t *manager, + const char * app); 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 a7918a5f..c706b7d7 100644 --- a/src/core/storage.c +++ b/src/core/storage.c @@ -147,7 +147,8 @@ int manager_load_subscribe(neu_manager_t *manager) utarray_foreach(sub_infos, neu_persist_subscription_info_t *, info) { rv = neu_manager_subscribe(manager, node->node, - info->driver_name, info->group_name); + info->driver_name, info->group_name, + NULL); const char *ok_or_err = (0 == rv) ? "success" : "fail"; nlog_info("%s load subscription app:%s driver:%s grp:%s", ok_or_err, node->node, info->driver_name, diff --git a/src/core/subscribe.c b/src/core/subscribe.c index 5c79a29c..fdb2b8d7 100644 --- a/src/core/subscribe.c +++ b/src/core/subscribe.c @@ -58,6 +58,10 @@ void neu_subscribe_manager_destroy(neu_subscribe_mgr_t *mgr) HASH_ITER(hh, mgr->ss, el, tmp) { HASH_DEL(mgr->ss, el); + utarray_foreach(el->apps, neu_app_subscribe_t *, sub_app) + { + neu_app_subscribe_fini(sub_app); + } utarray_free(el->apps); free(el); } @@ -118,6 +122,7 @@ UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app) strncpy(info.driver, el->key.driver, sizeof(info.driver)); strncpy(info.app, app, sizeof(info.app)); strncpy(info.group, el->key.group, sizeof(info.group)); + info.params = sub_app->params; // borrowed reference utarray_push_back(groups, &info); } @@ -145,7 +150,8 @@ 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, nng_pipe pipe) + const char *app, const char *group, + const char *params, nng_pipe pipe) { sub_elem_t * find = NULL; sub_elem_key_t key = { 0 }; @@ -156,12 +162,17 @@ int neu_subscribe_manager_sub(neu_subscribe_mgr_t *mgr, const char *driver, strncpy(app_sub.app_name, app, sizeof(app_sub.app_name)); app_sub.pipe = pipe; + if (params && NULL == (app_sub.params = strdup(params))) { + return NEU_ERR_EINTERNAL; + } + HASH_FIND(hh, mgr->ss, &key, sizeof(sub_elem_key_t), find); if (find) { utarray_foreach(find->apps, neu_app_subscribe_t *, sub) { if (strcmp(sub->app_name, app) == 0) { + neu_app_subscribe_fini(&app_sub); return NEU_ERR_GROUP_ALREADY_SUBSCRIBED; } } @@ -191,6 +202,7 @@ int neu_subscribe_manager_unsub(neu_subscribe_mgr_t *mgr, const char *driver, utarray_foreach(find->apps, neu_app_subscribe_t *, sub) { if (strcmp(sub->app_name, app) == 0) { + neu_app_subscribe_fini(sub); utarray_erase(find->apps, utarray_eltidx(find->apps, sub), 1); return NEU_ERR_SUCCESS; } @@ -210,6 +222,10 @@ void neu_subscribe_manager_remove(neu_subscribe_mgr_t *mgr, const char *driver, if (strcmp(driver, el->key.driver) == 0 && (group == NULL || strcmp(group, el->key.group) == 0)) { HASH_DEL(mgr->ss, el); + utarray_foreach(el->apps, neu_app_subscribe_t *, sub_app) + { + neu_app_subscribe_fini(sub_app); + } utarray_free(el->apps); free(el); } diff --git a/src/core/subscribe.h b/src/core/subscribe.h index cdf33c4c..782e957f 100644 --- a/src/core/subscribe.h +++ b/src/core/subscribe.h @@ -29,9 +29,15 @@ typedef struct neu_subscribe_mgr neu_subscribe_mgr_t; typedef struct neu_app_subscribe { char app_name[NEU_NODE_NAME_LEN]; + char * params; nng_pipe pipe; } neu_app_subscribe_t; +static inline void neu_app_subscribe_fini(neu_app_subscribe_t *app_sub) +{ + free(app_sub->params); +} + neu_subscribe_mgr_t *neu_subscribe_manager_create(); void neu_subscribe_manager_destroy(neu_subscribe_mgr_t *mgr); @@ -44,7 +50,7 @@ UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app); 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, - nng_pipe pipe); + const char *params, nng_pipe pipe); int neu_subscribe_manager_unsub(neu_subscribe_mgr_t *mgr, const char *driver, const char *app, const char *group); void neu_subscribe_manager_remove(neu_subscribe_mgr_t *mgr, const char *driver, diff --git a/src/parser/neu_json_group_config.c b/src/parser/neu_json_group_config.c index 2536e709..084aeec2 100644 --- a/src/parser/neu_json_group_config.c +++ b/src/parser/neu_json_group_config.c @@ -25,6 +25,8 @@ #include #include +#include + #include "json/json.h" #include "neu_json_group_config.h" @@ -231,7 +233,7 @@ int neu_json_encode_get_driver_group_resp(void *json_object, void *param) return ret; } -int neu_json_encode_get_subscribe_resp(void *json_object, void *param) +int neu_json_encode_get_subscribe_resp(void *object, void *param) { int ret = 0; neu_json_get_subscribe_resp_t *resp = @@ -240,18 +242,21 @@ int neu_json_encode_get_subscribe_resp(void *json_object, void *param) void * group_array = neu_json_array(); neu_json_get_subscribe_resp_group_t *p_group = resp->groups; for (int i = 0; i < resp->n_group; i++) { - neu_json_elem_t group_elems[] = { { - .name = "driver", - .t = NEU_JSON_STR, - .v.val_str = p_group->driver, - }, - { - .name = "group", - .t = NEU_JSON_STR, - .v.val_str = p_group->group, - } }; - group_array = neu_json_encode_array(group_array, group_elems, - NEU_JSON_ELEM_SIZE(group_elems)); + json_t *ob = json_object(); + + // rely on lib jansson to check for NULL pointers + json_object_set_new(ob, "driver", json_string(p_group->driver)); + json_object_set_new(ob, "group", json_string(p_group->group)); + + if (p_group->params) { + json_t *t = json_loads(p_group->params, 0, NULL); + json_t *p = json_object_get(t, "params"); + json_object_set(ob, "params", p); + json_decref(t); + } + + json_array_append_new(group_array, ob); + p_group++; } @@ -260,12 +265,17 @@ int neu_json_encode_get_subscribe_resp(void *json_object, void *param) .t = NEU_JSON_OBJECT, .v.val_object = group_array, } }; - ret = neu_json_encode_field(json_object, resp_elems, + ret = neu_json_encode_field(object, resp_elems, NEU_JSON_ELEM_SIZE(resp_elems)); return ret; } +static inline int dump_params(void *root, char **const result) +{ + return neu_json_dump_key(root, "params", result, false); +} + int neu_json_decode_subscribe_req(char *buf, neu_json_subscribe_req_t **result) { int ret = 0; @@ -299,6 +309,11 @@ int neu_json_decode_subscribe_req(char *buf, neu_json_subscribe_req_t **result) req->group = req_elems[1].v.val_str; req->driver = req_elems[2].v.val_str; + ret = dump_params(json_obj, &req->params); + if (0 != ret) { + goto decode_fail; + } + *result = req; goto decode_exit; @@ -318,6 +333,7 @@ void neu_json_decode_subscribe_req_free(neu_json_subscribe_req_t *req) free(req->app); free(req->driver); free(req->group); + free(req->params); free(req); } diff --git a/src/parser/neu_json_group_config.h b/src/parser/neu_json_group_config.h index 169106ee..1d683988 100644 --- a/src/parser/neu_json_group_config.h +++ b/src/parser/neu_json_group_config.h @@ -82,6 +82,7 @@ int neu_json_encode_get_driver_group_resp(void *json_object, void *param); typedef struct { char *group; char *driver; + char *params; } neu_json_get_subscribe_resp_group_t; typedef struct { @@ -95,6 +96,7 @@ typedef struct { char *group; char *app; char *driver; + char *params; } neu_json_subscribe_req_t; int neu_json_decode_subscribe_req(char *buf, neu_json_subscribe_req_t **result); diff --git a/src/utils/json.c b/src/utils/json.c index 0f172c3a..36ede662 100644 --- a/src/utils/json.c +++ b/src/utils/json.c @@ -406,3 +406,29 @@ int neu_json_decode_value(void *object, neu_json_elem_t *ele) { return decode_object(object, ele); } + +int neu_json_dump_key(void *object, const char *key, char **const result, + bool must_exist) +{ + int rv = 0; + json_t *new = NULL; + json_t *value = NULL; + + value = json_object_get(object, key); + if (NULL == value) { + if (must_exist) { + return -1; + } else { + *result = NULL; + return 0; + } + } + + if (NULL == (new = json_object()) || json_object_set(new, key, value) < 0 || + NULL == (*result = json_dumps(new, JSON_COMPACT))) { + rv = -1; + } + + json_decref(new); + return rv; +}