From ccb6a79a8ef66c7e3a16f633898e4ba8e0419bfd Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Thu, 4 Jan 2024 14:00:18 +0800 Subject: [PATCH 1/3] feat(core): get subscribe group request support filtering --- include/neuron/msg.h | 8 +++++++- src/core/manager.c | 6 +++--- src/core/manager_internal.c | 10 +++++++--- src/core/manager_internal.h | 4 +++- src/core/subscribe.c | 7 +++++-- src/core/subscribe.h | 3 ++- 6 files changed, 27 insertions(+), 11 deletions(-) diff --git a/include/neuron/msg.h b/include/neuron/msg.h index 62cfc9b5..086d2a2a 100644 --- a/include/neuron/msg.h +++ b/include/neuron/msg.h @@ -457,7 +457,13 @@ neu_req_subscribe_groups_fini(neu_req_subscribe_groups_t *req) 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; + char driver[NEU_NODE_NAME_LEN]; + char group[NEU_GROUP_NAME_LEN]; +} neu_req_get_subscribe_group_t; + +typedef struct { + char app[NEU_NODE_NAME_LEN]; +} neu_req_get_sub_driver_tags_t; typedef struct neu_resp_subscribe_info { char app[NEU_NODE_NAME_LEN]; diff --git a/src/core/manager.c b/src/core/manager.c index 68fe5071..ab23316b 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -722,7 +722,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) manager_storage_del_node(manager, cmd->node); if (neu_adapter_get_type(adapter) == NEU_NA_TYPE_APP) { UT_array *subscriptions = neu_subscribe_manager_get( - manager->subscribe_manager, cmd->node); + manager->subscribe_manager, cmd->node, NULL, NULL); neu_subscribe_manager_unsub_all(manager->subscribe_manager, cmd->node); @@ -887,8 +887,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_deep_copy(manager, cmd->app); + UT_array *groups = neu_manager_get_sub_group_deep_copy( + manager, cmd->app, cmd->driver, cmd->group); 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 c9645216..f74e2b8f 100644 --- a/src/core/manager_internal.c +++ b/src/core/manager_internal.c @@ -328,13 +328,17 @@ 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) { - return neu_subscribe_manager_get(manager->subscribe_manager, app); + return neu_subscribe_manager_get(manager->subscribe_manager, app, NULL, + NULL); } UT_array *neu_manager_get_sub_group_deep_copy(neu_manager_t *manager, - const char * app) + const char * app, + const char * driver, + const char * group) { - UT_array *subs = neu_subscribe_manager_get(manager->subscribe_manager, app); + UT_array *subs = neu_subscribe_manager_get(manager->subscribe_manager, app, + driver, group); utarray_foreach(subs, neu_resp_subscribe_info_t *, sub) { diff --git a/src/core/manager_internal.h b/src/core/manager_internal.h index b2882654..e89e0b63 100644 --- a/src/core/manager_internal.h +++ b/src/core/manager_internal.h @@ -74,7 +74,9 @@ 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); + const char * app, + const char * driver, + const char * group); int neu_manager_get_node_info(neu_manager_t *manager, const char *name, neu_persist_node_info_t *info); diff --git a/src/core/subscribe.c b/src/core/subscribe.c index ba1b49fd..e2b71b7c 100644 --- a/src/core/subscribe.c +++ b/src/core/subscribe.c @@ -105,7 +105,8 @@ UT_array *neu_subscribe_manager_find_by_driver(neu_subscribe_mgr_t *mgr, return apps; } -UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app) +UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app, + const char *driver, const char *group) { sub_elem_t *el = NULL, *tmp = NULL; UT_array * groups = NULL; @@ -116,7 +117,9 @@ UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app) { utarray_foreach(el->apps, neu_app_subscribe_t *, sub_app) { - if (strcmp(sub_app->app_name, app) == 0) { + if (strcmp(sub_app->app_name, app) == 0 && + (!driver || strstr(el->key.driver, driver)) && + (!group || strstr(el->key.group, group))) { neu_resp_subscribe_info_t info = { 0 }; strncpy(info.driver, el->key.driver, sizeof(info.driver)); diff --git a/src/core/subscribe.h b/src/core/subscribe.h index c399155e..9b8d5a1d 100644 --- a/src/core/subscribe.h +++ b/src/core/subscribe.h @@ -47,7 +47,8 @@ UT_array *neu_subscribe_manager_find(neu_subscribe_mgr_t *mgr, const char *driver, const char *group); 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(neu_subscribe_mgr_t *mgr, const char *app, + const char *driver, const char *group); size_t neu_subscribe_manager_group_count(const neu_subscribe_mgr_t *mgr, const char * app); void neu_subscribe_manager_unsub_all(neu_subscribe_mgr_t *mgr, const char *app); From 3a76d3cef50302fa02e4e2ce9d52010df6c9fd94 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Thu, 4 Jan 2024 14:01:57 +0800 Subject: [PATCH 2/3] feat(restful): get subscribe group api support filtering --- plugins/restful/group_config_handle.c | 28 ++++++++++++++++++++++----- 1 file changed, 23 insertions(+), 5 deletions(-) diff --git a/plugins/restful/group_config_handle.c b/plugins/restful/group_config_handle.c index 10408cd9..59b83780 100644 --- a/plugins/restful/group_config_handle.c +++ b/plugins/restful/group_config_handle.c @@ -356,23 +356,41 @@ void handle_grp_get_subscribe(nng_aio *aio) { int ret = 0; neu_plugin_t * plugin = neu_rest_get_plugin(); - char node_name[NEU_NODE_NAME_LEN] = { 0 }; - neu_req_get_subscribe_group_t cmd = { 0 }; - neu_reqresp_head_t header = { + neu_req_get_subscribe_group_t cmd = { 0 }; + neu_reqresp_head_t header = { .ctx = aio, .type = NEU_REQ_GET_SUBSCRIBE_GROUP, }; NEU_VALIDATE_JWT(aio); - if (neu_http_get_param_str(aio, "app", node_name, sizeof(node_name)) <= 0) { + // required parameter + ret = neu_http_get_param_str(aio, "app", cmd.app, sizeof(cmd.app)); + if (ret <= 0 || (size_t) ret == sizeof(cmd.driver)) { + NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, { + neu_http_response(aio, error_code.error, result_error); + }) + return; + } + + // optional parameter + ret = neu_http_get_param_str(aio, "driver", cmd.driver, sizeof(cmd.driver)); + if (-1 == ret || (size_t) ret == sizeof(cmd.driver)) { + NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, { + neu_http_response(aio, error_code.error, result_error); + }) + return; + } + + // optional parameter + ret = neu_http_get_param_str(aio, "group", cmd.group, sizeof(cmd.group)); + if (-1 == ret || (size_t) ret == sizeof(cmd.driver)) { NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, { neu_http_response(aio, error_code.error, result_error); }) return; } - strcpy(cmd.app, node_name); ret = neu_plugin_op(plugin, header, &cmd); if (ret != 0) { NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, { From 55e09d9a64bb56600d9db57ee97759581c59eabd Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Thu, 4 Jan 2024 14:02:57 +0800 Subject: [PATCH 3/3] test: get subscribe info ft --- tests/ft/http_api/test_http_api.py | 97 ++++++++++++++++++++++++++++++ tests/ft/neuron/api.py | 4 ++ 2 files changed, 101 insertions(+) diff --git a/tests/ft/http_api/test_http_api.py b/tests/ft/http_api/test_http_api.py index d88b97c0..01524553 100644 --- a/tests/ft/http_api/test_http_api.py +++ b/tests/ft/http_api/test_http_api.py @@ -188,6 +188,103 @@ class TestHttp: assert len(config_data['settings']) == 1 assert config_data['settings'][0]['node'] == 'mqtt' + @description(given="running neuron", when="get subscribe info", then="should success") + def test_get_subscribe_info(self): + app = 'MQTT-1' + params = {'topic':'/neuron/test'} + driver_1 = 'modbus-01' + driver_2 = 'modbus-02' + group_1 = 'group-01' + group_2 = 'group-02' + try: + api.add_node_check(app, 'MQTT') + api.add_node_check(driver_1, 'Modbus TCP') + api.add_node_check(driver_2, 'Modbus TCP') + api.add_group_check(driver_1, group_1, 1000) + api.add_group_check(driver_1, group_2, 1000) + api.add_group_check(driver_2, group_1, 1000) + api.add_group_check(driver_2, group_2, 1000) + api.subscribe_group_check(app, driver_1, group_1, params=params) + api.subscribe_group_check(app, driver_1, group_2, params=params) + api.subscribe_group_check(app, driver_2, group_1, params=params) + api.subscribe_group_check(app, driver_2, group_2, params=params) + + # match all drivers + response = api.get_subscribe_group(app, driver='modbus') + assert 200 == response.status_code + groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group'])) + assert len(groups) == 4 + assert groups[0]['driver'] == driver_1 + assert groups[0]['group'] == group_1 + assert groups[1]['driver'] == driver_1 + assert groups[1]['group'] == group_2 + assert groups[2]['driver'] == driver_2 + assert groups[2]['group'] == group_1 + assert groups[3]['driver'] == driver_2 + assert groups[3]['group'] == group_2 + + # match one driver + response = api.get_subscribe_group(app, driver=driver_1) + assert 200 == response.status_code + groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group'])) + assert len(groups) == 2 + assert groups[0]['driver'] == driver_1 + assert groups[0]['group'] == group_1 + assert groups[1]['driver'] == driver_1 + assert groups[1]['group'] == group_2 + + # match none driver + response = api.get_subscribe_group(app, driver='none') + assert 200 == response.status_code + assert len(response.json()['groups']) == 0 + + # match all groups + response = api.get_subscribe_group(app, group='group') + assert 200 == response.status_code + groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group'])) + assert len(groups) == 4 + assert groups[0]['driver'] == driver_1 + assert groups[0]['group'] == group_1 + assert groups[1]['driver'] == driver_1 + assert groups[1]['group'] == group_2 + assert groups[2]['driver'] == driver_2 + assert groups[2]['group'] == group_1 + assert groups[3]['driver'] == driver_2 + assert groups[3]['group'] == group_2 + + # match one group + response = api.get_subscribe_group(app, group=group_1) + assert 200 == response.status_code + groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group'])) + assert len(groups) == 2 + assert groups[0]['driver'] == driver_1 + assert groups[0]['group'] == group_1 + assert groups[1]['driver'] == driver_2 + assert groups[1]['group'] == group_1 + + # match none group + response = api.get_subscribe_group(app, group='none') + assert 200 == response.status_code + assert len(response.json()['groups']) == 0 + + # match driver and group + response = api.get_subscribe_group(app, driver=driver_1, group=group_1) + assert 200 == response.status_code + groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group'])) + assert len(groups) == 1 + assert groups[0]['driver'] == driver_1 + assert groups[0]['group'] == group_1 + response = api.get_subscribe_group(app, driver=driver_2, group=group_2) + assert 200 == response.status_code + groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group'])) + assert len(groups) == 1 + assert groups[0]['driver'] == driver_2 + assert groups[0]['group'] == group_2 + finally: + api.del_node(app) + api.del_node(driver_1) + api.del_node(driver_2) + @description(given="running neuron", when="get mqtt shcema", then="success") def test_get_plugin_schema(self): response = api.get_plugin_schema(plugin='mqtt') diff --git a/tests/ft/neuron/api.py b/tests/ft/neuron/api.py index 18620cb3..ee9cf9a1 100644 --- a/tests/ft/neuron/api.py +++ b/tests/ft/neuron/api.py @@ -222,6 +222,10 @@ def get_plugins(): return requests.get(url=config.BASE_URL + "/api/v2/plugin", headers={"Authorization": config.default_jwt}) +def get_subscribe_group(app, driver=None, group=None): + return requests.get(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, params={"app": app, "driver": driver, "group": group}) + + @gen_check def subscribe_group(app, driver, group, params={}): return requests.post(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, json={"app": app, "driver": driver, "group": group, "params": params})