GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-17 04:40:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add node name of sender in trans_data msg

Browse Source
This commit is contained in:
gc87
2022-03-04 14:00:03 +08:00
parent c01225dcda
commit 2eff71d508
9 changed files with 27 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neu_adapter.h
+1
View File
@@ -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;
plugins/mqtt/command/command.c
+3 -1
View File
@@ -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,
plugins/mqtt/command/command.h
+1
View File
@@ -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,
plugins/mqtt/command/read_write.c
+7 -28
View File
@@ -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;
}
plugins/mqtt/command/read_write.h
+1
View File
@@ -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,
plugins/mqtt/plugin_mqtt.c
+10 -9
View File
@@ -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;
}
src/adapter/adapter.c
+2
View File
@@ -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);
src/core/message.h
+1
View File
@@ -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;
src/core/neu_manager.c
+1
View File
@@ -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);