Make default persist adapter load persistence data at startup

This commit is contained in:
eeff
2022-01-13 18:04:20 +08:00
committed by Parker Liu
parent 3cf9030283
commit 790f0ebb0d
3 changed files with 68 additions and 3 deletions
+52 -1
View File
@@ -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
+1
View File
@@ -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,
+15 -2
View File
@@ -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, &reg_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;
}