GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-28 14:07:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

Update neu_manager to forward node event to persist adapter

Browse Source 
1. Forward add/update/del node event in neu_manager loop
  2. Handle add/update/del node event in persist adapter loop
This commit is contained in:
eeff
2022-01-13 18:04:20 +08:00
committed by Parker Liu
parent 434ce97f2b
commit 3cf9030283
4 changed files with 198 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapter/adapter.c
+80 -6
View File
@@ -21,6 +21,7 @@
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <threads.h>
#include <unistd.h>
#include <nng/nng.h>
@@ -28,6 +29,7 @@
#include <nng/supplemental/util/platform.h>
#include "adapter_internal.h"
#include "config.h"
#include "core/message.h"
#include "core/neu_manager.h"
#include "core/neu_trans_buf.h"
@@ -37,6 +39,7 @@
#include "neu_panic.h"
#include "neu_plugin.h"
#include "neu_subscribe.h"
#include "persist/persist.h"
#define to_node_id(adapter, id) \
neu_manager_adapter_id_to_node_id((adapter)->manager, id);
@@ -156,6 +159,66 @@ struct neu_adapter {
vector_t sub_grp_configs; // neu_sub_grp_config_t
};
static once_flag g_persister_init_flag = ONCE_FLAG_INIT;
static neu_persister_t *g_persister_singleton = NULL;
static void persister_singleton_init()
{
const char *persistence_dir = neu_config_get_value(
(char *) "./neuron.yaml", 2, (char *) "persistence", (char *) "dir");
if (NULL == persistence_dir) {
log_error("can't get persistence dir from config");
exit(EXIT_FAILURE);
}
neu_persister_t *persister = neu_persister_create(persistence_dir);
if (NULL == persister) {
log_error("can't create persister");
exit(EXIT_FAILURE);
}
free((char *) persistence_dir);
g_persister_singleton = persister;
}
static neu_persister_t *persister_singleton_get()
{
call_once(&g_persister_init_flag, persister_singleton_init);
return g_persister_singleton;
}
static int persister_singleton_handle_nodes(neu_adapter_t *adapter,
msg_type_e event,
const char * node_name)
{
neu_persister_t *persister = persister_singleton_get();
vector_t * adapter_infos = NULL;
log_info("%s handling node event %d of %s", adapter->name, event,
node_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) {
log_error("%s failed to store adapter infos", adapter->name);
} else if (MSG_EVENT_DEL_NODE == event) {
rv = neu_persister_delete_adapter(persister, node_name);
if (0 != rv) {
log_error("%s failed to del adapter %s", adapter->name, node_name);
}
}
neu_persist_adapter_infos_free(adapter_infos);
return rv;
}
neu_plugin_running_state_e
neu_adapter_state_to_plugin_state(neu_adapter_t *adapter)
{
@@ -206,10 +269,9 @@ static uint32_t adapter_get_req_id(neu_adapter_t *adapter)
static void adapter_loop(void *arg)
{
int rv;
neu_adapter_t *adapter;
nng_dialer dialer = { 0 };
nng_dialer dialer = { 0 };
neu_adapter_t *adapter = (neu_adapter_t *) arg;
adapter = (neu_adapter_t *) arg;
const char *manager_url;
manager_url = neu_manager_get_url(adapter->manager);
rv = nng_dial(adapter->sock, manager_url, &dialer, 0);
@@ -251,9 +313,9 @@ static void adapter_loop(void *arg)
continue;
}
message_t *pay_msg;
pay_msg = nng_msg_body(msg);
switch (msg_get_type(pay_msg)) {
message_t *pay_msg = nng_msg_body(msg);
msg_type_e pay_msg_type = msg_get_type(pay_msg);
switch (pay_msg_type) {
case MSG_CMD_RESP_PONG: {
char *buf_ptr;
@@ -294,6 +356,18 @@ static void adapter_loop(void *arg)
break;
}
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_CMD_SUBSCRIBE_NODE: {
subscribe_node_cmd_t *cmd_ptr;
cmd_ptr = (subscribe_node_cmd_t *) msg_get_buf_ptr(pay_msg);
src/core/neu_manager.c
+105 -1
View File
@@ -28,15 +28,16 @@
#include <nng/protocol/pair1/pair.h>
#include <nng/supplemental/util/platform.h>
#include "adapter/adapter_internal.h"
#include "message.h"
#include "neu_adapter.h"
#include "neu_datatag_manager.h"
#include "neu_errcodes.h"
#include "neu_log.h"
#include "neu_manager.h"
#include "neu_panic.h"
#include "neu_trans_buf.h"
#include "neu_vector.h"
#include "persist/persist.h"
#include "plugin_manager.h"
typedef struct adapter_reg_entity {
@@ -387,6 +388,17 @@ static adapter_reg_entity_t *find_reg_adapter_by_pipe(vector_t *adapters,
return vector_find_item(adapters, &p, match_pipe_reg_adapter);
}
static bool is_default_adapter(const char *adapter_name)
{
for (size_t i = 0; i < DEFAULT_ADAPTER_ADD_INFO_SIZE; ++i) {
if (0 ==
strcmp(default_adapter_reg_cmds[i].adapter_name, adapter_name)) {
return true;
}
}
return false;
}
static adapter_id_t manager_new_adapter_id(neu_manager_t *manager)
{
adapter_id_t adapter_id;
@@ -1284,6 +1296,36 @@ static void manager_loop(void *arg)
break;
}
case MSG_EVENT_ADD_NODE:
// fall through
case MSG_EVENT_UPDATE_NODE:
// fall through
case MSG_EVENT_DEL_NODE: {
adapter_reg_entity_t *persist_reg_entity = find_reg_adapter_by_name(
&manager->reg_adapters, DEFAULT_PERSIST_ADAPTER_NAME);
nng_pipe persist_pipe = persist_reg_entity->adapter_pipe;
msg_type_e msg_type = msg_get_type(pay_msg);
nng_msg * out_msg;
const char *name = msg_get_buf_ptr(pay_msg);
size_t len = msg_get_buf_len(pay_msg);
size_t msg_size = msg_inplace_data_get_size(len);
rv = nng_msg_alloc(&out_msg, msg_size);
if (rv == 0) {
message_t *msg_ptr = nng_msg_body(out_msg);
msg_inplace_data_init(msg_ptr, msg_type, len);
char *out_buf = msg_get_buf_ptr(msg_ptr);
memcpy(out_buf, name, len);
nng_msg_set_pipe(out_msg, persist_pipe);
log_info("Forward node event %d to %s pipe: %d", msg_type,
DEFAULT_PERSIST_ADAPTER_NAME, persist_pipe);
nng_sendmsg(manager_bind->mng_sock, out_msg, 0);
}
break;
}
case MSG_CMD_SUBSCRIBE_NODE: {
size_t msg_size;
nng_msg * out_msg;
@@ -2028,6 +2070,68 @@ int neu_manager_adapter_set_setting(neu_manager_t *manager,
return ret;
}
const char *find_adapter_plugin_name(neu_manager_t *manager,
neu_adapter_t *adapter)
{
plugin_id_t plugin_id = neu_adapter_get_plugin_id(adapter);
plugin_reg_info_t reg_info = {};
plugin_manager_get_reg_info(manager->plugin_manager, plugin_id, &reg_info);
return reg_info.plugin_name;
}
int neu_manager_get_persist_adapter_infos(neu_manager_t *manager,
vector_t ** result)
{
adapter_reg_entity_t *reg_entity = NULL;
vector_t * adapter_infos = NULL;
if (NULL == manager || NULL == result) {
log_error("get persist adapter infos with NULL manager or NULL result");
return NEU_ERR_EINVAL;
}
size_t count = manager->reg_adapters.size;
if (count >= DEFAULT_ADAPTER_ADD_INFO_SIZE) {
count -= DEFAULT_ADAPTER_ADD_INFO_SIZE;
}
adapter_infos = vector_new(count, sizeof(neu_persist_adapter_info_t));
if (NULL == adapter_infos) {
return NEU_ERR_ENOMEM;
}
nng_mtx_lock(manager->adapters_mtx);
VECTOR_FOR_EACH(&manager->reg_adapters, iter)
{
reg_entity = (adapter_reg_entity_t *) iterator_get(&iter);
const char *adapter_name = neu_adapter_get_name(reg_entity->adapter);
const char *plugin_name =
find_adapter_plugin_name(manager, reg_entity->adapter);
if (is_default_adapter(adapter_name)) {
continue;
}
neu_persist_adapter_info_t adapter_info = {};
adapter_info.type = neu_adapter_get_type(reg_entity->adapter);
adapter_info.state = neu_adapter_get_state(reg_entity->adapter).running;
adapter_info.name = strdup(adapter_name);
adapter_info.plugin_name = strdup(plugin_name);
if (0 != vector_push_back(adapter_infos, &adapter_info)) {
neu_persist_adapter_infos_free(adapter_infos);
nng_mtx_unlock(manager->adapters_mtx);
return NEU_ERR_ENOMEM;
}
}
nng_mtx_unlock(manager->adapters_mtx);
*result = adapter_infos;
return NEU_ERR_SUCCESS;
}
int neu_manager_adapter_get_setting(neu_manager_t *manager,
neu_node_id_t node_id, char **config)
{
src/core/neu_manager.h
+2
View File
@@ -78,6 +78,8 @@ int neu_manager_adapter_get_state(neu_manager_t *manager, neu_node_id_t node_id,
neu_plugin_state_t *state);
int neu_manager_adapter_ctl(neu_manager_t *manager, neu_node_id_t node_id,
neu_adapter_ctl_e ctl);
int neu_manager_get_persist_adapter_infos(neu_manager_t *manager,
vector_t ** result);
int neu_manager_start_adapter(neu_adapter_t *adapter);
int neu_manager_stop_adapter(neu_adapter_t *adapter);
int neu_manager_init_adapter(neu_manager_t *manager, neu_adapter_t *adapter);
src/persist/persist.h
+11
View File
@@ -39,6 +39,17 @@ typedef neu_json_group_configs_req_t neu_persist_group_config_info_t;
typedef neu_json_subscriptions_req_subscription_t
neu_persist_subscription_info_t;
static inline void neu_persist_adapter_infos_free(vector_t *adapter_infos)
{
VECTOR_FOR_EACH(adapter_infos, iter)
{
neu_persist_adapter_info_t *p = iterator_get(&iter);
free(p->name);
free(p->plugin_name);
}
vector_free(adapter_infos);
}
/**
* Persister, provide methods to persist data */
typedef struct neu_persister neu_persister_t;