diff --git a/src/core/manager.c b/src/core/manager.c index 9a3bcdb3..0178e202 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -985,8 +985,25 @@ 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); neu_subscribe_manager_unsub_all(manager->subscribe_manager, cmd->node); + + utarray_foreach(subscriptions, neu_resp_subscribe_info_t *, sub) + { + // NOTE: neu_req_unsubscribe_t and neu_resp_subscribe_info_t + // have compatible memory layout + msg = neu_msg_new(NEU_REQ_UNSUBSCRIBE_GROUP, NULL, sub); + if (NULL == msg) { + break; + } + neu_reqresp_head_t *hd = neu_msg_get_header(msg); + strcpy(hd->receiver, sub->driver); + strcpy(hd->sender, "manager"); + forward_msg(manager, hd, hd->receiver); + } + utarray_free(subscriptions); } neu_reqresp_node_deleted_t resp = { 0 }; strcpy(resp.node, header->receiver); diff --git a/tests/ft/app/test_app.py b/tests/ft/app/test_app.py index 65291ef8..59c8763a 100644 --- a/tests/ft/app/test_app.py +++ b/tests/ft/app/test_app.py @@ -24,6 +24,23 @@ class TestDriver: assert 200 == response.status_code assert error.NEU_ERR_SUCCESS == response.json()['error'] + @description(given="app node with subscriptions", when="delete node", then="should success") + @pytest.mark.parametrize('app,plugin', [('mqtt', config.PLUGIN_MQTT)]) + def test_delete_app(self, app, plugin): + driver = 'modbus' + group = 'group' + try: + api.add_node_check(node=app, plugin=plugin) + api.add_node_check(node=driver, plugin=config.PLUGIN_MODBUS_TCP) + api.modbus_tcp_node_setting(node=driver, port=502) + api.add_tags_check(node=driver, group=group, tags=hold_int16) + api.subscribe_group_check(app=app, driver=driver, group=group) + response = api.del_node(app) + assert 200 == response.status_code + assert error.NEU_ERR_SUCCESS == response.json()['error'] + finally: + api.del_node(driver) + @description(given="modbus tcp node", when="setting modbus tcp node and groups/tags", then="setting success") def test_modbus_tcp_setting(self): response = api.modbus_tcp_node_setting(node='modbus-tcp-1', port=502) diff --git a/tests/ft/neuron/api.py b/tests/ft/neuron/api.py index 8784940a..4e27b96a 100644 --- a/tests/ft/neuron/api.py +++ b/tests/ft/neuron/api.py @@ -284,12 +284,15 @@ def del_template_tags(tmpl, group, tags): def get_template_tags(tmpl, group, name): return requests.get(url=config.BASE_URL + "/api/v2/template/tag", headers={"Authorization": config.default_jwt}, params={"template": tmpl, "group": group, "name": name}) -def subscribe_group(app, driver, group): - return requests.post(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, json={"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}) +@gen_check def subscribe_groups(app, groups): return requests.post(url=config.BASE_URL + "/api/v2/subscribes", headers={"Authorization": config.default_jwt}, json={"app": app, "groups": groups}) +@gen_check def unsubscribe_group(app, driver, group): return requests.delete(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, json={"app": app, "driver": driver, "group": group})