GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-19 14:23:05 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(mqtt): state json string send

Browse Source
This commit is contained in:
gc87
2022-07-13 16:00:04 +08:00
parent 1f9e3ea0db
commit 596cd699fd
10 changed files with 154 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ft/resource/common.resource
+1 -1
View File
@@ -35,7 +35,7 @@ ${TOPIC_UPLOAD} neuron/${MQTT_CLIENTID}/upload
${MQTT_UUID} 123456-789abc-123456-aabbcc
${MQTT_CONFIG_HOST} 127.0.0.1
${MQTT_CONFIG_PORT} ${1883}
${MQTT_CONFIG} {"client-id": "${MQTT_CLIENTID}", "upload-topic": "", "format": 1,
${MQTT_CONFIG} {"client-id": "${MQTT_CLIENTID}", "upload-topic": "", "heartbeat-topic": "", "format": 1,
... "ssl": false, "host": "${MQTT_CONFIG_HOST}", "port": ${MQTT_CONFIG_PORT},
... "username": "", "password": "", "ca":"", "cert":"", "key":"", "keypass":""}
include/neuron/json/neu_json_mqtt.h
+19
View File
@@ -35,11 +35,30 @@ typedef struct {
char *uuid;
} neu_json_mqtt_t;
typedef struct {
char *node;
int link;
int running;
} neu_json_node_state_t;
typedef struct {
char * version;
uint32_t timpstamp;
} neu_json_states_head_t;
typedef struct {
int n_state;
neu_json_node_state_t *states;
} neu_json_states_t;
int neu_json_decode_mqtt_req(char *buf, neu_json_mqtt_t **result);
void neu_json_decode_mqtt_req_free(neu_json_mqtt_t *req);
int neu_json_encode_mqtt_resp(void *json_object, void *param);
int neu_json_encode_state_header_resp(void *json_object, void *param);
int neu_json_encode_states_resp(void *json_object, void *param);
#ifdef __cplusplus
}
#endif
plugins/mqtt/command/command.c
+2 -2
View File
@@ -131,7 +131,7 @@ char *command_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head,
return command_rw_write_response(plugin, head, data);
}
char *command_heartbeat_response(neu_plugin_t *plugin)
char *command_heartbeat_response(neu_plugin_t *plugin, UT_array *states)
{
return command_heartbeat_generate(plugin);
return command_heartbeat_generate(plugin, states);
}
plugins/mqtt/command/command.h
+1 -1
View File
@@ -57,7 +57,7 @@ char *command_read_periodic_response(neu_plugin_t * plugin,
char *command_write_response(neu_plugin_t *plugin, neu_reqresp_head_t *head,
neu_resp_error_t *data);
char *command_heartbeat_response(neu_plugin_t *plugin);
char *command_heartbeat_response(neu_plugin_t *plugin, UT_array *states);
#ifdef __cplusplus
}
plugins/mqtt/command/heartbeat.c
+24 -6
View File
@@ -26,13 +26,31 @@
#include "utils/log.h"
#include "version.h"
char *command_heartbeat_generate(neu_plugin_t *plugin)
char *command_heartbeat_generate(neu_plugin_t *plugin, UT_array *states)
{
UNUSED(plugin);
char * version = NEURON_VERSION;
neu_json_states_head_t header = { .version = version,
.timpstamp = plugin->common.timestamp };
neu_json_states_t json = { 0 };
char * json_str = NULL;
json.n_state = utarray_len(states);
if (0 < json.n_state) {
json.states = calloc(json.n_state, sizeof(neu_json_node_state_t));
}
// const char *version = NEURON_VERSION;
// const char *revision = NEURON_GIT_REV NEURON_GIT_DIFF;
// const char * build_date = NEURON_BUILD_DATE;
for (int i = 0; i < json.n_state; i++) {
struct node_state **nsp = utarray_eltptr(states, (unsigned int) i);
struct node_state * ns = (*nsp);
json.states[i].node = ns->node;
json.states[i].link = ns->link;
json.states[i].running = ns->running;
}
return NULL;
neu_json_encode_with_mqtt(&json, neu_json_encode_states_resp, &header,
neu_json_encode_state_header_resp, &json_str);
if (0 < json.n_state) {
free(json.states);
}
return json_str;
}
plugins/mqtt/command/heartbeat.h
+1 -1
View File
@@ -28,7 +28,7 @@ extern "C" {
#include "common.h"
char *command_heartbeat_generate(neu_plugin_t *plugin);
char *command_heartbeat_generate(neu_plugin_t *plugin, UT_array *states);
#ifdef __cplusplus
}
plugins/mqtt/mqtt.c
+34 -29
View File
@@ -50,19 +50,13 @@ struct topic_pair {
int type;
};
struct node_state {
char node[NEU_NODE_NAME_LEN];
char plugin[NEU_PLUGIN_NAME_LEN];
int running;
int link;
};
struct mqtt_routine {
neu_plugin_t * plugin;
neu_mqtt_option_t option;
neu_mqtt_client_t client;
UT_array * topics;
nng_aio * aio;
nng_mtx * mutex;
UT_array * states;
};
@@ -141,6 +135,7 @@ static void topics_generate(UT_array *topics, char *name, char *upload_topic,
char *heartbeat_req = NULL;
char *heartbeat_res = NULL;
if (NULL != heartbeat_topic && 0 < strlen(heartbeat_topic)) {
nlog_debug("------------->%s", heartbeat_topic);
heartbeat_res = strdup(heartbeat_topic);
} else {
heartbeat_res = topics_format(TOPIC_HEARTBEAT_RES, name);
@@ -243,28 +238,33 @@ static void mqtt_routine_heartbeat(nng_aio *aio, void *arg, int code)
ret = neu_plugin_op(routine->plugin, header, &cmd);
if (ret != 0) { }
nng_aio_set_timeout(aio, 3000);
nng_aio_defer(aio, mqtt_routine_heartbeat, routine);
// publish state data
int type = TOPIC_TYPE_HEARTBEAT;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char * json_str = "test";
const char * topic = pair->topic_response;
const int qos = pair->qos_response;
neu_err_code_e error = neu_mqtt_client_publish(
nng_mtx_lock(routine->mutex);
int type = TOPIC_TYPE_HEARTBEAT;
struct topic_pair *pair = topics_find_type(routine->topics, type);
char * json_str =
command_heartbeat_response(routine->plugin, routine->states);
if (NULL == json_str) {
nng_mtx_unlock(routine->mutex);
break;
}
const char * topic = pair->topic_response;
const int qos = pair->qos_response;
neu_err_code_e error = neu_mqtt_client_publish(
routine->client, topic, qos, (unsigned char *) json_str,
strlen(json_str));
if (NEU_ERR_SUCCESS != error) {
plog_error(plugin_log,
"heartbeat publish error code :%d, topoic:%s", error,
topic);
// free(json_str);
}
plog_info(plugin_log, "topic:%s, qos:%d, payload:%s", topic, qos,
json_str);
// free(json_str);
nng_aio_set_timeout(aio, 1000);
nng_aio_defer(aio, mqtt_routine_heartbeat, routine);
free(json_str);
nng_mtx_unlock(routine->mutex);
break;
}
case NNG_ECANCELED: {
@@ -323,9 +323,10 @@ static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin,
utarray_new(routine->states, &state_ptr_icd);
// start heartbeat send
nng_mtx_alloc(&routine->mutex);
nng_aio_alloc(&routine->aio, NULL, routine);
nng_aio_wait(routine->aio);
nng_aio_set_timeout(routine->aio, 1000);
nng_aio_set_timeout(routine->aio, 3000);
nng_aio_defer(routine->aio, mqtt_routine_heartbeat, routine);
return routine;
@@ -335,9 +336,18 @@ static void mqtt_routine_stop(mqtt_routine_t *routine)
{
assert(NULL != routine);
// cleanup node states
for (struct node_state **sp =
(struct node_state **) utarray_front(routine->states);
sp != NULL;
sp = (struct node_state **) utarray_next(routine->states, sp)) {
free((*sp));
}
utarray_free(routine->states);
// stop heartbeat send
nng_mtx_free(routine->mutex);
nng_aio_cancel(routine->aio);
nng_aio_free(routine->aio);
@@ -613,7 +623,7 @@ static neu_err_code_e node_list_get(neu_plugin_t * plugin,
neu_resp_get_node_t *nodes = (neu_resp_get_node_t *) data;
utarray_foreach(nodes->nodes, neu_resp_node_info_t *, info)
{
plog_info(plugin, "%s, %s", info->node, info->plugin);
// plog_info(plugin, "%s, %s", info->node, info->plugin);
neu_reqresp_head_t header = { 0 };
neu_req_get_node_state_t cmd = { 0 };
header.ctx = strdup(info->node);
@@ -634,20 +644,14 @@ static neu_err_code_e node_state_update(neu_plugin_t * plugin,
{
mqtt_routine_t * routine = plugin->routine;
neu_resp_get_node_state_t *state = (neu_resp_get_node_state_t *) data;
char * node = (char *) head->ctx;
int link = state->state.link;
int running = state->state.running;
plog_debug(plugin, "node:%s, link:%d, running:%d", node, link, running);
nng_mtx_lock(routine->mutex);
struct node_state *ns = NULL;
for (struct node_state **sp =
(struct node_state **) utarray_front(routine->states);
sp != NULL;
sp = (struct node_state **) utarray_next(routine->states, sp)) {
// plog_info(plugin, "node:%s, state->link:%d, state->running:%d",
// (char *) head->ctx, state->link, state->running);
if (0 == strcmp((char *) head->ctx, (*sp)->node)) {
ns = (*sp);
break;
@@ -662,6 +666,7 @@ static neu_err_code_e node_state_update(neu_plugin_t * plugin,
ns->link = state->state.link;
ns->running = state->state.running;
nng_mtx_unlock(routine->mutex);
free(head->ctx);
plugins/mqtt/mqtt.h
+7
View File
@@ -29,6 +29,13 @@ extern "C" {
typedef struct mqtt_routine mqtt_routine_t;
struct node_state {
char node[NEU_NODE_NAME_LEN];
char plugin[NEU_PLUGIN_NAME_LEN];
int running;
int link;
};
struct neu_plugin {
neu_plugin_common_t common;
bool running;
plugins/mqtt/util.c
+3 -3
View File
@@ -114,9 +114,9 @@ int mqtt_option_init(char *config, neu_mqtt_option_t *option)
neu_json_elem_t key = { .name = "key", .t = NEU_JSON_STR };
neu_json_elem_t keypass = { .name = "keypass", .t = NEU_JSON_STR };
ret = neu_parse_param(config, &error, 12, &id, &upload, &format, &ssl,
&host, &port, &username, &password, &ca, &cert, &key,
&keypass);
ret = neu_parse_param(config, &error, 13, &id, &upload, &heartbeat, &format,
&ssl, &host, &port, &username, &password, &ca, &cert,
&key, &keypass);
if (0 != ret) {
return ret;
}
src/parser/neu_json_mqtt.c
+62
View File
@@ -22,6 +22,7 @@
* It was automatically generated by `json-autotype`.
*/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
@@ -85,3 +86,64 @@ int neu_json_encode_mqtt_resp(void *json_object, void *param)
return ret;
}
int neu_json_encode_state_header_resp(void *json_object, void *param)
{
int ret = 0;
neu_json_states_head_t *resp = (neu_json_states_head_t *) param;
neu_json_elem_t resp_elems[] = { {
.name = "version",
.t = NEU_JSON_STR,
.v.val_str = resp->version,
},
{
.name = "timestamp",
.t = NEU_JSON_INT,
.v.val_int = resp->timpstamp,
} };
ret = neu_json_encode_field(json_object, resp_elems,
NEU_JSON_ELEM_SIZE(resp_elems));
return ret;
}
int neu_json_encode_states_resp(void *json_object, void *param)
{
int ret = 0;
neu_json_states_t *resp = (neu_json_states_t *) param;
void * state_array = neu_json_array();
neu_json_node_state_t *states = resp->states;
for (int i = 0; i < resp->n_state; i++) {
neu_json_elem_t state_elems[] = { {
.name = "node",
.t = NEU_JSON_STR,
.v.val_str = (&states[i])->node,
},
{
.name = "link",
.t = NEU_JSON_INT,
.v.val_int = (&states[i])->link,
},
{
.name = "running",
.t = NEU_JSON_INT,
.v.val_int =
(&states[i])->running,
} };
state_array = neu_json_encode_array(state_array, state_elems,
NEU_JSON_ELEM_SIZE(state_elems));
}
neu_json_elem_t resp_elems[] = { {
.name = "states",
.t = NEU_JSON_OBJECT,
.v.val_object = state_array,
} };
ret = neu_json_encode_field(json_object, resp_elems,
NEU_JSON_ELEM_SIZE(resp_elems));
return ret;
}