diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 3ef1b60e..1e5adfcd 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -186,6 +186,52 @@ static neu_persister_t *persister_singleton_get() return g_persister_singleton; } +static int persister_singleton_load_data(neu_adapter_t *adapter) +{ + vector_t * adapter_infos = NULL; + neu_persister_t *persister = persister_singleton_get(); + + log_info("%s start persistence loading", adapter->name); + + // TODO: load plugin infos + + int rv = neu_persister_load_adapters(persister, &adapter_infos); + if (0 != rv) { + log_error("%s failed to load adapter infos", adapter->name); + goto error_load_adapters; + } + + VECTOR_FOR_EACH(adapter_infos, iter) + { + neu_persist_adapter_info_t *adapter_info = iterator_get(&iter); + neu_cmd_add_node_t add_node_cmd = { + .node_type = adapter_info->type, + .adapter_name = adapter_info->name, + .plugin_name = adapter_info->plugin_name, + }; + + rv = neu_manager_add_node(adapter->manager, &add_node_cmd, NULL); + const char *ok_or_err = (0 == rv) ? "success" : "fail"; + log_info("%s load adapter %s type:%d, name:%s plugin:%s", adapter->name, + ok_or_err, adapter_info->type, adapter_info->name, + adapter_info->plugin_name); + if (0 != rv) { + goto error_add_adapters; + } + // TODO: node setting + // TODO: start node according to state + } + + neu_persist_adapter_infos_free(adapter_infos); + + return rv; + +error_add_adapters: + neu_persist_adapter_infos_free(adapter_infos); +error_load_adapters: + return rv; +} + static int persister_singleton_handle_nodes(neu_adapter_t *adapter, msg_type_e event, const char * node_name) @@ -279,7 +325,7 @@ static void adapter_loop(void *arg) neu_panic("The adapter can't dial to %s", manager_url); } - const char *adapter_str = "adapter started"; + const char *adapter_str = adapter->name; nng_msg * out_msg; size_t msg_size; msg_size = msg_inplace_data_get_size(strlen(adapter_str) + 1); @@ -356,6 +402,11 @@ static void adapter_loop(void *arg) break; } + case MSG_CMD_PERSISTENCE_LOAD: { + persister_singleton_load_data(adapter); + break; + } + case MSG_EVENT_ADD_NODE: // fall through diff --git a/src/core/message.h b/src/core/message.h index c034a0ec..f0d63da4 100644 --- a/src/core/message.h +++ b/src/core/message.h @@ -39,6 +39,7 @@ typedef enum msg_type { MSG_CMD_SUBSCRIBE_NODE, MSG_CMD_UNSUBSCRIBE_NODE, MSG_CMD_GROUP_CONFIG_CHANGED, + MSG_CMD_PERSISTENCE_LOAD, MSG_CMD_EXIT_LOOP, MSG_CMD_RESP_PONG, // response pong for ping MSG_TYPE_CMD_END, diff --git a/src/core/neu_manager.c b/src/core/neu_manager.c index 0aabb3bf..2a6b1bc3 100644 --- a/src/core/neu_manager.c +++ b/src/core/neu_manager.c @@ -1149,6 +1149,17 @@ static void manager_loop(void *arg) log_info("Reply pong to pipe: %d", msg_pipe); nng_sendmsg(manager_bind->mng_sock, out_msg, 0); } + if (0 == + strncmp(DEFAULT_PERSIST_ADAPTER_NAME, buf_ptr, + msg_get_buf_len(pay_msg))) { + msg_size = msg_inplace_data_get_size(0); + rv = nng_msg_alloc(&out_msg, msg_size); + message_t *msg_ptr = (message_t *) nng_msg_body(out_msg); + msg_inplace_data_init(msg_ptr, MSG_CMD_PERSISTENCE_LOAD, 0); + nng_msg_set_pipe(out_msg, msg_pipe); + log_info("Send persistence load to pipe: %d", msg_pipe); + nng_sendmsg(manager_bind->mng_sock, out_msg, 0); + } break; } @@ -1669,8 +1680,10 @@ int neu_manager_add_node(neu_manager_t *manager, neu_cmd_add_node_t *cmd, reg_cmd.plugin_id = cmd->plugin_id; rv = manager_reg_adapter(manager, ®_cmd, &adapter, &adapter_id); if (rv == NEU_ERR_SUCCESS) { - *p_node_id = neu_manager_adapter_id_to_node_id(manager, adapter_id); - rv = neu_manager_init_adapter(manager, adapter); + if (NULL != p_node_id) { + *p_node_id = neu_manager_adapter_id_to_node_id(manager, adapter_id); + } + rv = neu_manager_init_adapter(manager, adapter); } return rv; }