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

Add support subscribe/unsubscribe node in adapter and other functions

Browse Source 
* Add subscribe/unsubscribe node in command and request interface.
* Add neuron request structure for subscribe/unsubscribe node.
* Add subscribe node before enter message loop.
* Add unsubscribe node after exit message loop.
* Add compare function for primitive value.
This commit is contained in:
Parker Liu
2021-09-23 18:17:39 +08:00
parent f552b41253
commit f4ad674df0
4 changed files with 147 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neu_adapter.h
+9 -5
View File
@@ -47,8 +47,6 @@ typedef enum neu_reqresp_type {
NEU_REQRESP_ERR_CODE, // result code of command
NEU_REQRESP_READ_DATA,
NEU_REQRESP_WRITE_DATA,
NEU_REQRESP_START_PERIODIC_READ,
NEU_REQRESP_STOP_PERIODIC_READ,
NEU_REQRESP_SUBSCRIBE_NODE,
NEU_REQRESP_UNSUBSCRIBE_NODE,
NEU_REQRESP_TRANS_DATA,
@@ -88,11 +86,17 @@ typedef struct neu_reqresp_write {
neu_variable_t * data_var;
} neu_reqresp_write_t;
/* NEU_REQRESP_START_PERIODIC_READ */
typedef struct neu_reqresp_start_periodic_rd {
/* NEU_REQRESP_SUBSCRIBE_NODE */
typedef struct neu_reqresp_subscribe_node {
neu_taggrp_config_t *grp_config;
neu_node_id_t dst_node_id;
} neu_reqresp_start_periodic_rd_t;
} neu_reqresp_subscribe_node_t;
/* NEU_REQRESP_UNSUBSCRIBE_NODE */
typedef struct neu_reqresp_unsubscribe_node {
neu_taggrp_config_t *grp_config;
neu_node_id_t dst_node_id;
} neu_reqresp_unsubscribe_node_t;
/* NEU_REQRESP_TRANS_DATA */
typedef struct neu_reqresp_data {
include/neu_types.h
+42 -13
View File
@@ -63,27 +63,56 @@ typedef bool (*neu_key_match_func)(const void *key, const void *item);
/*
* define functions for implement equal function for primitive value
*/
#define prim_val_equal(locase_type, locase_type_t) \
#define gen_prim_val_equal(locase_type, locase_type_t) \
static inline bool locase_type##_equal(const void *a, const void *b) \
{ \
return *(locase_type_t *) (a) == *(locase_type_t *) (b); \
}
#define prim_val_equal_t(locase_type) \
prim_val_equal(locase_type, locase_type##_t)
#define gen_prim_val_equal_t(locase_type) \
gen_prim_val_equal(locase_type, locase_type##_t)
prim_val_equal_t(int8);
prim_val_equal_t(int16);
prim_val_equal_t(int32);
prim_val_equal_t(int64);
gen_prim_val_equal_t(int8); // generate function int8_equal
gen_prim_val_equal_t(int16); // generate function int16_equal
gen_prim_val_equal_t(int32); // generate function int32_equal
gen_prim_val_equal_t(int64); // generate function int64_equal
prim_val_equal_t(uint8);
prim_val_equal_t(uint16);
prim_val_equal_t(uint32);
prim_val_equal_t(uint64);
gen_prim_val_equal_t(uint8); // generate function uint8_equal
gen_prim_val_equal_t(uint16); // generate function uint16_equal
gen_prim_val_equal_t(uint32); // generate function uint32_equal
gen_prim_val_equal_t(uint64); // generate function uint64_equal
prim_val_equal(float, float);
prim_val_equal(double, double);
gen_prim_val_equal(float, float); // generate function float_equal
gen_prim_val_equal(double, double); // generate function double_equal
/*
* define functions for implement compare function for primitive value
*/
#define gen_prim_val_compare(locase_type, locase_type_t) \
static inline bool locase_type##_compare(const void *a, const void *b) \
{ \
locase_type_t val_a = *(locase_type_t *) (a); \
locase_type_t val_b = *(locase_type_t *) (b); \
return val_a == val_b ? NEU_ORDER_EQ \
: val_a > val_b ? NEU_ORDER_MORE \
: NEU_ORDER_LESS; \
}
#define gen_prim_val_compare_t(locase_type) \
gen_prim_val_compare(locase_type, locase_type##_t)
gen_prim_val_compare_t(int8); // generate function int8_compare
gen_prim_val_compare_t(int16); // generate function int16_compare
gen_prim_val_compare_t(int32); // generate function int32_compare
gen_prim_val_compare_t(int64); // generate function int64_compare
gen_prim_val_compare_t(uint8); // generate function uint8_compare
gen_prim_val_compare_t(uint16); // generate function uint16_compare
gen_prim_val_compare_t(uint32); // generate function uint32_compare
gen_prim_val_compare_t(uint64); // generate function uint64_compare
gen_prim_val_compare(float, float); // generate function float_compare
gen_prim_val_compare(double, double); // generate function double_compare
/**
* Boolean
src/adapter/adapter.c
+63 -9
View File
@@ -240,7 +240,7 @@ static void adapter_loop(void *arg)
req.buf_len = sizeof(neu_reqresp_data_t);
req.buf = (void *) &data_req;
intf_funs->request(adapter->plugin, &req);
// TODO: free the data_var of neu_variable_t
neu_variable_destroy(data_req.data_var);
} else {
neu_taggrp_cfg_free(databuf_ptr->grp_config);
}
@@ -248,6 +248,52 @@ static void adapter_loop(void *arg)
break;
}
case MSG_CMD_SUBSCRIBE_NODE: {
subscribe_node_cmd_t *cmd_ptr;
cmd_ptr = (subscribe_node_cmd_t *) msg_get_buf_ptr(pay_msg);
const neu_plugin_intf_funs_t *intf_funs;
neu_request_t req;
if (adapter->plugin_module) {
neu_reqresp_subscribe_node_t sub_node_req;
sub_node_req.grp_config = cmd_ptr->grp_config;
sub_node_req.dst_node_id = cmd_ptr->dst_node_id;
intf_funs = adapter->plugin_module->intf_funs;
req.req_id = adapter_get_req_id(adapter);
req.req_type = NEU_REQRESP_SUBSCRIBE_NODE;
req.buf_len = sizeof(neu_reqresp_subscribe_node_t);
req.buf = (void *) &sub_node_req;
intf_funs->request(adapter->plugin, &req);
} else {
neu_taggrp_cfg_free(cmd_ptr->grp_config);
}
break;
}
case MSG_CMD_UNSUBSCRIBE_NODE: {
unsubscribe_node_cmd_t *cmd_ptr;
cmd_ptr = (unsubscribe_node_cmd_t *) msg_get_buf_ptr(pay_msg);
const neu_plugin_intf_funs_t *intf_funs;
neu_request_t req;
if (adapter->plugin_module) {
neu_reqresp_unsubscribe_node_t unsub_node_req;
unsub_node_req.grp_config = cmd_ptr->grp_config;
unsub_node_req.dst_node_id = cmd_ptr->dst_node_id;
intf_funs = adapter->plugin_module->intf_funs;
req.req_id = adapter_get_req_id(adapter);
req.req_type = NEU_REQRESP_UNSUBSCRIBE_NODE;
req.buf_len = sizeof(neu_reqresp_unsubscribe_node_t);
req.buf = (void *) &unsub_node_req;
intf_funs->request(adapter->plugin, &req);
} else {
neu_taggrp_cfg_free(cmd_ptr->grp_config);
}
break;
}
case MSG_CMD_READ_DATA: {
read_data_cmd_t *cmd_ptr;
cmd_ptr = (read_data_cmd_t *) msg_get_buf_ptr(pay_msg);
@@ -291,7 +337,7 @@ static void adapter_loop(void *arg)
req.buf_len = sizeof(neu_reqresp_write_t);
req.buf = (void *) &write_req;
intf_funs->request(adapter->plugin, &req);
// TODO: free the data_var of neu_variable_t
neu_variable_destroy(write_req.data_var);
} else {
neu_taggrp_cfg_free(cmd_ptr->grp_config);
}
@@ -336,16 +382,23 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd,
log_info("Get command from plugin %d", cmd->req_type);
switch (cmd->req_type) {
case NEU_REQRESP_READ_DATA: {
ADAPTER_SEND_MSG(adapter, cmd, rv, MSG_CMD_READ_DATA, read_data_cmd_t,
neu_reqresp_read_t, {});
case NEU_REQRESP_SUBSCRIBE_NODE: {
ADAPTER_SEND_MSG(adapter, cmd, rv, MSG_CMD_SUBSCRIBE_NODE,
subscribe_node_cmd_t, neu_reqresp_subscribe_node_t,
{});
break;
}
case NEU_REQRESP_START_PERIODIC_READ: {
ADAPTER_SEND_MSG(adapter, cmd, rv, MSG_CMD_START_PERIODIC_READ,
start_periodic_read_cmd_t,
neu_reqresp_start_periodic_rd_t, {});
case NEU_REQRESP_UNSUBSCRIBE_NODE: {
ADAPTER_SEND_MSG(adapter, cmd, rv, MSG_CMD_UNSUBSCRIBE_NODE,
unsubscribe_node_cmd_t, neu_reqresp_unsubscribe_node_t,
{});
break;
}
case NEU_REQRESP_READ_DATA: {
ADAPTER_SEND_MSG(adapter, cmd, rv, MSG_CMD_READ_DATA, read_data_cmd_t,
neu_reqresp_read_t, {});
break;
}
@@ -654,6 +707,7 @@ static int adapter_response(neu_adapter_t *adapter, neu_response_t *resp)
neu_databuf->databuf = databuf;
nng_sendmsg(adapter->sock, read_msg, 0);
}
neu_variable_destroy(neu_data->data_var);
break;
}
src/core/neu_manager.c
+33 -80
View File
@@ -805,6 +805,7 @@ static int add_grp_config_to_adapter(adapter_reg_entity_t *reg_entity,
return rv;
}
/*
static int manager_add_config_by_name(neu_manager_t * manager,
config_add_cmd_t *cmd)
{
@@ -825,19 +826,46 @@ add_config_by_name_exit:
nng_mtx_unlock(manager->adapters_mtx);
return rv;
}
*/
static void add_default_grp_configs(neu_manager_t *manager)
{
uint32_t i;
config_add_cmd_t * config_add_cmd;
neu_taggrp_config_t *grp_config;
uint32_t i;
config_add_cmd_t * config_add_cmd;
neu_taggrp_config_t * grp_config;
adapter_reg_entity_t *reg_entity;
for (i = 0; i < DEFAULT_GROUP_CONFIG_COUNT; i++) {
config_add_cmd = &default_config_add_cmds[i];
grp_config = neu_taggrp_cfg_new(config_add_cmd->config_name);
neu_taggrp_cfg_set_interval(grp_config, config_add_cmd->read_interval);
config_add_cmd->grp_config = grp_config;
manager_add_config_by_name(manager, config_add_cmd);
reg_entity = find_reg_adapter_by_name(&manager->reg_adapters,
config_add_cmd->dst_adapter_name);
add_grp_config_to_adapter(reg_entity, grp_config);
sub_grp_config_with_adapter(manager, grp_config, reg_entity);
}
return;
}
static void remove_default_grp_configs(neu_manager_t *manager)
{
uint32_t i;
config_add_cmd_t * config_add_cmd;
neu_taggrp_config_t * grp_config;
adapter_reg_entity_t *reg_entity;
for (i = 0; i < DEFAULT_GROUP_CONFIG_COUNT; i++) {
config_add_cmd = &default_config_add_cmds[i];
grp_config = config_add_cmd->grp_config;
nng_mtx_lock(manager->adapters_mtx);
reg_entity = find_reg_adapter_by_name(&manager->reg_adapters,
config_add_cmd->dst_adapter_name);
unsub_grp_config_with_adapter(manager, grp_config, reg_entity);
neu_datatag_mng_del_grp_config(reg_entity->datatag_manager,
config_add_cmd->config_name);
nng_mtx_unlock(manager->adapters_mtx);
}
return;
}
@@ -1056,82 +1084,6 @@ static void manager_loop(void *arg)
break;
}
/* clang-format off
case MSG_CMD_START_PERIODIC_READ: {
size_t msg_size;
nng_msg * out_msg;
nng_pipe msg_pipe;
adapter_id_t adapter_id;
start_periodic_read_cmd_t *cmd_ptr;
adapter_reg_entity_t * reg_entity;
cmd_ptr = (start_periodic_read_cmd_t *) msg_get_buf_ptr(pay_msg);
nng_mtx_lock(manager->adapters_mtx);
adapter_id = neu_manager_adapter_id_from_node_id(
manager, cmd_ptr->dst_node_id);
reg_entity =
find_reg_adapter_by_id(&manager->reg_adapters, adapter_id);
msg_pipe = reg_entity->adapter_pipe;
nng_mtx_unlock(manager->adapters_mtx);
msg_size =
msg_inplace_data_get_size(sizeof(start_periodic_read_cmd_t));
rv = nng_msg_alloc(&out_msg, msg_size);
if (rv == 0) {
message_t * msg_ptr;
start_periodic_read_cmd_t *out_cmd_ptr;
msg_ptr = (message_t *) nng_msg_body(out_msg);
msg_inplace_data_init(msg_ptr, MSG_CMD_START_PERIODIC_READ,
sizeof(start_periodic_read_cmd_t));
out_cmd_ptr = msg_get_buf_ptr(msg_ptr);
memcpy(out_cmd_ptr, cmd_ptr, sizeof(start_periodic_read_cmd_t));
nng_msg_set_pipe(out_msg, msg_pipe);
log_info(
"Forward start periodic read command to driver pipe: %d",
msg_pipe);
nng_sendmsg(manager_bind->mng_sock, out_msg, 0);
}
break;
}
case MSG_CMD_STOP_PERIODIC_READ: {
size_t msg_size;
nng_msg * out_msg;
nng_pipe msg_pipe;
adapter_id_t adapter_id;
stop_periodic_read_cmd_t *cmd_ptr;
adapter_reg_entity_t * reg_entity;
cmd_ptr = (stop_periodic_read_cmd_t *) msg_get_buf_ptr(pay_msg);
nng_mtx_lock(manager->adapters_mtx);
adapter_id = neu_manager_adapter_id_from_node_id(
manager, cmd_ptr->dst_node_id);
reg_entity =
find_reg_adapter_by_id(&manager->reg_adapters, adapter_id);
msg_pipe = reg_entity->adapter_pipe;
nng_mtx_unlock(manager->adapters_mtx);
msg_size =
msg_inplace_data_get_size(sizeof(stop_periodic_read_cmd_t));
rv = nng_msg_alloc(&out_msg, msg_size);
if (rv == 0) {
message_t * msg_ptr;
stop_periodic_read_cmd_t *out_cmd_ptr;
msg_ptr = (message_t *) nng_msg_body(out_msg);
msg_inplace_data_init(msg_ptr, MSG_CMD_STOP_PERIODIC_READ,
sizeof(stop_periodic_read_cmd_t));
out_cmd_ptr = msg_get_buf_ptr(msg_ptr);
memcpy(out_cmd_ptr, cmd_ptr, sizeof(stop_periodic_read_cmd_t));
nng_msg_set_pipe(out_msg, msg_pipe);
log_info(
"Forward stop periodic read command to driver pipe: %d",
msg_pipe);
nng_sendmsg(manager_bind->mng_sock, out_msg, 0);
}
break;
}
clang-format on */
case MSG_CMD_SUBSCRIBE_NODE: {
size_t msg_size;
nng_msg * out_msg;
@@ -1270,6 +1222,7 @@ static void manager_loop(void *arg)
}
log_info("End message loop of neu_manager");
remove_default_grp_configs(manager);
stop_and_unreg_bind_adapters(manager);
unregister_all_reg_plugins(manager);
nng_close(manager_bind->mng_sock);