From afa0e52d43091f4aaddca93c9b8c8a45c1f25f74 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Tue, 10 Jan 2023 23:40:33 +0800 Subject: [PATCH] feat(mqtt): pub data to NEU_REQ_SUBSCRIBE_GROUP message specified topic --- plugins/mqtt/mqtt_handle.c | 48 ++++++++++++++++++++++- plugins/mqtt/mqtt_handle.h | 4 ++ plugins/mqtt/mqtt_plugin.c | 8 ++++ plugins/mqtt/mqtt_plugin.h | 80 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 139 insertions(+), 1 deletion(-) diff --git a/plugins/mqtt/mqtt_handle.c b/plugins/mqtt/mqtt_handle.c index f58cb5e5..3a451860 100644 --- a/plugins/mqtt/mqtt_handle.c +++ b/plugins/mqtt/mqtt_handle.c @@ -480,6 +480,14 @@ int handle_trans_data(neu_plugin_t * plugin, return NEU_ERR_MQTT_FAILURE; } + const route_entry_t *route = route_tbl_get( + &plugin->route_tbl, trans_data->driver, trans_data->group); + if (NULL == route) { + plog_error(plugin, "no route for driver:%s group:%s", + trans_data->driver, trans_data->group); + return NEU_ERR_GROUP_NOT_SUBSCRIBE; + } + char *json_str = generate_upload_json(plugin, trans_data, plugin->config.format); if (NULL == json_str) { @@ -487,10 +495,48 @@ int handle_trans_data(neu_plugin_t * plugin, return NEU_ERR_EINTERNAL; } - char * topic = plugin->config.upload_topic; + char * topic = route->topic; neu_mqtt_qos_e qos = NEU_MQTT_QOS0; rv = publish(plugin, qos, topic, json_str, strlen(json_str)); json_str = NULL; return rv; } + +int handle_subscribe_group(neu_plugin_t *plugin, neu_req_subscribe_t *sub_info) +{ + int rv = 0; + + 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_add_new(&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 to topic:%s fail, `%s`", + sub_info->driver, sub_info->group, topic.v.val_str, + sub_info->params); + goto end; + } + + plog_info(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_unsubscribe_group(neu_plugin_t * plugin, + neu_req_unsubscribe_t *unsub_info) +{ + route_tbl_del(&plugin->route_tbl, unsub_info->driver, unsub_info->group); + plog_info(plugin, "del route driver:%s group:%s", unsub_info->driver, + unsub_info->group); + return 0; +} diff --git a/plugins/mqtt/mqtt_handle.h b/plugins/mqtt/mqtt_handle.h index 81d3a58e..9f131713 100644 --- a/plugins/mqtt/mqtt_handle.h +++ b/plugins/mqtt/mqtt_handle.h @@ -42,6 +42,10 @@ int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json, 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_unsubscribe_group(neu_plugin_t * plugin, + neu_req_unsubscribe_t *unsub_info); + #ifdef __cplusplus } #endif diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index b37a6096..a4f8b2a9 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -86,6 +86,8 @@ static int mqtt_plugin_uninit(neu_plugin_t *plugin) free(plugin->write_resp_topic); plugin->write_resp_topic = NULL; + route_tbl_free(plugin->route_tbl); + plog_info(plugin, "uninitialize plugin `%s` success", neu_plugin_module.module_name); return NEU_ERR_SUCCESS; @@ -349,6 +351,12 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, case NEU_REQRESP_TRANS_DATA: error = handle_trans_data(plugin, data); break; + case NEU_REQ_SUBSCRIBE_GROUP: + error = handle_subscribe_group(plugin, data); + break; + case NEU_REQ_UNSUBSCRIBE_GROUP: + error = handle_unsubscribe_group(plugin, data); + break; case NEU_REQRESP_NODE_DELETED: break; default: diff --git a/plugins/mqtt/mqtt_plugin.h b/plugins/mqtt/mqtt_plugin.h index e6b06d78..931bd32e 100644 --- a/plugins/mqtt/mqtt_plugin.h +++ b/plugins/mqtt/mqtt_plugin.h @@ -29,6 +29,19 @@ extern "C" { #include "mqtt_config.h" +typedef struct { + char driver[NEU_NODE_NAME_LEN]; + char group[NEU_GROUP_NAME_LEN]; +} route_key_t; + +typedef struct { + route_key_t key; + + char *topic; + + UT_hash_handle hh; +} route_entry_t; + struct neu_plugin { neu_plugin_common_t common; mqtt_config_t config; @@ -38,8 +51,75 @@ struct neu_plugin { char * read_resp_topic; char * write_req_topic; char * write_resp_topic; + route_entry_t * route_tbl; }; +static inline void route_entry_free(route_entry_t *e) +{ + free(e->topic); + free(e); +} + +static inline void route_tbl_free(route_entry_t *tbl) +{ + route_entry_t *e = NULL, *tmp = NULL; + HASH_ITER(hh, tbl, e, tmp) + { + HASH_DEL(tbl, e); + route_entry_free(e); + } +} + +static inline route_entry_t * +route_tbl_get(route_entry_t **tbl, const char *driver, const char *group) +{ + route_entry_t *find = NULL; + route_key_t key = { 0 }; + + strncpy(key.driver, driver, sizeof(key.driver)); + strncpy(key.group, group, sizeof(key.group)); + + HASH_FIND(hh, *tbl, &key, sizeof(key), find); + return find; +} + +// NOTE: we take ownership of `topic` +static inline int route_tbl_add_new(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 (find) { + free(topic); + return NEU_ERR_GROUP_ALREADY_SUBSCRIBED; + } + + find = calloc(1, sizeof(*find)); + if (NULL == find) { + free(topic); + return NEU_ERR_EINTERNAL; + } + + strncpy(find->key.driver, driver, sizeof(find->key.driver)); + strncpy(find->key.group, group, sizeof(find->key.group)); + find->topic = topic; + HASH_ADD(hh, *tbl, key, sizeof(find->key), find); + + return 0; +} + +static inline void route_tbl_del(route_entry_t **tbl, const char *driver, + const char *group) +{ + route_entry_t *find = NULL; + find = route_tbl_get(tbl, driver, group); + if (find) { + HASH_DEL(*tbl, find); + route_entry_free(find); + } +} + #ifdef __cplusplus } #endif