From 596cd699fd3992221bb9b4b39ea52eb6ed369200 Mon Sep 17 00:00:00 2001 From: gc87 Date: Wed, 13 Jul 2022 16:00:04 +0800 Subject: [PATCH] feat(mqtt): state json string send --- ft/resource/common.resource | 2 +- include/neuron/json/neu_json_mqtt.h | 19 +++++++++ plugins/mqtt/command/command.c | 4 +- plugins/mqtt/command/command.h | 2 +- plugins/mqtt/command/heartbeat.c | 30 +++++++++++--- plugins/mqtt/command/heartbeat.h | 2 +- plugins/mqtt/mqtt.c | 63 ++++++++++++++++------------- plugins/mqtt/mqtt.h | 7 ++++ plugins/mqtt/util.c | 6 +-- src/parser/neu_json_mqtt.c | 62 ++++++++++++++++++++++++++++ 10 files changed, 154 insertions(+), 43 deletions(-) diff --git a/ft/resource/common.resource b/ft/resource/common.resource index c3032bae..1ffb5467 100644 --- a/ft/resource/common.resource +++ b/ft/resource/common.resource @@ -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":""} diff --git a/include/neuron/json/neu_json_mqtt.h b/include/neuron/json/neu_json_mqtt.h index a751ddae..c091b66d 100644 --- a/include/neuron/json/neu_json_mqtt.h +++ b/include/neuron/json/neu_json_mqtt.h @@ -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 diff --git a/plugins/mqtt/command/command.c b/plugins/mqtt/command/command.c index d29d2d12..c39f6bf8 100644 --- a/plugins/mqtt/command/command.c +++ b/plugins/mqtt/command/command.c @@ -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); } \ No newline at end of file diff --git a/plugins/mqtt/command/command.h b/plugins/mqtt/command/command.h index 7035b917..be19d464 100644 --- a/plugins/mqtt/command/command.h +++ b/plugins/mqtt/command/command.h @@ -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 } diff --git a/plugins/mqtt/command/heartbeat.c b/plugins/mqtt/command/heartbeat.c index 01a34a51..e60625fd 100644 --- a/plugins/mqtt/command/heartbeat.c +++ b/plugins/mqtt/command/heartbeat.c @@ -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; } \ No newline at end of file diff --git a/plugins/mqtt/command/heartbeat.h b/plugins/mqtt/command/heartbeat.h index 4f183939..5009a0c0 100644 --- a/plugins/mqtt/command/heartbeat.h +++ b/plugins/mqtt/command/heartbeat.h @@ -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 } diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt.c index 5f26a918..98faa837 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -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); diff --git a/plugins/mqtt/mqtt.h b/plugins/mqtt/mqtt.h index 6bac6bb5..93818d30 100644 --- a/plugins/mqtt/mqtt.h +++ b/plugins/mqtt/mqtt.h @@ -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; diff --git a/plugins/mqtt/util.c b/plugins/mqtt/util.c index 753fd250..ca73d2e6 100644 --- a/plugins/mqtt/util.c +++ b/plugins/mqtt/util.c @@ -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; } diff --git a/src/parser/neu_json_mqtt.c b/src/parser/neu_json_mqtt.c index 0d3d0c14..8f560997 100644 --- a/src/parser/neu_json_mqtt.c +++ b/src/parser/neu_json_mqtt.c @@ -22,6 +22,7 @@ * It was automatically generated by `json-autotype`. */ +#include #include #include @@ -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; +}