feat(nano-mqtt): handling read request and response generation

plugins/nano-mqtt/mqtt_handle.c

@@ -17,8 +17,11 @@
  * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  **/
 
+#include "connection/mqtt_client.h"
 #include "errcodes.h"
 #include "version.h"
+#include "json/neu_json_mqtt.h"
+#include "json/neu_json_rw.h"
 
 #include "mqtt_handle.h"
 #include "mqtt_plugin.h"
@@ -176,6 +179,49 @@ static char *generate_upload_json(neu_plugin_t *            plugin,
     return json_str;
 }
 
+static char *generate_read_resp_json(neu_plugin_t *         plugin,
+                                     neu_json_mqtt_t *      mqtt,
+                                     neu_resp_read_group_t *data)
+{
+    neu_resp_tag_value_t *tags     = data->tags;
+    uint16_t              len      = data->n_tag;
+    char *                json_str = NULL;
+    neu_json_read_resp_t  json     = { 0 };
+
+    if (0 != tag_values_to_json(tags, len, &json)) {
+        plog_error(plugin, "tag_values_to_json fail");
+        return NULL;
+    }
+
+    neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, mqtt,
+                              neu_json_encode_mqtt_resp, &json_str);
+
+    if (json.tags) {
+        free(json.tags);
+    }
+    return json_str;
+}
+
+static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
+                                neu_json_read_req_t *req)
+{
+    plog_info(plugin, "read uuid:%s, group:%s, node:%s", mqtt->uuid, req->group,
+              req->node);
+
+    neu_reqresp_head_t header = { 0 };
+    header.ctx                = mqtt;
+    header.type               = NEU_REQ_READ_GROUP;
+    neu_req_read_group_t cmd  = { 0 };
+    strcpy(cmd.driver, req->node);
+    strcpy(cmd.group, req->group);
+    if (0 != neu_plugin_op(plugin, header, &cmd)) {
+        plog_error(plugin, "neu_plugin_op(NEU_REQ_READ_GROUP) fail");
+        return -1;
+    }
+
+    return 0;
+}
+
 int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
                           neu_resp_error_t *data)
 {
@@ -185,13 +231,83 @@ int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
     return 0;
 }
 
+void handle_read_req(const uint8_t *payload, uint32_t len, void *data)
+{
+    int                  rv     = 0;
+    neu_plugin_t *       plugin = data;
+    neu_json_read_req_t *req    = NULL;
+
+    char *json_str = malloc(len + 1);
+    if (NULL == json_str) {
+        return;
+    }
+
+    memcpy(json_str, payload, len);
+    json_str[len] = '\0';
+
+    neu_json_mqtt_t *mqtt = NULL;
+    rv                    = neu_json_decode_mqtt_req(json_str, &mqtt);
+    if (0 != rv) {
+        plog_error(plugin, "neu_json_decode_mqtt_req failed");
+        free(json_str);
+        return;
+    }
+
+    rv = neu_json_decode_read_req(json_str, &req);
+    if (0 != rv) {
+        plog_error(plugin, "neu_json_decode_read_req fail");
+        neu_json_decode_mqtt_req_free(mqtt);
+        free(json_str);
+        return;
+    }
+
+    rv = send_read_req(plugin, mqtt, req);
+    if (0 != rv) {
+        neu_json_decode_mqtt_req_free(mqtt);
+    }
+
+    neu_json_decode_read_req_free(req);
+    free(json_str);
+}
+
 int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
                          neu_resp_read_group_t *data)
 {
-    (void) plugin;
-    (void) mqtt_json;
-    (void) data;
-    return 0;
+    int   rv       = 0;
+    char *json_str = NULL;
+
+    if (NULL == plugin->client) {
+        rv = NEU_ERR_MQTT_IS_NULL;
+        goto end;
+    }
+
+    if (!neu_mqtt_client_is_connected(plugin->client)) {
+        rv = NEU_ERR_MQTT_FAILURE;
+        goto end;
+    }
+
+    json_str = generate_read_resp_json(plugin, mqtt_json, data);
+    if (NULL == json_str) {
+        plog_error(plugin, "generate read resp json fail");
+        rv = NEU_ERR_EINTERNAL;
+        goto end;
+    }
+
+    const char *   topic = plugin->read_resp_topic;
+    neu_mqtt_qos_e qos   = NEU_MQTT_QOS0;
+    rv = neu_mqtt_client_publish(plugin->client, qos, topic, json_str,
+                                 strlen(json_str));
+    if (0 != rv) {
+        plog_error(plugin, "pub [%s, QoS%d] error", topic, qos);
+        rv = NEU_ERR_MQTT_PUBLISH_FAILURE;
+        goto end;
+    }
+
+end:
+    neu_json_decode_mqtt_req_free(mqtt_json);
+    free(json_str);
+    free(data->tags);
+    return rv;
 }
 
 int handle_trans_data(neu_plugin_t *            plugin,

plugins/nano-mqtt/mqtt_handle.h

@@ -29,6 +29,8 @@ extern "C" {
 int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
                           neu_resp_error_t *data);
 
+void handle_read_req(const uint8_t *payload, uint32_t len, void *data);
+
 int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
                          neu_resp_read_group_t *data);
 

