GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-29 07:06:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(nano-mqtt): handling heartbeat generation

Browse Source
This commit is contained in:
eeff
2022-11-15 00:23:44 +08:00
parent f919f5c565
commit a8cfeaa268
1 changed files with 66 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/nano-mqtt/mqtt_handle.c
+66 -2
View File
@@ -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;
}