GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-25 19:31:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refactor neu_manager loop MSG_CMD_SUBSCRIBE_NODE handler code

Browse Source 
The handler code for message type MSG_CMD_SUBSCRIBE_NODE is refactored
  into a new function neu_manager_subscribe_node
This commit is contained in:
eeff
2022-01-27 21:09:03 +08:00
committed by Parker Liu
parent fa22f535ed
commit b032785e6e
2 changed files with 50 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/neu_manager.c
+46 -47
View File
@@ -1389,53 +1389,8 @@ static void manager_loop(void *arg)
}
case MSG_CMD_SUBSCRIBE_NODE: {
size_t msg_size;
nng_msg * out_msg;
nng_pipe msg_pipe;
int need_forward;
adapter_id_t src_adapter_id;
adapter_id_t sub_adapter_id;
subscribe_node_cmd_t *cmd_ptr;
adapter_reg_entity_t *src_reg_entity;
adapter_reg_entity_t *sub_reg_entity;
cmd_ptr = (subscribe_node_cmd_t *) msg_get_buf_ptr(pay_msg);
nng_mtx_lock(manager->adapters_mtx);
src_adapter_id = neu_manager_adapter_id_from_node_id(
manager, cmd_ptr->src_node_id);
sub_adapter_id = neu_manager_adapter_id_from_node_id(
manager, cmd_ptr->dst_node_id);
src_reg_entity =
find_reg_adapter_by_id(&manager->reg_adapters, src_adapter_id);
sub_reg_entity =
find_reg_adapter_by_id(&manager->reg_adapters, sub_adapter_id);
neu_adapter_add_sub_grp_config(sub_reg_entity->adapter,
cmd_ptr->src_node_id,
cmd_ptr->grp_config);
msg_pipe = src_reg_entity->adapter_pipe;
nng_mtx_unlock(manager->adapters_mtx);
need_forward = sub_grp_config_with_pipe(
cmd_ptr->grp_config, sub_reg_entity->adapter_pipe);
if (0 == need_forward) {
break;
}
msg_size = msg_inplace_data_get_size(sizeof(subscribe_node_cmd_t));
rv = nng_msg_alloc(&out_msg, msg_size);
if (rv == 0) {
message_t * msg_ptr;
subscribe_node_cmd_t *out_cmd_ptr;
msg_ptr = (message_t *) nng_msg_body(out_msg);
msg_inplace_data_init(msg_ptr, MSG_CMD_SUBSCRIBE_NODE,
sizeof(subscribe_node_cmd_t));
out_cmd_ptr = msg_get_buf_ptr(msg_ptr);
memcpy(out_cmd_ptr, cmd_ptr, sizeof(subscribe_node_cmd_t));
nng_msg_set_pipe(out_msg, msg_pipe);
log_info("Forward subscribe driver command to driver pipe: %d",
msg_pipe);
nng_sendmsg(manager_bind->mng_sock, out_msg, 0);
}
subscribe_node_cmd_t *cmd_ptr = msg_get_buf_ptr(pay_msg);
neu_manager_subscribe_node(manager, cmd_ptr);
break;
}
@@ -1799,6 +1754,50 @@ int neu_manager_update_node(neu_manager_t *manager, neu_cmd_update_node_t *cmd)
return rv;
}
int neu_manager_subscribe_node(neu_manager_t * manager,
subscribe_node_cmd_t *cmd)
{
int rv = 0;
nng_mtx_lock(manager->adapters_mtx);
adapter_id_t src_adapter_id =
neu_manager_adapter_id_from_node_id(manager, cmd->src_node_id);
adapter_id_t sub_adapter_id =
neu_manager_adapter_id_from_node_id(manager, cmd->dst_node_id);
adapter_reg_entity_t *src_reg_entity =
find_reg_adapter_by_id(&manager->reg_adapters, src_adapter_id);
adapter_reg_entity_t *sub_reg_entity =
find_reg_adapter_by_id(&manager->reg_adapters, sub_adapter_id);
neu_adapter_add_sub_grp_config(sub_reg_entity->adapter, cmd->src_node_id,
cmd->grp_config);
nng_mtx_unlock(manager->adapters_mtx);
int need_forward =
sub_grp_config_with_pipe(cmd->grp_config, sub_reg_entity->adapter_pipe);
if (0 == need_forward) {
return rv;
}
size_t msg_size = msg_inplace_data_get_size(sizeof(*cmd));
nng_msg *out_msg = NULL;
rv = nng_msg_alloc(&out_msg, msg_size);
if (rv == 0) {
message_t * msg_ptr;
subscribe_node_cmd_t *out_cmd_ptr;
msg_ptr = (message_t *) nng_msg_body(out_msg);
msg_inplace_data_init(msg_ptr, MSG_CMD_SUBSCRIBE_NODE,
sizeof(subscribe_node_cmd_t));
out_cmd_ptr = msg_get_buf_ptr(msg_ptr);
memcpy(out_cmd_ptr, cmd, sizeof(subscribe_node_cmd_t));
nng_msg_set_pipe(out_msg, src_reg_entity->adapter_pipe);
log_info("Forward subscribe driver command to driver pipe: %d",
src_reg_entity->adapter_pipe);
nng_sendmsg(manager->bind_info.mng_sock, out_msg, 0);
}
return rv;
}
static bool adapter_match_node_type(neu_adapter_t * adapter,
neu_node_type_e node_type)
{
src/core/neu_manager.h
+4 -1
View File
@@ -25,7 +25,8 @@
#define DEFAULT_ADAPTER_REG_COUNT 8
typedef struct neu_manager neu_manager_t;
typedef struct neu_manager neu_manager_t;
typedef struct subscribe_driver_cmd subscribe_node_cmd_t;
typedef void (*bind_notify_fun)(nng_pipe p, nng_pipe_ev ev, void *arg);
@@ -45,6 +46,8 @@ int neu_manager_get_nodes(neu_manager_t *manager, neu_node_type_e node_type,
vector_t *result_nodes);
int neu_manager_get_node_name_by_id(neu_manager_t *manager,
neu_node_id_t node_id, char **name);
int neu_manager_subscribe_node(neu_manager_t * manager,
subscribe_node_cmd_t *cmd);
int neu_manager_add_grp_config(neu_manager_t * manager,
neu_cmd_add_grp_config_t *cmd);