Merge pull request #1709 from eeff/eeff/dev

fix(core): send unsubscribe group message when deleting app

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);

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)

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})