GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-22 09:23:52 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #430 from fengzeroz/main

Browse Source 
refactor adapter
This commit is contained in:
fengzeroz
2022-04-15 17:47:56 +08:00
committed by GitHub
parent e98c91a677 1b1984bcb0
commit c06280bc2d
19 changed files with 252 additions and 436 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ft/api_http.resource
-7
View File
@@ -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}}
ft/api_mqtt.resource
-6
View File
@@ -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}
ft/node_api_test.robot
-20
View File
@@ -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}
include/neuron/adapter_msg.h
-7
View File
@@ -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;
include/neuron/plugin.h
-2
View File
@@ -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,
plugins/mqtt/command/command.c
-15
View File
@@ -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);
}
plugins/mqtt/command/node.c
-13
View File
@@ -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)
{
plugins/mqtt/command/node.h
-2
View File
@@ -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,
plugins/restful/adapter_handle.c
-11
View File
@@ -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();
plugins/restful/adapter_handle.h
-1
View File
@@ -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);
plugins/restful/handle.c
-6
View File
@@ -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,
src/adapter/adapter.c
+123 -179
View File
@@ -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;
}
src/adapter/adapter_info.h
+7 -7
View File
@@ -22,6 +22,7 @@
#include <stdint.h>
#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
src/adapter/adapter_internal.h
+32 -30
View File
@@ -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);
src/adapter/driver/driver.c
+16 -16
View File
@@ -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);
src/base/neu_plugin_common.c
-15
View File
@@ -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 };
src/core/message.h
+13 -13
View File
@@ -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;
src/core/neu_manager.c
+58 -82
View File
@@ -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;
src/core/neu_manager.h
+3 -4
View File
@@ -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);