diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 628a8dba..3ef1b60e 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -21,6 +21,7 @@ #include #include #include +#include #include #include @@ -28,6 +29,7 @@ #include #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); diff --git a/src/core/neu_manager.c b/src/core/neu_manager.c index ddbb5a9f..0aabb3bf 100644 --- a/src/core/neu_manager.c +++ b/src/core/neu_manager.c @@ -28,15 +28,16 @@ #include #include -#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, ®_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) { diff --git a/src/core/neu_manager.h b/src/core/neu_manager.h index 13589619..b6bdd802 100644 --- a/src/core/neu_manager.h +++ b/src/core/neu_manager.h @@ -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); diff --git a/src/persist/persist.h b/src/persist/persist.h index bd1c7975..5b0dc059 100644 --- a/src/persist/persist.h +++ b/src/persist/persist.h @@ -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;