plugins/nano-mqtt/mqtt_plugin.c

@@ -17,9 +17,11 @@
  * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  **/
 
-#include "mqtt_plugin.h"
+#include "utils/asprintf.h"
+
 #include "mqtt_config.h"
 #include "mqtt_handle.h"
+#include "mqtt_plugin.h"
 
 const neu_plugin_module_t neu_plugin_module;
 
@@ -69,6 +71,12 @@ static int mqtt_plugin_uninit(neu_plugin_t *plugin)
         neu_mqtt_client_free(plugin->client);
         plugin->client = NULL;
     }
+
+    free(plugin->read_req_topic);
+    plugin->read_req_topic = NULL;
+    free(plugin->read_resp_topic);
+    plugin->read_resp_topic = NULL;
+
     plog_info(plugin, "uninitialize plugin `%s` success",
               neu_plugin_module.module_name);
     return NEU_ERR_SUCCESS;
@@ -185,6 +193,30 @@ error:
     return rv;
 }
 
+static int create_topic(neu_plugin_t *plugin)
+{
+    if (plugin->read_req_topic) {
+        // topics already created
+        return 0;
+    }
+
+    neu_asprintf(&plugin->read_req_topic, "/neuron/%s/read/req",
+                 plugin->common.name);
+    if (NULL == plugin->read_req_topic) {
+        return -1;
+    }
+
+    neu_asprintf(&plugin->read_resp_topic, "/neuron/%s/read/resp",
+                 plugin->common.name);
+    if (NULL == plugin->read_resp_topic) {
+        free(plugin->read_req_topic);
+        plugin->read_req_topic = NULL;
+        return -1;
+    }
+
+    return 0;
+}
+
 static int mqtt_plugin_start(neu_plugin_t *plugin)
 {
     int         rv          = 0;
@@ -202,6 +234,20 @@ static int mqtt_plugin_start(neu_plugin_t *plugin)
         goto end;
     }
 
+    if (0 != create_topic(plugin)) {
+        plog_error(plugin, "create topics fail");
+        rv = NEU_ERR_EINTERNAL;
+        goto end;
+    }
+
+    if (0 !=
+        neu_mqtt_client_subscribe(plugin->client, 0, plugin->read_req_topic,
+                                  handle_read_req, plugin)) {
+        plog_error(plugin, "subscribe [%s] fail", plugin->read_req_topic);
+        rv = NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
+        goto end;
+    }
+
 end:
     if (0 == rv) {
         plog_info(plugin, "start plugin `%s` success", plugin_name);

plugins/nano-mqtt/mqtt_plugin.h

@@ -33,6 +33,8 @@ struct neu_plugin {
     neu_plugin_common_t common;
     mqtt_config_t       config;
     neu_mqtt_client_t * client;
+    char *              read_req_topic;
+    char *              read_resp_topic;
 };
 
 #ifdef __cplusplus