From a8cfeaa268a8e20146898c4c8bd2be470afe5a4b Mon Sep 17 00:00:00 2001
From: eeff <33566201+eeff@users.noreply.github.com>
Date: Mon, 14 Nov 2022 23:54:28 +0800
Subject: [PATCH] feat(nano-mqtt): handling heartbeat generation

---
 plugins/nano-mqtt/mqtt_handle.c | 68 ++++++++++++++++++++++++++++++++-
 1 file changed, 66 insertions(+), 2 deletions(-)

diff --git a/plugins/nano-mqtt/mqtt_handle.c b/plugins/nano-mqtt/mqtt_handle.c
index 2e809f27..91659f51 100644
--- a/plugins/nano-mqtt/mqtt_handle.c
+++ b/plugins/nano-mqtt/mqtt_handle.c
@@ -17,9 +17,42 @@
  * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  **/
 
+#include "errcodes.h"
+#include "version.h"
+
 #include "mqtt_handle.h"
 #include "mqtt_plugin.h"
 
+static char *generate_heartbeat_json(neu_plugin_t *plugin, UT_array *states)
+{
+    (void) plugin;
+    char *                 version  = NEURON_VERSION;
+    neu_json_states_head_t header   = { .version   = version,
+                                      .timpstamp = global_timestamp };
+    neu_json_states_t      json     = { 0 };
+    char *                 json_str = NULL;
+
+    json.n_state = utarray_len(states);
+    json.states  = calloc(json.n_state, sizeof(neu_json_node_state_t));
+    if (NULL == json.states) {
+        return NULL;
+    }
+
+    utarray_foreach(states, neu_nodes_state_t *, state)
+    {
+        int index                  = utarray_eltidx(states, state);
+        json.states[index].node    = state->node;
+        json.states[index].link    = state->state.link;
+        json.states[index].running = state->state.running;
+    }
+
+    neu_json_encode_with_mqtt(&json, neu_json_encode_states_resp, &header,
+                              neu_json_encode_state_header_resp, &json_str);
+
+    free(json.states);
+    return json_str;
+}
+
 int handle_write_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
                           neu_resp_error_t *data)
 {
@@ -48,7 +81,38 @@ int handle_trans_data(neu_plugin_t *            plugin,
 
 int handle_nodes_state(neu_plugin_t *plugin, neu_reqresp_nodes_state_t *states)
 {
-    (void) plugin;
+    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_heartbeat_json(plugin, states->states);
+    if (NULL == json_str) {
+        plog_error(plugin, "generate heartbeat json fail");
+        rv = NEU_ERR_EINTERNAL;
+        goto end;
+    }
+
+    const char *   topic = plugin->config.heartbeat_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, "heartbeat pub [%s, QoS%d] error", topic, qos);
+        rv = NEU_ERR_MQTT_PUBLISH_FAILURE;
+    }
+
+end:
+    free(json_str);
     utarray_free(states->states);
-    return 0;
+
+    return rv;
 }