feat(mqtt): pub data to NEU_REQ_SUBSCRIBE_GROUP message specified topic

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;
+}

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

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:

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