From 2eff71d508e04f3cf40e4aea0caa53e2de2cf8c1 Mon Sep 17 00:00:00 2001 From: gc87 Date: Fri, 4 Mar 2022 14:00:03 +0800 Subject: [PATCH] Add node name of sender in trans_data msg --- include/neu_adapter.h | 1 + plugins/mqtt/command/command.c | 4 +++- plugins/mqtt/command/command.h | 1 + plugins/mqtt/command/read_write.c | 35 +++++++------------------------ plugins/mqtt/command/read_write.h | 1 + plugins/mqtt/plugin_mqtt.c | 19 +++++++++-------- src/adapter/adapter.c | 2 ++ src/core/message.h | 1 + src/core/neu_manager.c | 1 + 9 files changed, 27 insertions(+), 38 deletions(-) diff --git a/include/neu_adapter.h b/include/neu_adapter.h index fae713a9..508872bb 100644 --- a/include/neu_adapter.h +++ b/include/neu_adapter.h @@ -359,6 +359,7 @@ typedef struct neu_request { uint32_t req_id; neu_reqresp_type_e req_type; uint64_t sender_id; // adapter_id of sender + char * node_name; // adapter name of sender uint32_t buf_len; void * buf; } neu_request_t; diff --git a/plugins/mqtt/command/command.c b/plugins/mqtt/command/command.c index ab3e2eb9..c97740b1 100644 --- a/plugins/mqtt/command/command.c +++ b/plugins/mqtt/command/command.c @@ -555,10 +555,12 @@ char *command_read_once_response(neu_plugin_t * plugin, } 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) { - return command_rw_read_periodic_response(plugin, sender, config, resp_val); + return command_rw_read_periodic_response(plugin, sender, node_name, config, + resp_val); } char *command_write_response(neu_plugin_t * plugin, diff --git a/plugins/mqtt/command/command.h b/plugins/mqtt/command/command.h index 98114fe0..3602a2ae 100644 --- a/plugins/mqtt/command/command.h +++ b/plugins/mqtt/command/command.h @@ -64,6 +64,7 @@ char *command_read_once_response(neu_plugin_t * plugin, 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); char *command_write_response(neu_plugin_t * plugin, diff --git a/plugins/mqtt/command/read_write.c b/plugins/mqtt/command/read_write.c index d81034ce..bd77ec25 100644 --- a/plugins/mqtt/command/read_write.c +++ b/plugins/mqtt/command/read_write.c @@ -235,21 +235,18 @@ char *command_rw_read_once_response(neu_plugin_t * plugin, return json_str; } -static bool node_match(const void *key, const void *item) -{ - neu_node_id_t id = *(neu_node_id_t *) key; - neu_node_info_t *info = (neu_node_info_t *) item; - return (id == info->node_id) ? true : false; -} - 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) { + UNUSED(plugin); + neu_fixed_array_t *array = NULL; neu_dvalue_get_ref_array(resp_val, &array); - if (NULL == array || 0 >= array->length) { - log_error("Get array error"); + assert(NULL != array); + + if (0 >= array->length) { return NULL; } @@ -257,26 +254,10 @@ char *command_rw_read_periodic_response(neu_plugin_t *plugin, uint64_t sender, neu_json_read_periodic_t header = { .config_name = (char *) neu_taggrp_cfg_get_name(config), .node_id = sender, + .node_name = (char *) node_name, .timestamp = current_time() }; - vector_t nodes = neu_system_get_nodes(plugin, NEU_NODE_TYPE_DRIVER); - neu_node_info_t *info = NULL; - info = vector_find_item(&nodes, (void *) &sender, node_match); - if (NULL != info) { - header.node_name = strdup(info->node_name); - } else { - header.node_name = strdup(""); - } - - VECTOR_FOR_EACH(&nodes, iter) - { - neu_node_info_t *info = (neu_node_info_t *) iterator_get(&iter); - free(info->node_name); - } - - vector_uninit(&nodes); - log_debug("config:%s, node:%s, self_id:%ld, " "sender:%ld, time:%u", header.config_name, header.node_name, header.node_id, sender, @@ -287,8 +268,6 @@ char *command_rw_read_periodic_response(neu_plugin_t *plugin, uint64_t sender, neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, &header, neu_json_encode_read_periodic_resp, &json_str); clean_read_response_json_object(&json); - - free(header.node_name); return json_str; } diff --git a/plugins/mqtt/command/read_write.h b/plugins/mqtt/command/read_write.h index b59f65b2..3d3e4742 100644 --- a/plugins/mqtt/command/read_write.h +++ b/plugins/mqtt/command/read_write.h @@ -34,6 +34,7 @@ char *command_rw_read_once_response(neu_plugin_t * plugin, 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 command_rw_write_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, diff --git a/plugins/mqtt/plugin_mqtt.c b/plugins/mqtt/plugin_mqtt.c index d1237f6f..f9ab5563 100644 --- a/plugins/mqtt/plugin_mqtt.c +++ b/plugins/mqtt/plugin_mqtt.c @@ -609,17 +609,18 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_request_t *req) break; } case NEU_REQRESP_TRANS_DATA: { - neu_reqresp_data_t *neu_data = NULL; - neu_data = (neu_reqresp_data_t *) req->buf; - struct topic_pair * pair = NULL; - char * json_str = NULL; - int type = TOPIC_TYPE_READ; - uint64_t sender = req->sender_id; - neu_taggrp_config_t *config = neu_data->grp_config; + neu_reqresp_data_t *neu_data = NULL; + neu_data = (neu_reqresp_data_t *) req->buf; + struct topic_pair * pair = NULL; + char * json_str = NULL; + int type = TOPIC_TYPE_READ; + 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(&plugin->topics, (void *) &type, topic_type_match); - json_str = command_read_periodic_response(plugin, sender, config, - neu_data->data_val); + json_str = command_read_periodic_response(plugin, sender, node_name, + config, neu_data->data_val); mqtt_context_add(plugin, 0, NULL, json_str, pair, true); break; } diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index fce5cbe5..5c67780b 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -820,6 +820,7 @@ static void adapter_loop(void *arg) req.buf_len = sizeof(neu_reqresp_data_t); req.buf = (void *) &data_req; req.sender_id = trans_data->sender_id; + req.node_name = trans_data->node_name; intf_funs->request(adapter->plugin, &req); if (is_need_free) { neu_dvalue_free(data_req.data_val); @@ -1650,6 +1651,7 @@ static int adapter_response(neu_adapter_t *adapter, neu_response_t *resp) trans_data = (neuron_trans_data_t *) msg_get_buf_ptr(pay_msg); trans_data->grp_config = neu_data->grp_config; trans_data->sender_id = to_node_id((adapter), (adapter)->id); + trans_data->node_name = adapter->name; trans_buf = &trans_data->trans_buf; rv = neu_trans_buf_init(trans_buf, adapter->trans_kind, neu_data->data_val); diff --git a/src/core/message.h b/src/core/message.h index bfb4fc5d..9dfe1eb6 100644 --- a/src/core/message.h +++ b/src/core/message.h @@ -175,6 +175,7 @@ typedef struct grp_config_changed_event { typedef struct neuron_trans_data { neu_taggrp_config_t *grp_config; adapter_id_t sender_id; // adapter_id of sender + char * node_name; neu_trans_buf_t trans_buf; } neuron_trans_data_t; diff --git a/src/core/neu_manager.c b/src/core/neu_manager.c index 8ae182af..d85ed6ce 100644 --- a/src/core/neu_manager.c +++ b/src/core/neu_manager.c @@ -969,6 +969,7 @@ static int dispatch_databuf_to_adapters(neu_manager_t * manager, (neu_taggrp_config_t *) neu_taggrp_cfg_ref( neu_trans_data->grp_config); out_neu_trans_data->sender_id = neu_trans_data->sender_id; + out_neu_trans_data->node_name = neu_trans_data->node_name; neu_trans_buf_copy(&out_neu_trans_data->trans_buf, trans_buf); nng_msg_set_pipe(out_msg, msg_pipe); log_debug("Forward databuf to pipe: %d", msg_pipe);