diff --git a/include/neu_adapter.h b/include/neu_adapter.h index 7ca4e057..5cd562a4 100644 --- a/include/neu_adapter.h +++ b/include/neu_adapter.h @@ -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 { diff --git a/include/neu_types.h b/include/neu_types.h index d97387a2..8c8584b4 100644 --- a/include/neu_types.h +++ b/include/neu_types.h @@ -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 diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index c4d29f72..1e8b3e2f 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -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; } diff --git a/src/core/neu_manager.c b/src/core/neu_manager.c index 6c119259..ee63def7 100644 --- a/src/core/neu_manager.c +++ b/src/core/neu_manager.c @@ -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);