From e97e7aa04381544880f4a4f6dff324bc88ea3533 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Wed, 2 Aug 2023 18:05:56 +0800 Subject: [PATCH 1/3] fix(core): check against NULL subscription params --- src/core/subscribe.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/core/subscribe.c b/src/core/subscribe.c index 1b38c4c1..a1d466bc 100644 --- a/src/core/subscribe.c +++ b/src/core/subscribe.c @@ -239,8 +239,8 @@ int neu_subscribe_manager_update_params(neu_subscribe_mgr_t *mgr, return NEU_ERR_GROUP_NOT_SUBSCRIBE; } - char *p = strdup(params); - if (NULL == p) { + char *p = NULL; + if (params && NULL == (p = strdup(params))) { return NEU_ERR_EINTERNAL; } From d989a4b48ba0a2668328c67632488701020fa00f Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Wed, 2 Aug 2023 18:07:19 +0800 Subject: [PATCH 2/3] feat(ekuiper): handle subscription update --- plugins/ekuiper/plugin_ekuiper.c | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/plugins/ekuiper/plugin_ekuiper.c b/plugins/ekuiper/plugin_ekuiper.c index 27cfe3d0..562d6ed2 100644 --- a/plugins/ekuiper/plugin_ekuiper.c +++ b/plugins/ekuiper/plugin_ekuiper.c @@ -308,7 +308,8 @@ static int ekuiper_plugin_request(neu_plugin_t * plugin, send_data(plugin, trans_data); break; } - case NEU_REQ_SUBSCRIBE_GROUP: { + case NEU_REQ_SUBSCRIBE_GROUP: + case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: { neu_req_subscribe_t *sub_info = data; free(sub_info->params); break; From 1e296a58729369d936d2f3dbfe2b2d8121a066a8 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Wed, 2 Aug 2023 18:05:17 +0800 Subject: [PATCH 3/3] feat(mqtt): handle subscription update --- plugins/mqtt/mqtt_handle.c | 39 +++++++++++++++++++++++++++++++++++--- plugins/mqtt/mqtt_handle.h | 2 ++ plugins/mqtt/mqtt_plugin.c | 3 +++ plugins/mqtt/mqtt_plugin.h | 18 ++++++++++++++++++ 4 files changed, 59 insertions(+), 3 deletions(-) diff --git a/plugins/mqtt/mqtt_handle.c b/plugins/mqtt/mqtt_handle.c index 4d901bfb..a86a92a8 100644 --- a/plugins/mqtt/mqtt_handle.c +++ b/plugins/mqtt/mqtt_handle.c @@ -587,6 +587,7 @@ int handle_subscribe_group(neu_plugin_t *plugin, neu_req_subscribe_t *sub_info) // no parameters, try default topic topic.v.val_str = default_upload_topic(sub_info); if (NULL == topic.v.val_str) { + rv = NEU_ERR_EINTERNAL; goto end; } } else if (0 != neu_parse_param(sub_info->params, NULL, 1, &topic)) { @@ -599,9 +600,41 @@ int handle_subscribe_group(neu_plugin_t *plugin, neu_req_subscribe_t *sub_info) sub_info->group, topic.v.val_str); // topic.v.val_str ownership moved if (0 != rv) { - plog_error(plugin, "route driver:%s group:%s to topic:%s fail, `%s`", - sub_info->driver, sub_info->group, topic.v.val_str, - sub_info->params); + plog_error(plugin, "route driver:%s group:%s fail, `%s`", + sub_info->driver, sub_info->group, sub_info->params); + goto end; + } + + plog_notice(plugin, "route driver:%s group:%s to topic:%s", + sub_info->driver, sub_info->group, topic.v.val_str); + +end: + free(sub_info->params); + return rv; +} + +int handle_update_subscribe(neu_plugin_t *plugin, neu_req_subscribe_t *sub_info) +{ + int rv = 0; + + if (NULL == sub_info->params) { + rv = NEU_ERR_GROUP_PARAMETER_INVALID; + goto end; + } + + neu_json_elem_t topic = { .name = "topic", .t = NEU_JSON_STR }; + if (0 != neu_parse_param(sub_info->params, NULL, 1, &topic)) { + plog_error(plugin, "parse `%s` for topic fail", sub_info->params); + rv = NEU_ERR_GROUP_PARAMETER_INVALID; + goto end; + } + + rv = route_tbl_update(&plugin->route_tbl, sub_info->driver, sub_info->group, + topic.v.val_str); + // topic.v.val_str ownership moved + if (0 != rv) { + plog_error(plugin, "route driver:%s group:%s fail, `%s`", + sub_info->driver, sub_info->group, sub_info->params); goto end; } diff --git a/plugins/mqtt/mqtt_handle.h b/plugins/mqtt/mqtt_handle.h index 9f131713..cb4f2319 100644 --- a/plugins/mqtt/mqtt_handle.h +++ b/plugins/mqtt/mqtt_handle.h @@ -43,6 +43,8 @@ int handle_trans_data(neu_plugin_t * plugin, neu_reqresp_trans_data_t *trans_data); int handle_subscribe_group(neu_plugin_t *plugin, neu_req_subscribe_t *sub_info); +int handle_update_subscribe(neu_plugin_t * plugin, + neu_req_subscribe_t *sub_info); int handle_unsubscribe_group(neu_plugin_t * plugin, neu_req_unsubscribe_t *unsub_info); diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index 4fdc84b6..a164b6b7 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -350,6 +350,9 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, case NEU_REQ_SUBSCRIBE_GROUP: error = handle_subscribe_group(plugin, data); break; + case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: + error = handle_update_subscribe(plugin, data); + break; case NEU_REQ_UNSUBSCRIBE_GROUP: error = handle_unsubscribe_group(plugin, data); break; diff --git a/plugins/mqtt/mqtt_plugin.h b/plugins/mqtt/mqtt_plugin.h index ca5eb1f4..d500f49f 100644 --- a/plugins/mqtt/mqtt_plugin.h +++ b/plugins/mqtt/mqtt_plugin.h @@ -107,6 +107,24 @@ static inline int route_tbl_add_new(route_entry_t **tbl, const char *driver, return 0; } +// NOTE: we take ownership of `topic` +static inline int route_tbl_update(route_entry_t **tbl, const char *driver, + const char *group, char *topic) +{ + route_entry_t *find = NULL; + + find = route_tbl_get(tbl, driver, group); + if (NULL == find) { + free(topic); + return NEU_ERR_GROUP_NOT_SUBSCRIBE; + } + + free(find->topic); + find->topic = topic; + + return 0; +} + static inline void route_tbl_del(route_entry_t **tbl, const char *driver, const char *group) {