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

Fix adapter to handle node update events correctly

Browse Source
This commit is contained in:
eeff
2022-03-23 10:06:33 +08:00
parent c911216d55
commit de5fcb22fb
2 changed files with 66 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neu_adapter.h
+5
View File
@@ -383,6 +383,11 @@ typedef struct neu_event_tags {
char * grp_config_name;
} neu_event_tags_t;
typedef struct {
char *src_name;
char *dst_name;
} neu_event_update_node_t;
/**
* definition enum and structure for neuron config
*/
src/adapter/adapter.c
+61 -8
View File
@@ -551,6 +551,37 @@ static int persister_singleton_handle_nodes(neu_adapter_t *adapter,
return rv;
}
static int
persister_singleton_handle_update_node(neu_adapter_t * adapter,
neu_event_update_node_t *event)
{
neu_persister_t *persister = persister_singleton_get();
vector_t * adapter_infos = NULL;
log_info("%s handling update node event of %s", adapter->name,
event->src_name);
int rv =
neu_manager_get_persist_adapter_infos(adapter->manager, &adapter_infos);
if (0 != rv) {
log_error("%s unable to get adapter infos", adapter->name);
return rv;
}
// store the current set of adapter infos
rv = neu_persister_store_adapters(persister, adapter_infos);
if (0 == rv) {
rv = neu_persister_update_adapter(persister, event->src_name,
event->dst_name);
} else {
log_error("%s failed to store adapter infos", adapter->name);
}
neu_persist_adapter_infos_free(adapter_infos);
return rv;
}
static int persister_singleton_handle_plugins(neu_adapter_t *adapter,
msg_type_e event)
{
@@ -829,15 +860,20 @@ static void adapter_loop(void *arg)
case MSG_EVENT_ADD_NODE:
// fall through
case MSG_EVENT_UPDATE_NODE:
// fall through
case MSG_EVENT_DEL_NODE: {
const char *node_name = msg_get_buf_ptr(pay_msg);
persister_singleton_handle_nodes(adapter, pay_msg_type, node_name);
break;
}
case MSG_EVENT_UPDATE_NODE: {
neu_event_update_node_t *event = msg_get_buf_ptr(pay_msg);
persister_singleton_handle_update_node(adapter, event);
free(event->src_name);
free(event->dst_name);
break;
}
case MSG_EVENT_SET_NODE_SETTING: {
neu_cmd_set_node_setting_t *cmd = msg_get_buf_ptr(pay_msg);
neu_persister_t * persister = persister_singleton_get();
@@ -1212,10 +1248,27 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd,
ADAPTER_RESP_CODE(
adapter, cmd, intptr_t, neu_cmd_update_node_t, rv,
NEU_REQRESP_ERR_CODE, p_result, {
ret = neu_manager_update_node(adapter->manager, req_cmd);
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) {
ADAPTER_SEND_NODE_EVENT(adapter, rv, MSG_EVENT_UPDATE_NODE,
req_cmd->node_name);
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;
@@ -1510,8 +1563,8 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd,
rv = neu_manager_get_node_name_by_id(
adapter->manager, req_cmd->node_id, &node_name);
if (0 == rv) {
ADAPTER_SEND_NODE_EVENT(
adapter, rv, MSG_EVENT_UPDATE_NODE, node_name);
ADAPTER_SEND_NODE_EVENT(adapter, rv, MSG_EVENT_ADD_NODE,
node_name);
free(node_name);
}
}