From 36a16f1bb6845e07ecab3f19164ab2df691ad623 Mon Sep 17 00:00:00 2001 From: gc87 Date: Sun, 12 Jun 2022 21:01:28 +0800 Subject: [PATCH] refactor(mqtt): trans-data upload --- plugins/mqtt/command/command.c | 14 +- plugins/mqtt/command/command.h | 9 +- plugins/mqtt/command/read_write.c | 322 +++++++++++------------------- plugins/mqtt/command/read_write.h | 8 +- plugins/mqtt/mqtt.c | 110 ++++++---- src/connection/mqtt_c_client.c | 6 +- 6 files changed, 204 insertions(+), 265 deletions(-) diff --git a/plugins/mqtt/command/command.c b/plugins/mqtt/command/command.c index cefb1307..0db132e5 100644 --- a/plugins/mqtt/command/command.c +++ b/plugins/mqtt/command/command.c @@ -146,15 +146,11 @@ void command_response_handle(mqtt_response_t *response) // resp_val); //} -// char *command_read_periodic_response(neu_plugin_t *plugin, uint64_t sender, -// const char * node_name, -// neu_taggrp_config_t *config, -// neu_data_val_t * resp_val, -// int upload_format) -//{ -// return command_rw_read_periodic_response(plugin, sender, node_name, config, -// resp_val, upload_format); -//} +char *command_read_periodic_response(neu_reqresp_trans_data_t *data, + int upload_format) +{ + return command_rw_read_periodic_response(data, upload_format); +} // char *command_write_response(neu_json_mqtt_t *parse_header, // neu_data_val_t * resp_val) diff --git a/plugins/mqtt/command/command.h b/plugins/mqtt/command/command.h index 0a981c45..dcfd092b 100644 --- a/plugins/mqtt/command/command.h +++ b/plugins/mqtt/command/command.h @@ -64,11 +64,10 @@ void command_response_handle(mqtt_response_t *response); // char *command_read_once_response(neu_plugin_t *plugin, uint32_t node_id, // neu_json_mqtt_t *parse_header, // neu_data_val_t * resp_val); -// char *command_read_periodic_response(neu_plugin_t *plugin, uint64_t sender, -// const char * node_name, -// neu_taggrp_config_t *config, -// neu_data_val_t * resp_val, -// int upload_format); + +char *command_read_periodic_response(neu_reqresp_trans_data_t *data, + int upload_format); + // char *command_write_response(neu_json_mqtt_t *parse_header, // neu_data_val_t * resp_val); diff --git a/plugins/mqtt/command/read_write.c b/plugins/mqtt/command/read_write.c index a53b1e97..23e6552c 100644 --- a/plugins/mqtt/command/read_write.c +++ b/plugins/mqtt/command/read_write.c @@ -25,13 +25,13 @@ #include "read_write.h" #include "utils/log.h" -// static uint64_t current_time() -//{ -// struct timeval tv; -// gettimeofday(&tv, NULL); -// uint64_t ms = tv.tv_sec; -// return ms * 1000 + tv.tv_usec / 1000; -//} +static uint64_t current_time() +{ + struct timeval tv; + gettimeofday(&tv, NULL); + uint64_t ms = tv.tv_sec; + return ms * 1000 + tv.tv_usec / 1000; +} int command_rw_read_once_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, neu_json_read_req_t *req, uint32_t req_id) @@ -59,170 +59,97 @@ int command_rw_read_once_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, return 0; } -// static int wrap_read_response_json_object(neu_fixed_array_t * array, -// neu_json_read_resp_t *json, -// neu_plugin_t * plugin, -// uint32_t node_id) -//{ -//(void) array; -//(void) json; -//(void) plugin; -//(void) node_id; -// neu_int_val_t * int_val; -// neu_data_val_t * val; -// neu_datatag_table_t *datatag_table = -// neu_system_get_datatags_table(plugin, node_id); -// json->n_tag = array->length; +static void wrap_read_response_json(neu_reqresp_trans_data_t *data, + neu_json_read_resp_t * json) +{ + json->n_tag = data->n_tag; + if (0 < json->n_tag) { + json->tags = (neu_json_read_resp_tag_t *) calloc( + json->n_tag, sizeof(neu_json_read_resp_tag_t)); + } -// if (0 == json->n_tag) { -// return -1; -//} + for (int i = 0; i < json->n_tag; i++) { + neu_resp_tag_value_t *tag = &data->tags[i]; + neu_type_e type = tag->value.type; + json->tags[i].name = tag->tag; + json->tags[i].error = NEU_ERR_SUCCESS; -// json->tags = (neu_json_read_resp_tag_t *) calloc( -// array->length, sizeof(neu_json_read_resp_tag_t)); + switch (type) { + case NEU_TYPE_ERROR: + json->tags[i].t = NEU_JSON_INT; + json->tags[i].value.val_int = tag->value.value.i32; + json->tags[i].error = tag->value.value.i32; + break; + case NEU_TYPE_UINT8: + json->tags[i].t = NEU_JSON_INT; + json->tags[i].value.val_int = tag->value.value.u8; + break; + case NEU_TYPE_INT8: + json->tags[i].t = NEU_JSON_INT; + json->tags[i].value.val_int = tag->value.value.i8; + break; + case NEU_TYPE_INT16: + json->tags[i].t = NEU_JSON_INT; + json->tags[i].value.val_int = tag->value.value.i16; + break; + case NEU_TYPE_INT32: + json->tags[i].t = NEU_JSON_INT; + json->tags[i].value.val_int = tag->value.value.i32; + break; + case NEU_TYPE_INT64: + json->tags[i].t = NEU_JSON_INT; + json->tags[i].value.val_int = tag->value.value.i64; + break; + case NEU_TYPE_UINT16: + json->tags[i].t = NEU_JSON_INT; + json->tags[i].value.val_int = tag->value.value.u16; + break; + case NEU_TYPE_UINT32: + json->tags[i].t = NEU_JSON_INT; + json->tags[i].value.val_int = tag->value.value.u32; + break; + case NEU_TYPE_UINT64: + json->tags[i].t = NEU_JSON_INT; + json->tags[i].value.val_int = tag->value.value.u64; + break; + case NEU_TYPE_FLOAT: + json->tags[i].t = NEU_JSON_FLOAT; + json->tags[i].value.val_float = tag->value.value.f32; + break; + case NEU_TYPE_DOUBLE: + json->tags[i].t = NEU_JSON_DOUBLE; + json->tags[i].value.val_double = tag->value.value.d64; + break; + case NEU_TYPE_BOOL: + json->tags[i].t = NEU_JSON_BOOL; + json->tags[i].value.val_bool = tag->value.value.boolean; + break; + case NEU_TYPE_BIT: + json->tags[i].t = NEU_JSON_BIT; + json->tags[i].value.val_bit = tag->value.value.u8; + break; + case NEU_TYPE_STRING: + json->tags[i].t = NEU_JSON_STR; + json->tags[i].value.val_str = tag->value.value.str; + break; + default: + break; + } + } +} -// for (uint32_t i = 0; i < array->length; i++) { -// int_val = neu_fixed_array_get(array, i); -// val = int_val->val; -// neu_dtype_e type = neu_dvalue_get_value_type(val); +static void clean_read_response_json(neu_json_read_resp_t *json) +{ + if (NULL == json) { + return; + } -// neu_datatag_t *tag = neu_datatag_tbl_get(datatag_table, int_val->key); + if (NULL == json->tags) { + return; + } -// if (tag == NULL) { -// continue; -//} - -// json->tags[i].name = tag->name; -// json->tags[i].error = NEU_ERR_SUCCESS; - -// switch (type) { -// case NEU_DTYPE_ERRORCODE: { -// int32_t value; -// neu_dvalue_get_errorcode(val, &value); -// json->tags[i].t = NEU_JSON_INT; -// json->tags[i].value.val_int = value; -// json->tags[i].error = value; -// break; -//} -// case NEU_DTYPE_BYTE: { -// uint8_t value; -// neu_dvalue_get_uint8(val, &value); -// json->tags[i].t = NEU_JSON_INT; -// json->tags[i].value.val_int = value; -// break; -//} -// case NEU_DTYPE_INT8: { -// int8_t value; -// neu_dvalue_get_int8(val, &value); -// json->tags[i].t = NEU_JSON_INT; -// json->tags[i].value.val_int = value; -// break; -//} -// case NEU_DTYPE_INT16: { -// int16_t value; -// neu_dvalue_get_int16(val, &value); -// json->tags[i].t = NEU_JSON_INT; -// json->tags[i].value.val_int = value; -// break; -//} -// case NEU_DTYPE_INT32: { -// int32_t value; -// neu_dvalue_get_int32(val, &value); -// json->tags[i].t = NEU_JSON_INT; -// json->tags[i].value.val_int = value; -// break; -//} -// case NEU_DTYPE_INT64: { -// int64_t value; -// neu_dvalue_get_int64(val, &value); -// json->tags[i].t = NEU_JSON_INT; -// json->tags[i].value.val_int = value; -// break; -//} -// case NEU_DTYPE_UINT8: { -// uint8_t value; -// neu_dvalue_get_uint8(val, &value); -// json->tags[i].t = NEU_JSON_INT; -// json->tags[i].value.val_int = value; -// break; -//} -// case NEU_DTYPE_UINT16: { -// uint16_t value; -// neu_dvalue_get_uint16(val, &value); -// json->tags[i].t = NEU_JSON_INT; -// json->tags[i].value.val_int = value; -// break; -//} -// case NEU_DTYPE_UINT32: { -// uint32_t value; -// neu_dvalue_get_uint32(val, &value); -// json->tags[i].t = NEU_JSON_INT; -// json->tags[i].value.val_int = value; -// break; -//} -// case NEU_DTYPE_UINT64: { -// uint64_t value; -// neu_dvalue_get_uint64(val, &value); -// json->tags[i].t = NEU_JSON_INT; -// json->tags[i].value.val_int = value; -// break; -//} -// case NEU_DTYPE_FLOAT: { -// float value; -// neu_dvalue_get_float(val, &value); -// json->tags[i].t = NEU_JSON_FLOAT; -// json->tags[i].value.val_float = value; -// break; -//} -// case NEU_DTYPE_DOUBLE: { -// double value; -// neu_dvalue_get_double(val, &value); -// json->tags[i].t = NEU_JSON_DOUBLE; -// json->tags[i].value.val_double = value; -// break; -//} -// case NEU_DTYPE_BOOL: { -// bool value; -// neu_dvalue_get_bool(val, &value); -// json->tags[i].t = NEU_JSON_BOOL; -// json->tags[i].value.val_bool = value; -// break; -//} -// case NEU_DTYPE_BIT: { -// uint8_t value; -// neu_dvalue_get_bit(val, &value); -// json->tags[i].t = NEU_JSON_BIT; -// json->tags[i].value.val_bit = value; -// break; -//} -// case NEU_DTYPE_CSTR: { -// char *value; -// neu_dvalue_get_ref_cstr(val, &value); -// json->tags[i].t = NEU_JSON_STR; -// json->tags[i].value.val_str = value; -// break; -//} -// default: -// break; -//} -//} - -// return 0; -// return 0; -//} - -// static void clean_read_response_json_object(neu_json_read_resp_t *json) -//{ -// if (NULL == json) { -// return; -//} - -// if (NULL == json->tags) { -// return; -//} - -// free(json->tags); -//} + free(json->tags); +} // char *command_rw_read_once_response(neu_plugin_t *plugin, uint32_t node_id, // neu_json_mqtt_t *parse_header, @@ -262,47 +189,32 @@ int command_rw_read_once_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, // return json_str; //} -// char *command_rw_read_periodic_response(neu_plugin_t *plugin, uint64_t -// sender, const char * node_name, neu_taggrp_config_t *config, -// neu_data_val_t * resp_val, -// int upload_format) -//{ -// UNUSED(plugin); +char *command_rw_read_periodic_response(neu_reqresp_trans_data_t *data, + int upload_format) +{ + char * json_str = NULL; + neu_json_read_periodic_t header = { .group = (char *) data->group, + .node = (char *) data->driver, + .timestamp = current_time() }; + neu_json_read_resp_t json = { 0 }; + wrap_read_response_json(data, &json); -// neu_fixed_array_t *array = NULL; -// neu_dvalue_get_ref_array(resp_val, &array); -// assert(NULL != array); + if (0 == upload_format) { // values + neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp1, &header, + neu_json_encode_read_periodic_resp, + &json_str); + } else if (1 == upload_format) { // tags + neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, &header, + neu_json_encode_read_periodic_resp, + &json_str); + } else { + clean_read_response_json(&json); + return NULL; + } -// if (0 >= array->length) { -// return NULL; -//} - -// char * json_str = NULL; -// neu_json_read_periodic_t header = { -//.group_name = (char *) neu_taggrp_cfg_get_name(config), -//.node_name = (char *) node_name, -//.timestamp = current_time() -//}; - -// neu_json_read_resp_t json = { 0 }; -// wrap_read_response_json_object(array, &json, plugin, (uint32_t) sender); - -// if (0 == upload_format) { // values -// neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp1, &header, -// neu_json_encode_read_periodic_resp, -//&json_str); -//} else if (1 == upload_format) { // tags -// neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, &header, -// neu_json_encode_read_periodic_resp, -//&json_str); -//} else { -// clean_read_response_json_object(&json); -// return NULL; -//} - -// clean_read_response_json_object(&json); -// return json_str; -//} + clean_read_response_json(&json); + return json_str; +} // static int write_command(neu_plugin_t *plugin, uint32_t dest_node_id, // const char * group_config_name, diff --git a/plugins/mqtt/command/read_write.h b/plugins/mqtt/command/read_write.h index 7b04464c..b7870e65 100644 --- a/plugins/mqtt/command/read_write.h +++ b/plugins/mqtt/command/read_write.h @@ -33,10 +33,10 @@ int command_rw_read_once_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, // char *command_rw_read_once_response(neu_plugin_t *plugin, uint32_t node_id, // neu_json_mqtt_t *parse_header, // neu_data_val_t * resp_val); -// char *command_rw_read_periodic_response(neu_plugin_t *plugin, uint64_t -// sender, const char * node_name, neu_taggrp_config_t *grp_config, -// neu_data_val_t * resp_val, -// int upload_format); + +char *command_rw_read_periodic_response(neu_reqresp_trans_data_t *data, + int upload_format); + int command_rw_write_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, neu_json_write_req_t *write_req, uint32_t req_id); // char *command_rw_write_response(neu_json_mqtt_t *parse_header, diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt.c index 6c72fa19..730c0d9f 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -167,19 +167,19 @@ static struct topic_pair *topics_find_topic(UT_array *topics, const char *topic) return NULL; } -// static struct topic_pair *topics_find_type(UT_array *topics, int type) -// { -// for (struct topic_pair **pair = -// (struct topic_pair **) utarray_front(topics); -// NULL != pair; -// pair = (struct topic_pair **) utarray_next(topics, pair)) { -// if (type == (*pair)->type) { -// return *pair; -// } -// } +static struct topic_pair *topics_find_type(UT_array *topics, int type) +{ + for (struct topic_pair **pair = + (struct topic_pair **) utarray_front(topics); + NULL != pair; + pair = (struct topic_pair **) utarray_next(topics, pair)) { + if (type == (*pair)->type) { + return *pair; + } + } -// return NULL; -// } + return NULL; +} static int mqtt_routine_send(mqtt_routine_t *routine, struct context *context); @@ -710,35 +710,67 @@ static int mqtt_plugin_stop(neu_plugin_t *plugin) //} // static neu_err_code_e trans_data(neu_plugin_t *plugin, neu_request_t *req) -//{ -//(void) plugin; -//(void) req; -// return 0; -// pthread_mutex_lock(&plugin->mutex); +// { +// (void) plugin; +// (void) req; +// return 0; +// pthread_mutex_lock(&plugin->mutex); -// if (!plugin->routine_running) { -// pthread_mutex_unlock(&plugin->mutex); -// return NEU_ERR_FAILURE; -//} +// if (!plugin->routine_running) { +// pthread_mutex_unlock(&plugin->mutex); +// return NEU_ERR_FAILURE; +// } -// mqtt_routine_t * routine = plugin->routine; -// neu_reqresp_data_t * neu_data = (neu_reqresp_data_t *) req->buf; -// struct topic_pair * pair = NULL; -// int type = TOPIC_TYPE_UPLOAD; -// uint64_t sender = req->sender_id; -// const char * node_name = req->node_name; -// neu_taggrp_config_t *config = neu_data->grp_config; -// pair = vector_find_item(&routine->topics, (void *) &type, -// topic_type_match); char *json = command_read_periodic_response(plugin, -// sender, node_name, config, neu_data->data_val, -// plugin->routine->option.format); -// struct context *c = mqtt_routine_context_create(0, NULL, json, pair, -// true); if (NULL != c) { mqtt_routine_send(routine, c); -//} +// mqtt_routine_t * routine = plugin->routine; +// neu_reqresp_data_t * neu_data = (neu_reqresp_data_t *) req->buf; +// struct topic_pair * pair = NULL; +// int type = TOPIC_TYPE_UPLOAD; +// uint64_t sender = req->sender_id; +// const char * node_name = req->node_name; +// neu_taggrp_config_t *config = neu_data->grp_config; +// pair = vector_find_item(&routine->topics, (void *) &type, +// topic_type_match); char *json = command_read_periodic_response(plugin, +// sender, node_name, +// config, neu_data->data_val, +// plugin->routine->option.format); +// struct context *c = mqtt_routine_context_create(0, NULL, json, pair, +// true); if (NULL != c) { +// mqtt_routine_send(routine, c); +// } -// pthread_mutex_unlock(&plugin->mutex); -// return NEU_ERR_SUCCESS; -//} +// pthread_mutex_unlock(&plugin->mutex); +// return NEU_ERR_SUCCESS; +// } + +static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data) +{ + neu_reqresp_trans_data_t *trans_data = data; + + pthread_mutex_lock(&plugin->mutex); + + if (!plugin->routine_running) { + pthread_mutex_unlock(&plugin->mutex); + return NEU_ERR_FAILURE; + } + + mqtt_routine_t * routine = plugin->routine; + int type = TOPIC_TYPE_UPLOAD; + struct topic_pair *pair = topics_find_type(routine->topics, type); + char * json = + command_read_periodic_response(trans_data, routine->option.format); + if (NULL == json) { + pthread_mutex_unlock(&plugin->mutex); + return NEU_ERR_FAILURE; + } + + struct context *c = mqtt_routine_context_create(0, NULL, json, pair, true); + if (NULL != c) { + mqtt_routine_send(routine, c); + } + + pthread_mutex_unlock(&plugin->mutex); + return NEU_ERR_SUCCESS; +} static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, void *data) @@ -757,7 +789,7 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, // error = write_response(plugin, req); // break; case NEU_REQRESP_TRANS_DATA: - // error = trans_data(plugin, req); + error = trans_data(plugin, data); break; default: error = NEU_ERR_FAILURE; diff --git a/src/connection/mqtt_c_client.c b/src/connection/mqtt_c_client.c index df42d988..8ce2e254 100644 --- a/src/connection/mqtt_c_client.c +++ b/src/connection/mqtt_c_client.c @@ -828,9 +828,9 @@ neu_err_code_e mqtt_c_client_publish(mqtt_c_client_t *client, const char *topic, assert(NULL != client); assert(NULL != topic && 0 < strlen(topic)); - if (SEND_BUF_SIZE < len) { - return NEU_ERR_MQTT_PUBLISH_OVER_LENGTH; - } + // if (SEND_BUF_SIZE < len) { + // return NEU_ERR_MQTT_PUBLISH_OVER_LENGTH; + // } neu_err_code_e error = mqtt_c_client_is_connected(client); if (0 != error) {