diff --git a/ft/api_http.resource b/ft/api_http.resource index 8d9ac43f..da13dd43 100644 --- a/ft/api_http.resource +++ b/ft/api_http.resource @@ -64,13 +64,6 @@ Get Nodes [RETURN] ${res}[0] -Update Node - [Arguments] ${node_id} ${node_name} - PUT /api/v2/node {"id": ${node_id}, "name": "${node_name}"} - ${res} = Object response body - - [RETURN] ${res}[0] - Del Node [Arguments] ${node_id} Delete /api/v2/node {"id": ${node_id}} diff --git a/ft/api_mqtt.resource b/ft/api_mqtt.resource index 2654e76f..11e2cecc 100644 --- a/ft/api_mqtt.resource +++ b/ft/api_mqtt.resource @@ -133,12 +133,6 @@ Get Nodes [RETURN] ${res} -Update Node - [Arguments] ${node_id} ${node_name} - ${res} = Pub And Recv ${MQTT_CMD_UPDATE} {"id": ${node_id}, "name": "${node_name}"} ${TOPIC_NODE_REQ} ${TOPIC_NODE_RES} - - [RETURN] ${res} - Del Node [Arguments] ${node_id} ${res} = Pub And Recv ${MQTT_CMD_DELETE} {"id": ${node_id}} ${TOPIC_NODE_REQ} ${TOPIC_NODE_RES} diff --git a/ft/node_api_test.robot b/ft/node_api_test.robot index c907fee2..acc219a6 100644 --- a/ft/node_api_test.robot +++ b/ft/node_api_test.robot @@ -150,26 +150,6 @@ Add wrong settings to node, it should return failure Check Response Status ${res} 400 Check Error Code ${res} ${ERR_NODE_SETTING_INVALID} -Update a node that does not exist, it should return failure - ${res} = Update Node ${99} test-name - - Check Response Status ${res} 404 - Check Error Code ${res} ${ERR_NODE_NOT_EXIST} - -Update the name of the node, it should return success - ${app_node_id} = Add Node And Return ID ${NODE_MQTT} app-node ${PLUGIN_MQTT} - ${res} = Update Node ${app_node_id} test-name - - Check Response Status ${res} 200 - Check Error Code ${res} ${ERR_SUCCESS} - - ${res} = Get Nodes ${NODE_MQTT} - Check Response Status ${res} 200 - Node With Name Should Exist ${res}[nodes] test-name - - [Teardown] Del Node Check ${app_node_id} - - Start an unconfigured node, it should return failure ${modbus_node_id} = Add Node And Return ID ${NODE_DRIVER} modbus-test-node ${PLUGIN_MODBUS_TCP} diff --git a/include/neuron/adapter_msg.h b/include/neuron/adapter_msg.h index f7354f53..018f5008 100644 --- a/include/neuron/adapter_msg.h +++ b/include/neuron/adapter_msg.h @@ -54,7 +54,6 @@ typedef enum neu_reqresp_type { NEU_REQRESP_TRANS_DATA, NEU_REQRESP_ADD_NODE, NEU_REQRESP_DEL_NODE, - NEU_REQRESP_UPDATE_NODE, NEU_REQRESP_GET_NODES, NEU_REQRESP_NODES, NEU_REQRESP_ADD_GRP_CONFIG, @@ -152,12 +151,6 @@ typedef struct neu_cmd_del_node { neu_node_id_t node_id; } neu_cmd_del_node_t; -/* NEU_REQRESP_UPDATE_NODE */ -typedef struct neu_cmd_update_node { - neu_node_id_t node_id; - const char * node_name; -} neu_cmd_update_node_t; - /* NEU_REQRESP_GET_NODES */ typedef struct neu_cmd_get_nodes { neu_node_type_e node_type; diff --git a/include/neuron/plugin.h b/include/neuron/plugin.h index aa525426..4fb4d539 100644 --- a/include/neuron/plugin.h +++ b/include/neuron/plugin.h @@ -104,8 +104,6 @@ neu_datatag_table_t *neu_system_get_datatags_table(neu_plugin_t *plugin, intptr_t neu_system_add_node(neu_plugin_t *plugin, neu_node_type_e node_type, const char *adapter_name, const char *plugin_name); intptr_t neu_system_del_node(neu_plugin_t *plugin, neu_node_id_t node_id); -intptr_t neu_system_update_node(neu_plugin_t *plugin, neu_node_id_t node_id, - const char *node_name); // uninit vector vector_t neu_system_get_nodes(neu_plugin_t *plugin, neu_node_type_e node_type); intptr_t neu_system_add_group_config(neu_plugin_t * plugin, diff --git a/plugins/mqtt/command/command.c b/plugins/mqtt/command/command.c index 304f267a..22e7c55d 100644 --- a/plugins/mqtt/command/command.c +++ b/plugins/mqtt/command/command.c @@ -47,19 +47,6 @@ static int node_response_add(char **output_str, neu_plugin_t *plugin, return rc; } -static int node_response_update(char **output_str, neu_plugin_t *plugin, - neu_json_mqtt_t *mqtt, char *json_str) -{ - neu_json_update_node_req_t *req = NULL; - int rc = neu_json_decode_update_node_req(json_str, &req); - if (0 == rc) { - *output_str = command_node_update(plugin, mqtt, req); - neu_json_decode_update_node_req_free(req); - } - - return rc; -} - static int node_response_delete(char **output_str, neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, char *json_str) { @@ -427,8 +414,6 @@ void command_response_handle(mqtt_response_t *response) rc = node_response_get(&ret_str, plugin, mqtt, json_str); } else if (0 == strcmp(NEU_MQTT_CMD_ADD, mqtt->command)) { rc = node_response_add(&ret_str, plugin, mqtt, json_str); - } else if (0 == strcmp(NEU_MQTT_CMD_UPDATE, mqtt->command)) { - rc = node_response_update(&ret_str, plugin, mqtt, json_str); } else if (0 == strcmp(NEU_MQTT_CMD_DELETE, mqtt->command)) { rc = node_response_delete(&ret_str, plugin, mqtt, json_str); } diff --git a/plugins/mqtt/command/node.c b/plugins/mqtt/command/node.c index 40522f61..617b10d1 100644 --- a/plugins/mqtt/command/node.c +++ b/plugins/mqtt/command/node.c @@ -77,19 +77,6 @@ char *command_node_add(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, return result; } -char *command_node_update(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, - neu_json_update_node_req_t *req) -{ - log_info("Update node uuid:%s, node name:%s", mqtt->uuid, req->name); - neu_json_error_resp_t error = { 0 }; - error.error = neu_system_update_node(plugin, req->id, req->name); - - char *result = NULL; - neu_json_encode_with_mqtt(&error, neu_json_encode_error_resp, mqtt, - neu_json_encode_mqtt_resp, &result); - return result; -} - char *command_node_delete(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, neu_json_del_node_req_t *req) { diff --git a/plugins/mqtt/command/node.h b/plugins/mqtt/command/node.h index 1f1baba1..c19f529e 100644 --- a/plugins/mqtt/command/node.h +++ b/plugins/mqtt/command/node.h @@ -32,8 +32,6 @@ char *command_node_get(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, neu_json_get_nodes_req_t *req); char *command_node_add(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, neu_json_add_node_req_t *req); -char *command_node_update(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, - neu_json_update_node_req_t *req); char *command_node_delete(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, neu_json_del_node_req_t *req); char *command_node_setting_set(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, diff --git a/plugins/restful/adapter_handle.c b/plugins/restful/adapter_handle.c index a410158c..45fd27e2 100644 --- a/plugins/restful/adapter_handle.c +++ b/plugins/restful/adapter_handle.c @@ -64,17 +64,6 @@ void handle_del_adapter(nng_aio *aio) }) } -void handle_update_adapter(nng_aio *aio) -{ - neu_plugin_t *plugin = neu_rest_get_plugin(); - - REST_PROCESS_HTTP_REQUEST_VALIDATE_JWT( - aio, neu_json_update_node_req_t, neu_json_decode_update_node_req, - { NEU_JSON_RESPONSE_ERROR( - neu_system_update_node(plugin, req->id, req->name), - { http_response(aio, error_code.error, result_error); }) }) -} - void handle_get_adapter(nng_aio *aio) { neu_plugin_t * plugin = neu_rest_get_plugin(); diff --git a/plugins/restful/adapter_handle.h b/plugins/restful/adapter_handle.h index 25ddb351..10f033f6 100644 --- a/plugins/restful/adapter_handle.h +++ b/plugins/restful/adapter_handle.h @@ -23,7 +23,6 @@ void handle_add_adapter(nng_aio *aio); void handle_del_adapter(nng_aio *aio); -void handle_update_adapter(nng_aio *aio); void handle_get_adapter(nng_aio *aio); void handle_set_node_setting(nng_aio *aio); void handle_get_node_setting(nng_aio *aio); diff --git a/plugins/restful/handle.c b/plugins/restful/handle.c index 7f08c3be..852ff4cd 100644 --- a/plugins/restful/handle.c +++ b/plugins/restful/handle.c @@ -186,12 +186,6 @@ struct neu_rest_handler api_handlers[] = { .url = "/api/v2/node", .value.handler = handle_add_adapter, }, - { - .method = NEU_REST_METHOD_PUT, - .type = NEU_REST_HANDLER_FUNCTION, - .url = "/api/v2/node", - .value.handler = handle_update_adapter, - }, { .method = NEU_REST_METHOD_DELETE, .type = NEU_REST_HANDLER_FUNCTION, diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 8f651a35..bc5845f0 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -68,7 +68,7 @@ cmd_ptr->grp_config = reqresp_cmd->grp_config; \ cmd_ptr->req_id = cmd->req_id; \ { func }; \ - nng_sendmsg((adapter)->sock, msg, 0); \ + nng_sendmsg((adapter)->nng.sock, msg, 0); \ } \ } @@ -84,7 +84,7 @@ msg_inplace_data_init(msg_ptr, (msg_type), (len)); \ char *dst = msg_get_buf_ptr(msg_ptr); \ memcpy(dst, (src), len); \ - nng_sendmsg((adapter)->sock, msg, 0); \ + nng_sendmsg((adapter)->nng.sock, msg, 0); \ } \ } @@ -740,15 +740,7 @@ neu_adapter_state_to_plugin_state(neu_adapter_t *adapter) static uint32_t adapter_get_req_id(neu_adapter_t *adapter) { - uint32_t req_id; - - adapter->new_req_id++; - if (adapter->new_req_id == 0) { - adapter->new_req_id = 1; - } - - req_id = adapter->new_req_id; - return req_id; + return adapter->req_id += 2; } static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd, @@ -834,36 +826,6 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd, break; } - case NEU_REQRESP_UPDATE_NODE: { - ADAPTER_RESP_CODE( - adapter, cmd, intptr_t, neu_cmd_update_node_t, rv, - NEU_REQRESP_ERR_CODE, p_result, { - neu_event_update_node_t event = {}; - char * src_name = NULL; - char * dst_name = NULL; - - ret = neu_manager_get_node_name_by_id( - adapter->manager, req_cmd->node_id, &src_name); - if (0 == ret) { - dst_name = strdup(req_cmd->node_name); - ret = NULL == dst_name ? NEU_ERR_ENOMEM : 0; - } - if (0 == ret) { - ret = neu_manager_update_node(adapter->manager, req_cmd); - } - if (0 == ret) { - event.src_name = src_name; - event.dst_name = dst_name; - ADAPTER_SEND_BUF(adapter, rv, MSG_EVENT_UPDATE_NODE, &event, - sizeof(event)); - } else { - free(src_name); - free(dst_name); - } - }); - break; - } - case NEU_REQRESP_GET_NODES: { ADAPTER_RESP_CMD( adapter, cmd, neu_reqresp_nodes_t, neu_cmd_get_nodes_t, rv, @@ -1245,7 +1207,7 @@ static int adapter_response(neu_adapter_t *adapter, neu_response_t *resp) nng_msg_free(trans_data_msg); break; } - nng_sendmsg(adapter->sock, trans_data_msg, 0); + nng_sendmsg(adapter->nng.sock, trans_data_msg, 0); } break; } @@ -1280,7 +1242,7 @@ static int adapter_response(neu_adapter_t *adapter, neu_response_t *resp) nng_msg_free(read_resp_msg); break; } - nng_sendmsg(adapter->sock, read_resp_msg, 0); + nng_sendmsg(adapter->nng.sock, read_resp_msg, 0); } break; } @@ -1315,7 +1277,7 @@ static int adapter_response(neu_adapter_t *adapter, neu_response_t *resp) nng_msg_free(write_resp_msg); break; } - nng_sendmsg(adapter->sock, write_resp_msg, 0); + nng_sendmsg(adapter->nng.sock, write_resp_msg, 0); } break; } @@ -1396,35 +1358,32 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, break; } - adapter->plugin_lib = + adapter->plugin_info.handle = load_plugin_library((char *) info->plugin_lib_name, info->plugin_kind, - &adapter->plugin_module); - assert(adapter->plugin_lib != NULL); + &adapter->plugin_info.module); + assert(adapter->plugin_info.handle != NULL); - adapter->id = info->id; - adapter->type = info->type; - adapter->name = strdup(info->name); - adapter->state = ADAPTER_STATE_IDLE; - adapter->new_req_id = 1; - adapter->plugin_id = info->plugin_id; - adapter->plugin_kind = info->plugin_kind; - adapter->manager = manager; - adapter->trans_kind = NEURON_TRANS_DATAVAL; - adapter->node_id = neu_manager_adapter_id_to_node_id(manager, info->id); - adapter->stop = false; - adapter->cb_funs = callback_funs; + adapter->id = info->id; + adapter->name = strdup(info->name); + adapter->state = ADAPTER_STATE_IDLE; + adapter->req_id = 1; + adapter->plugin_info.id = info->plugin_id; + adapter->plugin_info.kind = info->plugin_kind; + adapter->manager = manager; + adapter->trans_kind = NEURON_TRANS_DATAVAL; + adapter->cb_funs = callback_funs; - rv = nng_mtx_alloc(&adapter->mtx); + rv = nng_mtx_alloc(&adapter->nng.mtx); assert(rv == 0); - rv = nng_mtx_alloc(&adapter->sub_grp_mtx); + rv = nng_mtx_alloc(&adapter->nng.sub_grp_mtx); assert(rv == 0); vector_init(&adapter->sub_grp_configs, 0, sizeof(neu_sub_grp_config_t)); - adapter->plugin = - adapter->plugin_module->intf_funs->open(adapter, &adapter->cb_funs); - assert(adapter->plugin != NULL); - assert(neu_plugin_common_check(adapter->plugin)); + adapter->plugin_info.plugin = adapter->plugin_info.module->intf_funs->open( + adapter, &adapter->cb_funs); + assert(adapter->plugin_info.plugin != NULL); + assert(neu_plugin_common_check(adapter->plugin_info.plugin)); adapter->events = neu_event_new(); @@ -1437,28 +1396,27 @@ void neu_adapter_destroy(neu_adapter_t *adapter) { assert(adapter != NULL); - if (adapter->type == ADAPTER_TYPE_DRIVERX) { + if (adapter->plugin_info.module->type == NEU_NODE_TYPE_DRIVERX) { neu_adapter_driver_destroy((neu_adapter_driver_t *) adapter); } - if (adapter->plugin_module != NULL) { - adapter->plugin_module->intf_funs->close(adapter->plugin); - } - if (adapter->plugin_lib != NULL) { - unload_plugin_library(adapter->plugin_lib, adapter->plugin_kind); + adapter->plugin_info.module->intf_funs->close(adapter->plugin_info.plugin); + if (adapter->plugin_info.handle != NULL) { + unload_plugin_library(adapter->plugin_info.handle, + adapter->plugin_info.kind); } if (adapter->name != NULL) { free(adapter->name); } - if (NULL != adapter->node_setting.buf) { - free(adapter->node_setting.buf); + if (NULL != adapter->setting.buf) { + free(adapter->setting.buf); } - nng_mtx_lock(adapter->sub_grp_mtx); + nng_mtx_lock(adapter->nng.sub_grp_mtx); vector_uninit(&adapter->sub_grp_configs); - nng_mtx_unlock(adapter->sub_grp_mtx); + nng_mtx_unlock(adapter->nng.sub_grp_mtx); - nng_mtx_free(adapter->mtx); - nng_mtx_free(adapter->sub_grp_mtx); + nng_mtx_free(adapter->nng.mtx); + nng_mtx_free(adapter->nng.sub_grp_mtx); neu_event_close(adapter->events); free(adapter); } @@ -1473,12 +1431,11 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) switch (type) { case NEU_EVENT_IO_READ: - rv = nng_recvmsg(adapter->sock, &msg, 0); + rv = nng_recvmsg(adapter->nng.sock, &msg, 0); if (rv == NNG_ECLOSED) { - nng_mtx_lock(adapter->mtx); + nng_mtx_lock(adapter->nng.mtx); adapter->state = ADAPTER_STATE_IDLE; - adapter->stop = true; - nng_mtx_unlock(adapter->mtx); + nng_mtx_unlock(adapter->nng.mtx); log_warn("nng socket closed."); return 0; @@ -1491,7 +1448,8 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) case MSG_CMD_RESP_PONG: { char *buf_ptr = msg_get_buf_ptr(pay_msg); - adapter->plugin_module->intf_funs->init(adapter->plugin); + adapter->plugin_info.module->intf_funs->init( + adapter->plugin_info.plugin); adapter->state = ADAPTER_STATE_INIT; log_info("Adapter(%s) received pong: %s\n", adapter->name, buf_ptr); break; @@ -1499,11 +1457,9 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) case MSG_CMD_EXIT_LOOP: { uint32_t exit_code = *(uint32_t *) msg_get_buf_ptr(pay_msg); - adapter->plugin_module->intf_funs->uninit(adapter->plugin); - nng_mtx_lock(adapter->mtx); + nng_mtx_lock(adapter->nng.mtx); adapter->state = ADAPTER_STATE_IDLE; - adapter->stop = true; - nng_mtx_unlock(adapter->mtx); + nng_mtx_unlock(adapter->nng.mtx); log_info("adapter(%s) exit loop by exit_code=%d", adapter->name, exit_code); @@ -1528,14 +1484,14 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) break; } - intf_funs = adapter->plugin_module->intf_funs; + intf_funs = adapter->plugin_info.module->intf_funs; req.req_id = adapter_get_req_id(adapter); req.req_type = NEU_REQRESP_TRANS_DATA; 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); + intf_funs->request(adapter->plugin_info.plugin, &req); if (is_need_free) { neu_dvalue_free(data_req.data_val); } @@ -1648,13 +1604,13 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) const neu_plugin_intf_funs_t *intf_funs = NULL; neu_request_t req = {}; - intf_funs = adapter->plugin_module->intf_funs; + intf_funs = adapter->plugin_info.module->intf_funs; req.req_id = adapter_get_req_id(adapter); req.req_type = NEU_REQRESP_UPDATE_LICENSE; req.sender_id = 0; req.buf_len = 0; req.buf = NULL; - intf_funs->request(adapter->plugin, &req); + intf_funs->request(adapter->plugin_info.plugin, &req); break; } @@ -1670,17 +1626,17 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) sub_node_req.dst_node_id = cmd_ptr->dst_node_id; sub_node_req.src_node_id = cmd_ptr->src_node_id; - intf_funs = adapter->plugin_module->intf_funs; + intf_funs = adapter->plugin_info.module->intf_funs; req.req_id = adapter_get_req_id(adapter); req.req_type = NEU_REQRESP_SUBSCRIBE_NODE; req.sender_id = cmd_ptr->sender_id; req.buf_len = sizeof(neu_reqresp_subscribe_node_t); req.buf = (void *) &sub_node_req; - if (adapter->type == ADAPTER_TYPE_DRIVERX) { + if (adapter->plugin_info.module->type == NEU_NODE_TYPE_DRIVERX) { neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, &req); } else { - intf_funs->request(adapter->plugin, &req); + intf_funs->request(adapter->plugin_info.plugin, &req); } break; } @@ -1697,17 +1653,17 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) unsub_node_req.dst_node_id = cmd_ptr->dst_node_id; unsub_node_req.src_node_id = cmd_ptr->src_node_id; - intf_funs = adapter->plugin_module->intf_funs; + intf_funs = adapter->plugin_info.module->intf_funs; req.req_id = adapter_get_req_id(adapter); req.req_type = NEU_REQRESP_UNSUBSCRIBE_NODE; req.sender_id = cmd_ptr->sender_id; req.buf_len = sizeof(neu_reqresp_unsubscribe_node_t); req.buf = (void *) &unsub_node_req; - if (adapter->type == ADAPTER_TYPE_DRIVERX) { + if (adapter->plugin_info.module->type == NEU_NODE_TYPE_DRIVERX) { neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, &req); } else { - intf_funs->request(adapter->plugin, &req); + intf_funs->request(adapter->plugin_info.plugin, &req); } break; } @@ -1723,17 +1679,17 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) read_req.grp_config = cmd_ptr->grp_config; read_req.dst_node_id = cmd_ptr->dst_node_id; - intf_funs = adapter->plugin_module->intf_funs; + intf_funs = adapter->plugin_info.module->intf_funs; req.req_id = cmd_ptr->req_id; req.req_type = NEU_REQRESP_READ_DATA; req.sender_id = cmd_ptr->sender_id; req.buf_len = sizeof(neu_reqresp_read_t); req.buf = (void *) &read_req; - if (adapter->type == ADAPTER_TYPE_DRIVERX) { + if (adapter->plugin_info.module->type == NEU_NODE_TYPE_DRIVERX) { neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, &req); } else { - intf_funs->request(adapter->plugin, &req); + intf_funs->request(adapter->plugin_info.plugin, &req); } break; } @@ -1758,13 +1714,13 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) break; } - intf_funs = adapter->plugin_module->intf_funs; + intf_funs = adapter->plugin_info.module->intf_funs; req.req_id = cmd_ptr->req_id; req.req_type = NEU_REQRESP_READ_RESP; req.sender_id = cmd_ptr->sender_id; req.buf_len = sizeof(neu_reqresp_read_resp_t); req.buf = (void *) &read_resp; - intf_funs->request(adapter->plugin, &req); + intf_funs->request(adapter->plugin_info.plugin, &req); if (is_need_free) { neu_dvalue_free(read_resp.data_val); } @@ -1793,17 +1749,17 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) break; } - intf_funs = adapter->plugin_module->intf_funs; + intf_funs = adapter->plugin_info.module->intf_funs; req.req_id = cmd_ptr->req_id; req.req_type = NEU_REQRESP_WRITE_DATA; req.sender_id = cmd_ptr->sender_id; req.buf_len = sizeof(neu_reqresp_write_t); req.buf = (void *) &write_req; - if (adapter->type == ADAPTER_TYPE_DRIVERX) { + if (adapter->plugin_info.module->type == NEU_NODE_TYPE_DRIVERX) { neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, &req); } else { - intf_funs->request(adapter->plugin, &req); + intf_funs->request(adapter->plugin_info.plugin, &req); } if (is_need_free) { neu_dvalue_free(write_req.data_val); @@ -1832,13 +1788,13 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) break; } - intf_funs = adapter->plugin_module->intf_funs; + intf_funs = adapter->plugin_info.module->intf_funs; req.req_id = cmd_ptr->req_id; req.req_type = NEU_REQRESP_WRITE_RESP; req.sender_id = cmd_ptr->sender_id; req.buf_len = sizeof(neu_reqresp_write_resp_t); req.buf = (void *) &write_resp; - intf_funs->request(adapter->plugin, &req); + intf_funs->request(adapter->plugin_info.plugin, &req); if (is_need_free) { neu_dvalue_free(write_resp.data_val); } @@ -1865,7 +1821,7 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) int neu_adapter_init(neu_adapter_t *adapter) { - int rv = nng_pair1_open(&adapter->sock); + int rv = nng_pair1_open(&adapter->nng.sock); neu_event_io_param_t param = { .usr_data = (void *) adapter, @@ -1874,17 +1830,17 @@ int neu_adapter_init(neu_adapter_t *adapter) assert(rv == 0); - nng_socket_get_int(adapter->sock, NNG_OPT_RECVFD, &adapter->recv_fd); + nng_socket_get_int(adapter->nng.sock, NNG_OPT_RECVFD, &adapter->recv_fd); param.fd = adapter->recv_fd; adapter->nng_io = neu_event_add_io(adapter->events, param); - if (adapter->type == ADAPTER_TYPE_DRIVERX) { + if (adapter->plugin_info.module->type == NEU_NODE_TYPE_DRIVERX) { neu_adapter_driver_init((neu_adapter_driver_t *) adapter); } - rv = nng_dial(adapter->sock, neu_manager_get_url(adapter->manager), - &adapter->dialer, 0); + rv = nng_dial(adapter->nng.sock, neu_manager_get_url(adapter->manager), + &adapter->nng.dialer, 0); assert(rv == 0); nng_msg *out_msg; @@ -1899,7 +1855,7 @@ int neu_adapter_init(neu_adapter_t *adapter) buf_ptr = msg_get_buf_ptr(msg_ptr); memcpy(buf_ptr, adapter->name, strlen(adapter->name)); buf_ptr[strlen(adapter->name)] = 0; - nng_sendmsg(adapter->sock, out_msg, 0); + nng_sendmsg(adapter->nng.sock, out_msg, 0); } return rv; } @@ -1908,18 +1864,15 @@ int neu_adapter_uninit(neu_adapter_t *adapter) { int rv = 0; - if (adapter->type == ADAPTER_TYPE_DRIVERX) { + adapter->plugin_info.module->intf_funs->uninit(adapter->plugin_info.plugin); + if (adapter->plugin_info.module->type == NEU_NODE_TYPE_DRIVERX) { neu_adapter_driver_uninit((neu_adapter_driver_t *) adapter); } - nng_mtx_lock(adapter->mtx); - adapter->stop = true; - nng_mtx_unlock(adapter->mtx); - neu_event_del_io(adapter->events, adapter->nng_io); - nng_dialer_close(adapter->dialer); - nng_close(adapter->sock); + nng_dialer_close(adapter->nng.dialer); + nng_close(adapter->nng.sock); log_info("Stop the adapter(%s)", adapter->name); return rv; @@ -1928,10 +1881,11 @@ int neu_adapter_uninit(neu_adapter_t *adapter) int neu_adapter_start(neu_adapter_t *adapter) { // TODO start by send msg - const neu_plugin_intf_funs_t *intf_funs = adapter->plugin_module->intf_funs; - neu_err_code_e error = NEU_ERR_SUCCESS; + const neu_plugin_intf_funs_t *intf_funs = + adapter->plugin_info.module->intf_funs; + neu_err_code_e error = NEU_ERR_SUCCESS; - nng_mtx_lock(adapter->mtx); + nng_mtx_lock(adapter->nng.mtx); switch (adapter->state) { case ADAPTER_STATE_IDLE: case ADAPTER_STATE_INIT: @@ -1944,17 +1898,17 @@ int neu_adapter_start(neu_adapter_t *adapter) case ADAPTER_STATE_STOPPED: break; } - nng_mtx_unlock(adapter->mtx); + nng_mtx_unlock(adapter->nng.mtx); if (error != NEU_ERR_SUCCESS) { return error; } - error = intf_funs->start(adapter->plugin); + error = intf_funs->start(adapter->plugin_info.plugin); if (error == NEU_ERR_SUCCESS) { - nng_mtx_lock(adapter->mtx); + nng_mtx_lock(adapter->nng.mtx); adapter->state = ADAPTER_STATE_RUNNING; - nng_mtx_unlock(adapter->mtx); + nng_mtx_unlock(adapter->nng.mtx); } return error; @@ -1964,10 +1918,11 @@ int neu_adapter_stop(neu_adapter_t *adapter) { // TODO stop by send msg - const neu_plugin_intf_funs_t *intf_funs = adapter->plugin_module->intf_funs; - neu_err_code_e error = NEU_ERR_SUCCESS; + const neu_plugin_intf_funs_t *intf_funs = + adapter->plugin_info.module->intf_funs; + neu_err_code_e error = NEU_ERR_SUCCESS; - nng_mtx_lock(adapter->mtx); + nng_mtx_lock(adapter->nng.mtx); switch (adapter->state) { case ADAPTER_STATE_IDLE: case ADAPTER_STATE_INIT: @@ -1980,17 +1935,17 @@ int neu_adapter_stop(neu_adapter_t *adapter) case ADAPTER_STATE_RUNNING: break; } - nng_mtx_unlock(adapter->mtx); + nng_mtx_unlock(adapter->nng.mtx); if (error != NEU_ERR_SUCCESS) { return error; } - error = intf_funs->stop(adapter->plugin); + error = intf_funs->stop(adapter->plugin_info.plugin); if (error == NEU_ERR_SUCCESS) { - nng_mtx_lock(adapter->mtx); + nng_mtx_lock(adapter->nng.mtx); adapter->state = ADAPTER_STATE_STOPPED; - nng_mtx_unlock(adapter->mtx); + nng_mtx_unlock(adapter->nng.mtx); } return error; @@ -2005,12 +1960,6 @@ const char *neu_adapter_get_name(neu_adapter_t *adapter) return (const char *) adapter->name; } -void neu_adapter_rename(neu_adapter_t *adapter, const char *name) -{ - free(adapter->name); - adapter->name = strdup(name); -} - neu_manager_t *neu_adapter_get_manager(neu_adapter_t *adapter) { if (adapter == NULL) { @@ -2020,7 +1969,7 @@ neu_manager_t *neu_adapter_get_manager(neu_adapter_t *adapter) return (neu_manager_t *) adapter->manager; } -adapter_id_t neu_adapter_get_id(neu_adapter_t *adapter) +neu_adapter_id_t neu_adapter_get_id(neu_adapter_t *adapter) { if (adapter == NULL) { return 0; @@ -2030,14 +1979,14 @@ adapter_id_t neu_adapter_get_id(neu_adapter_t *adapter) } // NOTE: Do NOT expose this function in header files, use for persistence only! -adapter_id_t neu_adapter_set_id(neu_adapter_t *adapter, adapter_id_t id) +neu_adapter_id_t neu_adapter_set_id(neu_adapter_t *adapter, neu_adapter_id_t id) { if (adapter == NULL) { return 0; } - adapter_id_t old = adapter->id; - adapter->id = id; + neu_adapter_id_t old = adapter->id; + adapter->id = id; return old; } @@ -2047,7 +1996,7 @@ adapter_type_e neu_adapter_get_type(neu_adapter_t *adapter) return 0; } - return adapter->type; + return (adapter_type_e) adapter->plugin_info.module->type; } plugin_id_t neu_adapter_get_plugin_id(neu_adapter_t *adapter) @@ -2057,37 +2006,30 @@ plugin_id_t neu_adapter_get_plugin_id(neu_adapter_t *adapter) return id; } - return adapter->plugin_id; + return adapter->plugin_info.id; } int neu_adapter_set_setting(neu_adapter_t *adapter, neu_config_t *config) { int rv = -1; - if (adapter == NULL) { - log_error("Config adapter with NULL adapter"); - return NEU_ERR_NODE_NOT_EXIST; - } + const neu_plugin_intf_funs_t *intf_funs; - if (adapter->plugin_module != NULL) { - const neu_plugin_intf_funs_t *intf_funs; + intf_funs = adapter->plugin_info.module->intf_funs; + rv = intf_funs->config(adapter->plugin_info.plugin, config); + if (rv == 0) { + adapter->setting.buf_len = config->buf_len; + adapter->setting.type = config->type; - intf_funs = adapter->plugin_module->intf_funs; - rv = intf_funs->config(adapter->plugin, config); - if (rv == 0) { - adapter->node_setting.buf_len = config->buf_len; - adapter->node_setting.type = config->type; - - if (adapter->node_setting.buf != NULL) - free(adapter->node_setting.buf); - adapter->node_setting.buf = strdup(config->buf); - if (adapter->state == ADAPTER_STATE_INIT) { - adapter->state = ADAPTER_STATE_READY; - } - - } else { - rv = NEU_ERR_NODE_SETTING_INVALID; + if (adapter->setting.buf != NULL) + free(adapter->setting.buf); + adapter->setting.buf = strdup(config->buf); + if (adapter->state == ADAPTER_STATE_INIT) { + adapter->state = ADAPTER_STATE_READY; } + + } else { + rv = NEU_ERR_NODE_SETTING_INVALID; } return rv; @@ -2100,8 +2042,8 @@ int neu_adapter_get_setting(neu_adapter_t *adapter, char **config) return NEU_ERR_NODE_NOT_EXIST; } - if (adapter->node_setting.buf != NULL) { - *config = strdup(adapter->node_setting.buf); + if (adapter->setting.buf != NULL) { + *config = strdup(adapter->setting.buf); return NEU_ERR_SUCCESS; } @@ -2110,8 +2052,9 @@ int neu_adapter_get_setting(neu_adapter_t *adapter, char **config) neu_plugin_state_t neu_adapter_get_state(neu_adapter_t *adapter) { - neu_plugin_state_t state = { 0 }; - neu_plugin_common_t *common = neu_plugin_to_plugin_common(adapter->plugin); + neu_plugin_state_t state = { 0 }; + neu_plugin_common_t *common = + neu_plugin_to_plugin_common(adapter->plugin_info.plugin); state.link = common->link_state; state.running = neu_adapter_state_to_plugin_state(adapter); @@ -2126,7 +2069,7 @@ void neu_adapter_add_sub_grp_config(neu_adapter_t * adapter, bool find = false; char *name = (char *) neu_taggrp_cfg_get_name(grp_config); - nng_mtx_lock(adapter->sub_grp_mtx); + nng_mtx_lock(adapter->nng.sub_grp_mtx); VECTOR_FOR_EACH(&adapter->sub_grp_configs, iter) { @@ -2147,7 +2090,7 @@ void neu_adapter_add_sub_grp_config(neu_adapter_t * adapter, vector_push_back(&adapter->sub_grp_configs, &sgc); } - nng_mtx_unlock(adapter->sub_grp_mtx); + nng_mtx_unlock(adapter->nng.sub_grp_mtx); } void neu_adapter_del_sub_grp_config(neu_adapter_t * adapter, @@ -2156,7 +2099,7 @@ void neu_adapter_del_sub_grp_config(neu_adapter_t * adapter, { char *name = (char *) neu_taggrp_cfg_get_name(grp_config); - nng_mtx_lock(adapter->sub_grp_mtx); + nng_mtx_lock(adapter->nng.sub_grp_mtx); VECTOR_FOR_EACH(&adapter->sub_grp_configs, iter) { @@ -2170,26 +2113,27 @@ void neu_adapter_del_sub_grp_config(neu_adapter_t * adapter, } } - nng_mtx_unlock(adapter->sub_grp_mtx); + nng_mtx_unlock(adapter->nng.sub_grp_mtx); } vector_t *neu_adapter_get_sub_grp_configs(neu_adapter_t *adapter) { vector_t *sgc = NULL; - nng_mtx_lock(adapter->sub_grp_mtx); + nng_mtx_lock(adapter->nng.sub_grp_mtx); sgc = vector_clone(&adapter->sub_grp_configs); - nng_mtx_unlock(adapter->sub_grp_mtx); + nng_mtx_unlock(adapter->nng.sub_grp_mtx); return sgc; } int neu_adapter_validate_tag(neu_adapter_t *adapter, neu_datatag_t *tag) { - const neu_plugin_intf_funs_t *intf_funs = adapter->plugin_module->intf_funs; - neu_err_code_e error = NEU_ERR_SUCCESS; + const neu_plugin_intf_funs_t *intf_funs = + adapter->plugin_info.module->intf_funs; + neu_err_code_e error = NEU_ERR_SUCCESS; - error = intf_funs->validate_tag(adapter->plugin, tag); + error = intf_funs->validate_tag(adapter->plugin_info.plugin, tag); return error; } diff --git a/src/adapter/adapter_info.h b/src/adapter/adapter_info.h index 028f7a56..3d8f9b77 100644 --- a/src/adapter/adapter_info.h +++ b/src/adapter/adapter_info.h @@ -22,6 +22,7 @@ #include +#include "adapter.h" #include "plugin_info.h" typedef enum adapter_type { @@ -36,14 +37,13 @@ typedef enum adapter_type { ADAPTER_TYPE_MAX, } adapter_type_e; -typedef uint32_t adapter_id_t; typedef struct neu_adapter_info { - adapter_id_t id; - adapter_type_e type; - plugin_id_t plugin_id; - const char * name; - plugin_kind_e plugin_kind; - const char * plugin_lib_name; + neu_adapter_id_t id; + adapter_type_e type; + plugin_id_t plugin_id; + const char * name; + plugin_kind_e plugin_kind; + const char * plugin_lib_name; } neu_adapter_info_t; #endif diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h index e6b52d22..e66902af 100644 --- a/src/adapter/adapter_internal.h +++ b/src/adapter/adapter_internal.h @@ -43,33 +43,36 @@ typedef enum adapter_state { } adapter_state_e; struct neu_adapter { - adapter_id_t id; - neu_node_id_t node_id; - adapter_type_e type; - nng_mtx * mtx; - nng_mtx * sub_grp_mtx; - adapter_state_e state; - bool stop; - char * name; - neu_manager_t * manager; - nng_pipe pipe; - nng_socket sock; - nng_dialer dialer; - int recv_fd; - nng_thread * thrd; - uint32_t new_req_id; - plugin_id_t plugin_id; - plugin_kind_e plugin_kind; - void * plugin_lib; // handle of dynamic lib - neu_plugin_module_t *plugin_module; - neu_plugin_t * plugin; - neu_trans_kind_e trans_kind; - adapter_callbacks_t cb_funs; - neu_config_t node_setting; - vector_t sub_grp_configs; // neu_sub_grp_config_t + neu_adapter_id_t id; + adapter_state_e state; + char * name; + neu_manager_t * manager; + + uint32_t req_id; + + neu_trans_kind_e trans_kind; + adapter_callbacks_t cb_funs; + neu_config_t setting; + vector_t sub_grp_configs; // neu_sub_grp_config_t + + struct { + plugin_id_t id; + plugin_kind_e kind; + void * handle; // handle of dynamic lib + neu_plugin_module_t *module; + neu_plugin_t * plugin; + } plugin_info; + + struct { + nng_socket sock; + nng_dialer dialer; + nng_mtx * mtx; + nng_mtx * sub_grp_mtx; + } nng; neu_events_t * events; neu_event_io_t *nng_io; + int recv_fd; }; neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, @@ -84,12 +87,11 @@ neu_event_timer_t *neu_adapter_add_timer(neu_adapter_t * adapter, neu_event_timer_param_t param); void neu_adapter_del_timer(neu_adapter_t *adapter, neu_event_timer_t *timer); -const char * neu_adapter_get_name(neu_adapter_t *adapter); -void neu_adapter_rename(neu_adapter_t *adapter, const char *name); -neu_manager_t *neu_adapter_get_manager(neu_adapter_t *adapter); -adapter_id_t neu_adapter_get_id(neu_adapter_t *adapter); -adapter_type_e neu_adapter_get_type(neu_adapter_t *adapter); -plugin_id_t neu_adapter_get_plugin_id(neu_adapter_t *adapter); +const char * neu_adapter_get_name(neu_adapter_t *adapter); +neu_manager_t * neu_adapter_get_manager(neu_adapter_t *adapter); +neu_adapter_id_t neu_adapter_get_id(neu_adapter_t *adapter); +adapter_type_e neu_adapter_get_type(neu_adapter_t *adapter); +plugin_id_t neu_adapter_get_plugin_id(neu_adapter_t *adapter); int neu_adapter_set_setting(neu_adapter_t *adapter, neu_config_t *config); int neu_adapter_get_setting(neu_adapter_t *adapter, char **config); neu_plugin_state_t neu_adapter_get_state(neu_adapter_t *adapter); diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 4fad9c82..9bd1a0d4 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -143,7 +143,7 @@ int neu_adapter_driver_stop(neu_adapter_driver_t *driver) int neu_adapter_driver_init(neu_adapter_driver_t *driver) { driver->tag_table = neu_manager_get_datatag_tbl(driver->adapter.manager, - driver->adapter.node_id); + driver->adapter.id); return 0; } @@ -156,8 +156,8 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver) pthread_mutex_lock(&driver->grp_mtx); HASH_ITER(hh, driver->grps, sg, tmp) { - driver->adapter.plugin_module->intf_funs->driver.del_group( - driver->adapter.plugin, &sg->grp); + driver->adapter.plugin_info.module->intf_funs->driver.del_group( + driver->adapter.plugin_info.plugin, &sg->grp); HASH_DEL(driver->grps, sg); neu_adapter_del_timer((neu_adapter_t *) driver, sg->report); neu_event_del_timer(driver->driver_events, sg->read); @@ -221,8 +221,8 @@ void neu_adapter_driver_process_msg(neu_adapter_driver_t *driver, int_val->key, (void *) &error); driver->adapter.cb_funs.response(&driver->adapter, &resp); } else { - driver->adapter.plugin_module->intf_funs->driver.write_tag( - driver->adapter.plugin, req->req_id, tag, value); + driver->adapter.plugin_info.module->intf_funs->driver.write_tag( + driver->adapter.plugin_info.plugin, req->req_id, tag, value); } break; @@ -245,8 +245,8 @@ void neu_adapter_driver_process_msg(neu_adapter_driver_t *driver, sg->name = strdup(neu_taggrp_cfg_get_name(sg->group)); sg->grp.group_name = strdup(neu_taggrp_cfg_get_name(sg->group)); load_tags(driver, sg->group, &sg->grp.tags); - driver->adapter.plugin_module->intf_funs->driver.add_group( - driver->adapter.plugin, &sg->grp); + driver->adapter.plugin_info.module->intf_funs->driver.add_group( + driver->adapter.plugin_info.plugin, &sg->grp); pthread_mutex_lock(&driver->grp_mtx); HASH_ADD_STR(driver->grps, name, sg); @@ -268,8 +268,8 @@ void neu_adapter_driver_process_msg(neu_adapter_driver_t *driver, HASH_FIND_STR(driver->grps, neu_taggrp_cfg_get_name(unsub_req->grp_config), sg); if (sg != NULL) { - driver->adapter.plugin_module->intf_funs->driver.del_group( - driver->adapter.plugin, &sg->grp); + driver->adapter.plugin_info.module->intf_funs->driver.del_group( + driver->adapter.plugin_info.plugin, &sg->grp); HASH_DEL(driver->grps, sg); neu_adapter_del_timer((neu_adapter_t *) driver, sg->report); neu_event_del_timer(driver->driver_events, sg->read); @@ -315,22 +315,22 @@ static int read_callback(void *usr_data) pthread_mutex_lock(&sg->driver->grp_mtx); if (neu_taggrp_cfg_is_anchored(sg->group)) { - sg->driver->adapter.plugin_module->intf_funs->driver.group_timer( - sg->driver->adapter.plugin, &sg->grp); + sg->driver->adapter.plugin_info.module->intf_funs->driver.group_timer( + sg->driver->adapter.plugin_info.plugin, &sg->grp); } else { neu_taggrp_config_t *new_config = neu_system_find_group_config( - sg->driver->adapter.plugin, sg->driver->adapter.node_id, + sg->driver->adapter.plugin_info.plugin, sg->driver->adapter.id, neu_taggrp_cfg_get_name(sg->group)); assert(new_config != NULL); - sg->driver->adapter.plugin_module->intf_funs->driver.del_group( - sg->driver->adapter.plugin, &sg->grp); + sg->driver->adapter.plugin_info.module->intf_funs->driver.del_group( + sg->driver->adapter.plugin_info.plugin, &sg->grp); neu_plugin_group_t grp = { .group_name = sg->grp.group_name }; load_tags(sg->driver, new_config, &grp.tags); - sg->driver->adapter.plugin_module->intf_funs->driver.add_group( - sg->driver->adapter.plugin, &grp); + sg->driver->adapter.plugin_info.module->intf_funs->driver.add_group( + sg->driver->adapter.plugin_info.plugin, &grp); sg->group = new_config; free_tags(sg->grp.tags); diff --git a/src/base/neu_plugin_common.c b/src/base/neu_plugin_common.c index 412ebdd2..c9c2d45a 100644 --- a/src/base/neu_plugin_common.c +++ b/src/base/neu_plugin_common.c @@ -150,21 +150,6 @@ intptr_t neu_system_del_node(neu_plugin_t *plugin, neu_node_id_t node_id) return errorcode; } -intptr_t neu_system_update_node(neu_plugin_t *plugin, neu_node_id_t node_id, - const char *node_name) -{ - intptr_t errorcode = NEU_ERR_SUCCESS; - neu_cmd_update_node_t node_update_cmd = { 0 }; - - node_update_cmd.node_name = node_name; - node_update_cmd.node_id = node_id; - - PLUGIN_CALL_CMD(plugin, NEU_REQRESP_UPDATE_NODE, node_update_cmd, intptr_t, - { errorcode = (intptr_t) resp; }) - - return errorcode; -} - vector_t neu_system_get_nodes(neu_plugin_t *plugin, neu_node_type_e node_type) { vector_t nodes = { 0 }; diff --git a/src/core/message.h b/src/core/message.h index 70b26717..73d8d0f9 100644 --- a/src/core/message.h +++ b/src/core/message.h @@ -101,7 +101,7 @@ static_assert(MSG_TYPE_END < MSG_DATABUF_KIND_MASK, /* MSG_CMD_READ_DATA */ typedef struct read_data_cmd { neu_taggrp_config_t *grp_config; - adapter_id_t sender_id; // adapter_id of sender + neu_adapter_id_t sender_id; // adapter_id of sender neu_node_id_t dst_node_id; uint32_t req_id; } read_data_cmd_t; @@ -109,7 +109,7 @@ typedef struct read_data_cmd { /* MSG_CMD_WRITE_DATA */ typedef struct write_data_cmd { neu_taggrp_config_t *grp_config; - adapter_id_t sender_id; // adapter_id of sender + neu_adapter_id_t sender_id; // adapter_id of sender neu_node_id_t dst_node_id; neu_trans_buf_t trans_buf; uint32_t req_id; @@ -118,21 +118,21 @@ typedef struct write_data_cmd { /* MSG_CMD_START_PERIODIC_READ */ typedef struct start_periodic_read_cmd { neu_taggrp_config_t *grp_config; - adapter_id_t sender_id; // adapter_id of sender + neu_adapter_id_t sender_id; // adapter_id of sender neu_node_id_t dst_node_id; } start_periodic_read_cmd_t; /* MSG_CMD_STOP_PERIODIC_READ */ typedef struct stop_periodic_read_cmd { neu_taggrp_config_t *grp_config; - adapter_id_t sender_id; // adapter_id of sender + neu_adapter_id_t sender_id; // adapter_id of sender neu_node_id_t dst_node_id; } stop_periodic_read_cmd_t; /* MSG_CMD_SUBSCRIBE_DRIVER */ typedef struct subscribe_driver_cmd { neu_taggrp_config_t *grp_config; - adapter_id_t sender_id; // adapter_id of sender + neu_adapter_id_t sender_id; // adapter_id of sender neu_node_id_t src_node_id; neu_node_id_t dst_node_id; uint32_t req_id; @@ -141,7 +141,7 @@ typedef struct subscribe_driver_cmd { /* MSG_CMD_UNSUBSCRIBE_DRIVER */ typedef struct unsubscribe_driver_cmd { neu_taggrp_config_t *grp_config; - adapter_id_t sender_id; // adapter_id of sender + neu_adapter_id_t sender_id; // adapter_id of sender neu_node_id_t src_node_id; neu_node_id_t dst_node_id; uint32_t req_id; @@ -150,21 +150,21 @@ typedef struct unsubscribe_driver_cmd { /* MSG_CMD_GROUP_CONFIG_CHANGED */ typedef struct grp_config_changed_cmd { neu_taggrp_config_t *grp_config; - adapter_id_t sender_id; // adapter_id of sender + neu_adapter_id_t sender_id; // adapter_id of sender neu_node_id_t dst_node_id; } grp_config_changed_cmd_t; /* MSG_EVENT_GROUP_CONFIG_CHANGED */ typedef struct grp_config_changed_event { neu_taggrp_config_t *grp_config; - adapter_id_t sender_id; // adapter_id of sender + neu_adapter_id_t sender_id; // adapter_id of sender neu_node_id_t dst_node_id; } grp_config_changed_event_t; /* MSG_DATA_NEURON_TRANS_DATA */ typedef struct neuron_trans_data { neu_taggrp_config_t *grp_config; - adapter_id_t sender_id; // adapter_id of sender + neu_adapter_id_t sender_id; // adapter_id of sender char * node_name; neu_trans_buf_t trans_buf; } neuron_trans_data_t; @@ -173,8 +173,8 @@ typedef struct neuron_trans_data { typedef struct read_data_resp { neu_taggrp_config_t *grp_config; neu_trans_buf_t trans_buf; - adapter_id_t sender_id; - adapter_id_t recver_id; + neu_adapter_id_t sender_id; + neu_adapter_id_t recver_id; uint32_t req_id; } read_data_resp_t; @@ -182,8 +182,8 @@ typedef struct read_data_resp { typedef struct write_data_resp { neu_taggrp_config_t *grp_config; neu_trans_buf_t trans_buf; - adapter_id_t sender_id; - adapter_id_t recver_id; + neu_adapter_id_t sender_id; + neu_adapter_id_t recver_id; uint32_t req_id; } write_data_resp_t; diff --git a/src/core/neu_manager.c b/src/core/neu_manager.c index 7bb2b49c..622b53ef 100644 --- a/src/core/neu_manager.c +++ b/src/core/neu_manager.c @@ -42,7 +42,7 @@ #include "subscribe.h" typedef struct adapter_reg_entity { - adapter_id_t adapter_id; + neu_adapter_id_t adapter_id; neu_adapter_t * adapter; neu_datatag_manager_t *datatag_manager; nng_cv * cv; @@ -184,7 +184,8 @@ static int uninit_bind_info(manager_bind_info_t *mng_bind_info) static bool match_id_reg_adapter(const void *key, const void *item) { - return *(adapter_id_t *) key == ((adapter_reg_entity_t *) item)->adapter_id; + return *(neu_adapter_id_t *) key == + ((adapter_reg_entity_t *) item)->adapter_id; } static bool match_name_reg_adapter(const void *key, const void *item) @@ -206,13 +207,14 @@ static bool match_pipe_reg_adapter(const void *key, const void *item) } // Return SIZE_MAX if can't find a adapter -static size_t find_reg_adapter_index_by_id(vector_t *adapters, adapter_id_t id) +static size_t find_reg_adapter_index_by_id(vector_t * adapters, + neu_adapter_id_t id) { return vector_find_index(adapters, &id, match_id_reg_adapter); } -static adapter_reg_entity_t *find_reg_adapter_by_id(vector_t * adapters, - adapter_id_t id) +static adapter_reg_entity_t *find_reg_adapter_by_id(vector_t * adapters, + neu_adapter_id_t id) { return vector_find_item(adapters, &id, match_id_reg_adapter); } @@ -250,9 +252,9 @@ static bool is_static_plugin(const char *plugin_name) return false; } -static adapter_id_t manager_new_adapter_id(neu_manager_t *manager) +static neu_adapter_id_t manager_new_adapter_id(neu_manager_t *manager) { - adapter_id_t adapter_id; + neu_adapter_id_t adapter_id; nng_mtx_lock(manager->adapters_mtx); adapter_id = manager->new_adapter_id++; @@ -276,14 +278,14 @@ static inline int manager_send_msg_to_pipe(neu_manager_t *manager, return rv; } -neu_node_id_t neu_manager_adapter_id_to_node_id(neu_manager_t *manager, - adapter_id_t adapter_id) +neu_node_id_t neu_manager_adapter_id_to_node_id(neu_manager_t * manager, + neu_adapter_id_t adapter_id) { (void) manager; return adapter_id; } -adapter_id_t neu_manager_adapter_id_from_node_id(neu_manager_t *manager, - neu_node_id_t node_id) +neu_adapter_id_t neu_manager_adapter_id_from_node_id(neu_manager_t *manager, + neu_node_id_t node_id) { (void) manager; return node_id; @@ -291,7 +293,7 @@ adapter_id_t neu_manager_adapter_id_from_node_id(neu_manager_t *manager, static void manager_bind_adapter(nng_pipe p, nng_pipe_ev ev, void *arg) { - adapter_id_t adapter_id; + neu_adapter_id_t adapter_id; neu_adapter_t * adapter; neu_manager_t * manager; adapter_reg_entity_t *reg_entity; @@ -354,7 +356,7 @@ static void manager_unbind_adapter(nng_pipe p, nng_pipe_ev ev, void *arg) static neu_err_code_e manager_reg_adapter(neu_manager_t * manager, adapter_reg_cmd_t *reg_param, neu_adapter_t ** p_adapter, - adapter_id_t * adapter_id) + neu_adapter_id_t * adapter_id) { int rv = NEU_ERR_SUCCESS; neu_adapter_t * adapter; @@ -438,7 +440,7 @@ static neu_err_code_e manager_reg_adapter(neu_manager_t * manager, return rv; } -static int manager_unreg_adapter(neu_manager_t *manager, adapter_id_t id, +static int manager_unreg_adapter(neu_manager_t *manager, neu_adapter_id_t id, bool need_erase) { size_t index; @@ -540,7 +542,7 @@ int neu_manager_uninit_adapter(neu_manager_t *manager, neu_adapter_t *adapter) nng_pipe msg_pipe = { 0 }; adapter_reg_entity_t *reg_entity; - adapter_id_t adapter_id; + neu_adapter_id_t adapter_id; adapter_id = neu_adapter_get_id(adapter); nng_mtx_lock(manager->adapters_mtx); @@ -618,9 +620,9 @@ static void unregister_all_reg_plugins(neu_manager_t *manager) // Call this function before start manager loop, so it don't need lock static void reg_and_start_static_adapters(neu_manager_t *manager) { - uint32_t i; - adapter_id_t id; - neu_adapter_t *p_adapter; + uint32_t i; + neu_adapter_id_t id; + neu_adapter_t * p_adapter; for (i = 0; i < STATIC_ADAPTER_ADD_INFO_SIZE; i++) { p_adapter = NULL; @@ -889,7 +891,7 @@ static void manager_loop(void *arg) size_t msg_size; nng_msg * out_msg; nng_pipe msg_pipe; - adapter_id_t adapter_id; + neu_adapter_id_t adapter_id; read_data_cmd_t * cmd_ptr; adapter_reg_entity_t *reg_entity; @@ -922,7 +924,7 @@ static void manager_loop(void *arg) size_t msg_size; nng_msg * out_msg; nng_pipe msg_pipe; - adapter_id_t adapter_id; + neu_adapter_id_t adapter_id; read_data_resp_t * cmd_ptr; adapter_reg_entity_t *reg_entity; @@ -955,7 +957,7 @@ static void manager_loop(void *arg) size_t msg_size; nng_msg * out_msg; nng_pipe msg_pipe; - adapter_id_t adapter_id; + neu_adapter_id_t adapter_id; write_data_cmd_t * cmd_ptr; adapter_reg_entity_t *reg_entity; @@ -988,7 +990,7 @@ static void manager_loop(void *arg) size_t msg_size; nng_msg * out_msg; nng_pipe msg_pipe; - adapter_id_t adapter_id; + neu_adapter_id_t adapter_id; write_data_resp_t * cmd_ptr; adapter_reg_entity_t *reg_entity; @@ -1128,10 +1130,10 @@ static void manager_loop(void *arg) } case MSG_EVENT_GROUP_CONFIG_CHANGED: { - size_t msg_size; - nng_msg * out_msg; - nng_pipe msg_pipe; - adapter_id_t adapter_id; + size_t msg_size; + nng_msg * out_msg; + nng_pipe msg_pipe; + neu_adapter_id_t adapter_id; grp_config_changed_event_t *event_ptr; adapter_reg_entity_t * reg_entity; @@ -1368,7 +1370,7 @@ int neu_manager_add_node(neu_manager_t *manager, neu_cmd_add_node_t *cmd, neu_node_id_t *p_node_id) { int rv = 0; - adapter_id_t adapter_id; + neu_adapter_id_t adapter_id; neu_adapter_t * adapter = NULL; adapter_reg_cmd_t reg_cmd; @@ -1391,12 +1393,13 @@ int neu_manager_update_node_id(neu_manager_t *manager, neu_node_id_t node_id, neu_node_id_t new_node_id) { int rv = 0; - adapter_id_t adapter_id; - adapter_id_t new_adapter_id; + neu_adapter_id_t adapter_id; + neu_adapter_id_t new_adapter_id; adapter_reg_entity_t *reg_entity; // declaration for neu_adapter_set_id since it is not exposed - adapter_id_t neu_adapter_set_id(neu_adapter_t * adapter, adapter_id_t id); + neu_adapter_id_t neu_adapter_set_id(neu_adapter_t * adapter, + neu_adapter_id_t id); if (0 == new_node_id) { return NEU_ERR_EINVAL; @@ -1430,7 +1433,7 @@ int neu_manager_update_node_id(neu_manager_t *manager, neu_node_id_t node_id, int neu_manager_del_node(neu_manager_t *manager, neu_node_id_t node_id) { int rv = 0; - adapter_id_t adapter_id; + neu_adapter_id_t adapter_id; neu_adapter_t * adapter = NULL; adapter_reg_entity_t *reg_entity; vector_t group_config; @@ -1471,42 +1474,15 @@ int neu_manager_del_node(neu_manager_t *manager, neu_node_id_t node_id) return rv; } -int neu_manager_update_node(neu_manager_t *manager, neu_cmd_update_node_t *cmd) -{ - int rv = NEU_ERR_NODE_NOT_EXIST; - size_t index = 0; - neu_adapter_t *adapter = NULL; - vector_t * reg_adapters = &manager->reg_adapters; - adapter_id_t adapter_id = - neu_manager_adapter_id_from_node_id(manager, cmd->node_id); - - nng_mtx_lock(manager->adapters_mtx); - index = find_reg_adapter_index_by_id(reg_adapters, adapter_id); - if (index != SIZE_MAX) { - adapter_reg_entity_t *reg_entity; - - reg_entity = (adapter_reg_entity_t *) vector_get(reg_adapters, index); - adapter = reg_entity->adapter; - - if (adapter != NULL) { - neu_adapter_rename(adapter, cmd->node_name); - rv = NEU_ERR_SUCCESS; - } - } - nng_mtx_unlock(manager->adapters_mtx); - - 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_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_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); @@ -1548,9 +1524,9 @@ int neu_manager_unsubscribe_node(neu_manager_t * manager, int rv = 0; nng_mtx_lock(manager->adapters_mtx); - adapter_id_t src_adapter_id = + neu_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_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); @@ -1623,9 +1599,9 @@ int neu_manager_get_nodes(neu_manager_t *manager, neu_node_type_e node_type, { reg_entity = (adapter_reg_entity_t *) iterator_get(&iter); if (adapter_match_node_type(reg_entity->adapter, node_type)) { - neu_node_info_t node_info; - adapter_id_t adapter_id; - const char * adapter_name; + neu_node_info_t node_info; + neu_adapter_id_t adapter_id; + const char * adapter_name; adapter_name = neu_adapter_get_name(reg_entity->adapter); adapter_id = neu_adapter_get_id(reg_entity->adapter); @@ -1646,7 +1622,7 @@ int neu_manager_get_node_name_by_id(neu_manager_t *manager, neu_node_id_t node_id, char **name) { int rv = 0; - adapter_id_t adapter_id; + neu_adapter_id_t adapter_id; adapter_reg_entity_t *reg_entity; if (NULL == name) { @@ -1696,7 +1672,7 @@ int neu_manager_add_grp_config(neu_manager_t * manager, neu_cmd_add_grp_config_t *cmd) { int rv = NEU_ERR_SUCCESS; - adapter_id_t adapter_id; + neu_adapter_id_t adapter_id; adapter_reg_entity_t *reg_entity; if (manager == NULL || cmd == NULL) { @@ -1728,7 +1704,7 @@ int neu_manager_del_grp_config(neu_manager_t *manager, neu_node_id_t node_id, const char *config_name) { int rv = NEU_ERR_SUCCESS; - adapter_id_t adapter_id; + neu_adapter_id_t adapter_id; adapter_reg_entity_t *reg_entity; if (manager == NULL || config_name == NULL) { @@ -1760,7 +1736,7 @@ int neu_manager_update_grp_config(neu_manager_t * manager, neu_cmd_update_grp_config_t *cmd) { int rv = NEU_ERR_SUCCESS; - adapter_id_t adapter_id; + neu_adapter_id_t adapter_id; adapter_reg_entity_t *reg_entity; if (manager == NULL || cmd == NULL) { @@ -1801,7 +1777,7 @@ int neu_manager_get_grp_configs(neu_manager_t *manager, neu_node_id_t node_id, { int rv = 0; adapter_reg_entity_t *reg_entity; - adapter_id_t adapter_id; + neu_adapter_id_t adapter_id; if (manager == NULL || result_grp_configs == NULL) { log_error("get nodes with NULL manager or result_nodes"); @@ -1830,7 +1806,7 @@ int neu_manager_get_persist_datatag_infos(neu_manager_t *manager, { int rv = 0; adapter_reg_entity_t *reg_entity; - adapter_id_t adapter_id; + neu_adapter_id_t adapter_id; if (manager == NULL || result == NULL) { log_error("get persist datatag infos with NULL manager or result"); @@ -2050,7 +2026,7 @@ neu_datatag_table_t *neu_manager_get_datatag_tbl(neu_manager_t *manager, { adapter_reg_entity_t *reg_entity; neu_datatag_table_t * tag_table; - adapter_id_t adapter_id; + neu_adapter_id_t adapter_id; if (manager == NULL) { log_error("get datatag table with NULL manager"); @@ -2076,7 +2052,7 @@ neu_datatag_table_t *neu_manager_get_datatag_tbl(neu_manager_t *manager, int neu_manager_adapter_set_setting(neu_manager_t *manager, neu_node_id_t node_id, const char *config) { - adapter_id_t adapter_id = { 0 }; + neu_adapter_id_t adapter_id = { 0 }; adapter_reg_entity_t *reg_entity = NULL; int ret = 0; neu_config_t neu_config = { .type = NEU_CONFIG_SETTING }; @@ -2170,7 +2146,7 @@ int neu_manager_get_persist_adapter_infos(neu_manager_t *manager, int neu_manager_adapter_get_setting(neu_manager_t *manager, neu_node_id_t node_id, char **config) { - adapter_id_t adapter_id = { 0 }; + neu_adapter_id_t adapter_id = { 0 }; adapter_reg_entity_t *reg_entity = NULL; int ret = 0; @@ -2194,7 +2170,7 @@ int neu_manager_adapter_get_setting(neu_manager_t *manager, int neu_manager_start_adapter_with_id(neu_manager_t *manager, neu_node_id_t node_id) { - adapter_id_t adapter_id = { 0 }; + neu_adapter_id_t adapter_id = { 0 }; adapter_reg_entity_t *reg_entity = NULL; nng_mtx_lock(manager->adapters_mtx); @@ -2219,7 +2195,7 @@ int neu_manager_start_adapter_with_id(neu_manager_t *manager, int neu_manager_adapter_get_state(neu_manager_t *manager, neu_node_id_t node_id, neu_plugin_state_t *state) { - adapter_id_t adapter_id = { 0 }; + neu_adapter_id_t adapter_id = { 0 }; adapter_reg_entity_t *reg_entity = NULL; nng_mtx_lock(manager->adapters_mtx); @@ -2242,7 +2218,7 @@ int neu_manager_adapter_get_state(neu_manager_t *manager, neu_node_id_t node_id, int neu_manager_adapter_ctl(neu_manager_t *manager, neu_node_id_t node_id, neu_adapter_ctl_e ctl) { - adapter_id_t adapter_id = { 0 }; + neu_adapter_id_t adapter_id = { 0 }; adapter_reg_entity_t *reg_entity = NULL; neu_adapter_t * adapter; int ret = NEU_ERR_SUCCESS; @@ -2277,7 +2253,7 @@ int neu_manager_adapter_get_sub_grp_configs(neu_manager_t *manager, neu_node_id_t node_id, vector_t ** result_sgc) { - adapter_id_t adapter_id = { 0 }; + neu_adapter_id_t adapter_id = { 0 }; adapter_reg_entity_t *reg_entity = NULL; nng_mtx_lock(manager->adapters_mtx); @@ -2310,7 +2286,7 @@ int neu_manager_adapter_get_grp_config_ref_by_name( } nng_mtx_lock(manager->adapters_mtx); - adapter_id_t adapter_id = + neu_adapter_id_t adapter_id = neu_manager_adapter_id_from_node_id(manager, node_id); adapter_reg_entity_t *reg_entity = find_reg_adapter_by_id(&manager->reg_adapters, adapter_id); @@ -2344,7 +2320,7 @@ int neu_manager_get_persist_subscription_infos(neu_manager_t *manager, return NEU_ERR_EINTERNAL; } - adapter_id_t adapter_id = { 0 }; + neu_adapter_id_t adapter_id = { 0 }; adapter_reg_entity_t *reg_entity = NULL; nng_mtx_lock(manager->adapters_mtx); @@ -2374,7 +2350,7 @@ int neu_manager_get_persist_subscription_infos(neu_manager_t *manager, VECTOR_FOR_EACH(sub_grp_configs, iter) { neu_sub_grp_config_t *sgc = iterator_get(&iter); - adapter_id_t adapter_id = + neu_adapter_id_t adapter_id = neu_manager_adapter_id_from_node_id(manager, sgc->node_id); adapter_reg_entity_t *src_reg_entity = find_reg_adapter_by_id(&manager->reg_adapters, adapter_id); @@ -2427,7 +2403,7 @@ final: int neu_manager_adapter_validate_tag(neu_manager_t *manager, neu_node_id_t node_id, neu_datatag_t *tags) { - adapter_id_t adapter_id = { 0 }; + neu_adapter_id_t adapter_id = { 0 }; adapter_reg_entity_t *reg_entity = NULL; neu_adapter_t * adapter; int ret = -1; diff --git a/src/core/neu_manager.h b/src/core/neu_manager.h index cf0af9d6..c775f575 100644 --- a/src/core/neu_manager.h +++ b/src/core/neu_manager.h @@ -40,7 +40,6 @@ void neu_manager_trigger_running(neu_manager_t *manager); int neu_manager_add_node(neu_manager_t *manager, neu_cmd_add_node_t *cmd, neu_node_id_t *p_node_id); int neu_manager_del_node(neu_manager_t *manager, neu_node_id_t node_id); -int neu_manager_update_node(neu_manager_t *manager, neu_cmd_update_node_t *cmd); 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, @@ -81,9 +80,9 @@ int neu_manager_get_persist_plugin_infos(neu_manager_t *manager, neu_datatag_table_t *neu_manager_get_datatag_tbl(neu_manager_t *manager, neu_node_id_t node_id); -neu_node_id_t neu_manager_adapter_id_to_node_id(neu_manager_t *manager, - adapter_id_t adapter_id); -adapter_id_t neu_manager_adapter_id_from_node_id(neu_manager_t *manager, +neu_node_id_t neu_manager_adapter_id_to_node_id(neu_manager_t * manager, + neu_adapter_id_t adapter_id); +neu_adapter_id_t neu_manager_adapter_id_from_node_id(neu_manager_t *manager, neu_node_id_t node_id); int neu_manager_adapter_set_setting(neu_manager_t *manager, neu_node_id_t node_id, const char *config);