diff --git a/plugins/mqtt/command/command.c b/plugins/mqtt/command/command.c index 0db132e5..e53ef354 100644 --- a/plugins/mqtt/command/command.c +++ b/plugins/mqtt/command/command.c @@ -37,17 +37,13 @@ static int ping_response(char **output_str, neu_plugin_t *plugin, static int read_response(mqtt_response_t *response, neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, char *json_str) { - neu_json_read_req_t *req = NULL; - int rc = neu_json_decode_read_req(json_str, &req); - if (0 == rc) { - uint32_t req_id = neu_plugin_get_event_id(plugin); - if (0 < req_id) { - response->context_add(plugin, req_id, mqtt, NULL, - response->topic_pair, false); - } + UNUSED(response); - command_rw_read_once_request(plugin, mqtt, req, req_id); - neu_json_decode_read_req_free(req); + neu_json_read_req_t *json = NULL; + int rc = neu_json_decode_read_req(json_str, &json); + if (0 == rc) { + command_rw_read_once_request(plugin, mqtt, json); + neu_json_decode_read_req_free(json); } return rc; @@ -138,18 +134,16 @@ void command_response_handle(mqtt_response_t *response) neu_json_decode_mqtt_req_free(mqtt); } -// 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) -//{ -// return command_rw_read_once_response(plugin, node_id, parse_header, -// resp_val); -//} - -char *command_read_periodic_response(neu_reqresp_trans_data_t *data, - int upload_format) +char *command_read_once_response(neu_reqresp_head_t * head, + neu_resp_read_group_t *data, int format) { - return command_rw_read_periodic_response(data, upload_format); + return command_rw_read_once_response(head, data, format); + return NULL; +} + +char *command_read_periodic_response(neu_reqresp_trans_data_t *data, int format) +{ + return command_rw_read_periodic_response(data, format); } // char *command_write_response(neu_json_mqtt_t *parse_header, diff --git a/plugins/mqtt/command/command.h b/plugins/mqtt/command/command.h index dcfd092b..31683f84 100644 --- a/plugins/mqtt/command/command.h +++ b/plugins/mqtt/command/command.h @@ -61,12 +61,12 @@ typedef struct { } mqtt_response_t; 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_once_response(neu_reqresp_head_t * head, + neu_resp_read_group_t *data, int format); char *command_read_periodic_response(neu_reqresp_trans_data_t *data, - int upload_format); + int 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 23e6552c..43adf737 100644 --- a/plugins/mqtt/command/read_write.c +++ b/plugins/mqtt/command/read_write.c @@ -34,42 +34,33 @@ static uint64_t current_time() } 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) + neu_json_read_req_t *req) { - (void) plugin; - (void) mqtt; - (void) req; - (void) req_id; - // neu_node_id_t node_id = - // neu_plugin_get_node_id_by_node_name(plugin, req->node_name); - // if (node_id == 0) { - // zlog_debug(neuron, "node %s does not exist", req->node_name); - // return -1; - //} - // zlog_info(neuron, "READ uuid:%s, node id:%u", mqtt->uuid, node_id); - // neu_taggrp_config_t *config = - // neu_system_find_group_config(plugin, node_id, req->group_name); - // if (NULL == config) { - // zlog_debug(neuron, "The requested config does not exist"); - // return -2; - //} + neu_reqresp_head_t header = { 0 }; + header.ctx = mqtt; + header.type = NEU_REQ_READ_GROUP; + neu_req_read_group_t cmd = { 0 }; + strcpy(cmd.driver, req->node); + strcpy(cmd.group, req->group); + int ret = neu_plugin_op(plugin, header, &cmd); + if (ret != 0) { + return -1; + } - // neu_plugin_send_read_cmd(plugin, req_id, node_id, config); - // return 0; return 0; } -static void wrap_read_response_json(neu_reqresp_trans_data_t *data, - neu_json_read_resp_t * json) +static void wrap_read_response_json(neu_resp_tag_value_t *tags, uint16_t len, + neu_json_read_resp_t *json) { - json->n_tag = data->n_tag; + json->n_tag = len; if (0 < json->n_tag) { json->tags = (neu_json_read_resp_tag_t *) calloc( json->n_tag, sizeof(neu_json_read_resp_tag_t)); } for (int i = 0; i < json->n_tag; i++) { - neu_resp_tag_value_t *tag = &data->tags[i]; + neu_resp_tag_value_t *tag = &tags[i]; neu_type_e type = tag->value.type; json->tags[i].name = tag->tag; json->tags[i].error = NEU_ERR_SUCCESS; @@ -151,59 +142,39 @@ static void clean_read_response_json(neu_json_read_resp_t *json) free(json->tags); } -// 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) -//{ -// UNUSED(plugin); +char *command_rw_read_once_response(neu_reqresp_head_t * head, + neu_resp_read_group_t *data, int format) +{ + UNUSED(format); -// neu_fixed_array_t *array; -// int rc = neu_dvalue_get_ref_array(resp_val, &array); -// if (0 != rc) { -// zlog_info(neuron, "Get array ref error"); -// return NULL; -//} - -// if (0 >= array->length) { -// return NULL; -//} - -// char *json_str = NULL; - -// neu_json_read_resp_t json; -// memset(&json, 0, sizeof(neu_json_read_resp_t)); -// rc = wrap_read_response_json_object(array, &json, plugin, node_id); -// if (0 != rc) { -// return NULL; -//} - -// rc = neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, -// parse_header, neu_json_encode_mqtt_resp, -//&json_str); -// clean_read_response_json_object(&json); -// if (0 != rc) { -// zlog_info(neuron, "Json string parse error:%d", rc); -// return NULL; -//} - -// return json_str; -//} + neu_resp_tag_value_t *tags = data->tags; + uint16_t len = data->n_tag; + char * json_str = NULL; + neu_json_read_resp_t json = { 0 }; + wrap_read_response_json(tags, len, &json); + neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, head->ctx, + neu_json_encode_mqtt_resp, &json_str); + clean_read_response_json(&json); + return json_str; +} char *command_rw_read_periodic_response(neu_reqresp_trans_data_t *data, - int upload_format) + int format) { + neu_resp_tag_value_t * tags = data->tags; + uint16_t len = data->n_tag; 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); + wrap_read_response_json(tags, len, &json); - if (0 == upload_format) { // values + if (0 == 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 + } else if (1 == format) { // tags neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, &header, neu_json_encode_read_periodic_resp, &json_str); diff --git a/plugins/mqtt/command/read_write.h b/plugins/mqtt/command/read_write.h index b7870e65..48ebc7bb 100644 --- a/plugins/mqtt/command/read_write.h +++ b/plugins/mqtt/command/read_write.h @@ -29,13 +29,13 @@ extern "C" { #include "common.h" 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); -// 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); + neu_json_read_req_t *req); + +char *command_rw_read_once_response(neu_reqresp_head_t * head, + neu_resp_read_group_t *data, int format); char *command_rw_read_periodic_response(neu_reqresp_trans_data_t *data, - int upload_format); + int 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); diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt.c index 730c0d9f..f7f96949 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -709,38 +709,33 @@ static int mqtt_plugin_stop(neu_plugin_t *plugin) // return NEU_ERR_SUCCESS; //} -// 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); +static neu_err_code_e read_response(neu_plugin_t * plugin, + neu_reqresp_head_t *head, void *data) +{ + neu_resp_read_group_t *read_data = data; -// if (!plugin->routine_running) { -// pthread_mutex_unlock(&plugin->mutex); -// return NEU_ERR_FAILURE; -// } + pthread_mutex_lock(&plugin->mutex); -// 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); -// } + if (!plugin->routine_running) { + pthread_mutex_unlock(&plugin->mutex); + return NEU_ERR_FAILURE; + } -// pthread_mutex_unlock(&plugin->mutex); -// return NEU_ERR_SUCCESS; -// } + pthread_mutex_lock(&plugin->routine->contexts_mtx); + mqtt_routine_t * routine = plugin->routine; + int type = TOPIC_TYPE_READ; + struct topic_pair *pair = topics_find_type(routine->topics, type); + char * json = + command_read_once_response(head, read_data, 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->routine->contexts_mtx); + + pthread_mutex_unlock(&plugin->mutex); + return NEU_ERR_SUCCESS; +} static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data) { @@ -782,10 +777,10 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, neu_err_code_e error = NEU_ERR_SUCCESS; switch (head->type) { - // case NEU_REQRESP_READ_RESP: - // error = read_response(plugin, req); - // break; - // case NEU_REQRESP_WRITE_RESP: + case NEU_RESP_READ_GROUP: + error = read_response(plugin, head, data); + break; + // case NEU_RESP_WRITE_TAG: // error = write_response(plugin, req); // break; case NEU_REQRESP_TRANS_DATA: