Merge pull request #1870 from eeff/v2.7

Pick to v2.7 fix: notify north app about driver group deletion

plugins/ekuiper/plugin_ekuiper.c

@@ -340,6 +340,8 @@ static int ekuiper_plugin_request(neu_plugin_t *      plugin,
     case NEU_REQ_UPDATE_GROUP: {
         break;
     }
+    case NEU_REQ_DEL_GROUP:
+        break;
     case NEU_REQ_SUBSCRIBE_GROUP:
     case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: {
         neu_req_subscribe_t *sub_info = data;

plugins/mqtt/mqtt_handle.c

@@ -606,6 +606,14 @@ int handle_unsubscribe_group(neu_plugin_t *         plugin,
     return 0;
 }
 
+int handle_del_group(neu_plugin_t *plugin, neu_req_del_group_t *req)
+{
+    route_tbl_del(&plugin->route_tbl, req->driver, req->group);
+    plog_notice(plugin, "del route driver:%s group:%s", req->driver,
+                req->group);
+    return 0;
+}
+
 int handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req)
 {
     route_tbl_update_group(&plugin->route_tbl, req->driver, req->group,

plugins/mqtt/mqtt_handle.h

@@ -49,6 +49,7 @@ int handle_unsubscribe_group(neu_plugin_t *         plugin,
                              neu_req_unsubscribe_t *unsub_info);
 
 int handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req);
+int handle_del_group(neu_plugin_t *plugin, neu_req_del_group_t *cmd);
 
 int handle_update_driver(neu_plugin_t *plugin, neu_req_update_node_t *req);
 int handle_del_driver(neu_plugin_t *plugin, neu_reqresp_node_deleted_t *req);

plugins/mqtt/mqtt_plugin.c

@@ -384,6 +384,9 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
     case NEU_REQ_UPDATE_GROUP:
         error = handle_update_group(plugin, data);
         break;
+    case NEU_REQ_DEL_GROUP:
+        error = handle_del_group(plugin, data);
+        break;
     case NEU_REQ_UPDATE_NODE:
         error = handle_update_driver(plugin, data);
         break;

src/adapter/adapter.c

@@ -919,7 +919,10 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
             error.error = neu_adapter_driver_del_group(
                 (neu_adapter_driver_t *) adapter, cmd->group);
         } else {
-            error.error = NEU_ERR_GROUP_NOT_ALLOW;
+            adapter->module->intf_funs->request(
+                adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
+            neu_msg_free(msg);
+            break;
         }
 
         if (error.error == NEU_ERR_SUCCESS) {

src/core/manager.c

@@ -1370,9 +1370,24 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
             neu_msg_exchange(header);
             reply(manager, header, &e);
         } else {
+            UT_array *apps = neu_subscribe_manager_find(
+                manager->subscribe_manager, cmd->driver, cmd->group);
+
             forward_msg(manager, header, header->receiver);
             neu_subscribe_manager_remove(manager->subscribe_manager,
                                          cmd->driver, cmd->group);
+
+            if (NULL == apps) {
+                break;
+            }
+
+            // notify app node about group deletion
+            utarray_foreach(apps, neu_app_subscribe_t *, app)
+            {
+                forward_msg_copy(manager, header, app->app_name);
+            }
+
+            utarray_free(apps);
         }
         break;
     }

tests/ft/app/test_mqtt.py

@@ -490,6 +490,36 @@ class TestMQTT:
         finally:
             api.update_group_check(DRIVER, new_name, GROUP)
 
+    @description(
+        given="MQTT node",
+        when="delete subscribed driver group and add back group",
+        then="broker should receive data on upload topic",
+    )
+    def test_mqtt_upload_when_group_deleted(self, mocker, conf_fmt_tags):
+        api.node_setting_check(NODE, conf_fmt_tags)
+        api.node_ctl(NODE, config.NEU_CTL_START)
+
+        # delete driver group
+        api.del_group_check(DRIVER, GROUP)
+        # add back driver group
+        api.add_group_check(DRIVER, GROUP, INTERVAL)
+        api.add_tags_check(DRIVER, GROUP, tags=[TAG0])
+        # add back subscription on new topic
+        topic = UPLOAD_TOPIC + "-new"
+        params = {"topic": topic}
+        api.subscribe_group_check(NODE, DRIVER, GROUP, params)
+
+        msg = mocker.get(topic, timeout=3)
+        try:
+            assert msg is not None
+            assert DRIVER == msg["node"]
+            assert GROUP == msg["group"]
+            assert TAG0["name"] in [tag["name"] for tag in msg["tags"]]
+        finally:
+            params["topic"] = UPLOAD_TOPIC
+            api.unsubscribe_group(NODE, DRIVER, GROUP)
+            api.subscribe_group_check(NODE, DRIVER, GROUP, params)
+
     @description(
         given="MQTT node and conf_cache",
         when="broken network connection restored",