From ef636d2950f2eda2535399b7c5fc581a0b5d323f Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Tue, 5 Mar 2024 17:20:15 +0800 Subject: [PATCH 1/4] fix(core): notify app about driver group deletion --- src/adapter/adapter.c | 5 ++++- src/core/manager.c | 15 +++++++++++++++ 2 files changed, 19 insertions(+), 1 deletion(-) diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 16b0859f..7c161ae1 100644 --- a/src/adapter/adapter.c +++ b/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) { diff --git a/src/core/manager.c b/src/core/manager.c index f627cb42..6b8bbbbc 100644 --- a/src/core/manager.c +++ b/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; } From 6b24d4bf7a2588e2d6e99567df2c938a99051a74 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Tue, 5 Mar 2024 17:20:52 +0800 Subject: [PATCH 2/4] fix(mqtt): handle NEU_REQ_DEL_GROUP --- plugins/mqtt/mqtt_handle.c | 8 ++++++++ plugins/mqtt/mqtt_handle.h | 1 + plugins/mqtt/mqtt_plugin.c | 3 +++ 3 files changed, 12 insertions(+) diff --git a/plugins/mqtt/mqtt_handle.c b/plugins/mqtt/mqtt_handle.c index 7f62eacd..b249af4d 100644 --- a/plugins/mqtt/mqtt_handle.c +++ b/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, diff --git a/plugins/mqtt/mqtt_handle.h b/plugins/mqtt/mqtt_handle.h index 905944ff..ba40696e 100644 --- a/plugins/mqtt/mqtt_handle.h +++ b/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); diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index 0c7d668f..96b58607 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/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; From f542f41631d28cb13bbd09272823222cf243a165 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Tue, 5 Mar 2024 17:48:01 +0800 Subject: [PATCH 3/4] test(mqtt): data upload when driver group deleted --- tests/ft/app/test_mqtt.py | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/tests/ft/app/test_mqtt.py b/tests/ft/app/test_mqtt.py index 651afdef..d90b85d7 100644 --- a/tests/ft/app/test_mqtt.py +++ b/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", From b08cbc776169701e00d2d20c372e6be36fc59482 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Tue, 5 Mar 2024 17:21:37 +0800 Subject: [PATCH 4/4] fix(ekuiper): handle NEU_REQ_DEL_GROUP --- plugins/ekuiper/plugin_ekuiper.c | 2 ++ 1 file changed, 2 insertions(+) diff --git a/plugins/ekuiper/plugin_ekuiper.c b/plugins/ekuiper/plugin_ekuiper.c index 7b70df6b..e7cf682e 100644 --- a/plugins/ekuiper/plugin_ekuiper.c +++ b/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;