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

node ctl/state

Browse Source
This commit is contained in:
fengzero
2022-06-10 04:43:05 +00:00
parent 27375c5d6c
commit 51ebf90294
9 changed files with 143 additions and 121 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/errcodes.h
+6 -5
View File
@@ -101,11 +101,12 @@ typedef enum {
NEU_ERR_TAG_TYPE_NOT_SUPPORT = 2204,
NEU_ERR_TAG_ADDRESS_FORMAT_INVALID = 2205,
NEU_ERR_LIBRARY_NOT_FOUND = 2301,
NEU_ERR_LIBRARY_INFO_INVALID = 2302,
NEU_ERR_LIBRARY_NAME_CONFLICT = 2303,
NEU_ERR_LIBRARY_FAILED_TO_OPEN = 2304,
NEU_ERR_LIBRARY_MODULE_INVALID = 2305,
NEU_ERR_LIBRARY_NOT_FOUND = 2301,
NEU_ERR_LIBRARY_INFO_INVALID = 2302,
NEU_ERR_LIBRARY_NAME_CONFLICT = 2303,
NEU_ERR_LIBRARY_FAILED_TO_OPEN = 2304,
NEU_ERR_LIBRARY_MODULE_INVALID = 2305,
NEU_ERR_LIBRARY_SYSTEM_NOT_ALLOW_DEL = 2306,
NEU_ERR_LICENSE_NOT_FOUND = 2400,
NEU_ERR_LICENSE_INVALID = 2401,
include/neuron/json/neu_json_node.h
+1 -1
View File
@@ -82,8 +82,8 @@ int neu_json_decode_update_node_req(char * buf,
void neu_json_decode_update_node_req_free(neu_json_update_node_req_t *req);
typedef struct {
char * name;
int64_t cmd;
int64_t id;
} neu_json_node_ctl_req_t;
int neu_json_decode_node_ctl_req(char *buf, neu_json_node_ctl_req_t **result);
plugins/restful/adapter_handle.c
+33 -34
View File
@@ -144,48 +144,47 @@ void handle_get_node_setting(nng_aio *aio)
void handle_node_ctl(nng_aio *aio)
{
(void) aio;
// neu_plugin_t *plugin = neu_rest_get_plugin();
neu_plugin_t *plugin = neu_rest_get_plugin();
// REST_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
// aio, neu_json_node_ctl_req_t, neu_json_decode_node_ctl_req, {
// NEU_JSON_RESPONSE_ERROR(
// neu_plugin_node_ctl(plugin, req->id, req->cmd),
//{ http_response(aio, error_code.error, result_error); });
//})
REST_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
aio, neu_json_node_ctl_req_t, neu_json_decode_node_ctl_req, {
NEU_JSON_RESPONSE_ERROR(
neu_plugin_node_ctl(plugin, req->name, req->cmd),
{ http_response(aio, error_code.error, result_error); });
})
}
void handle_get_node_state(nng_aio *aio)
{
(void) aio;
// neu_plugin_t * plugin = neu_rest_get_plugin();
// neu_node_id_t node_id = 0;
// neu_json_get_node_state_resp_t res = { 0 };
// neu_plugin_state_t state = { 0 };
// char * result = NULL;
neu_plugin_t * plugin = neu_rest_get_plugin();
neu_json_get_node_state_resp_t res = { 0 };
neu_plugin_state_t state = { 0 };
char node_name[NEU_NODE_NAME_LEN] = { 0 };
char * result = NULL;
// VALIDATE_JWT(aio);
VALIDATE_JWT(aio);
// if (http_get_param_node_id(aio, "node_id", &node_id) != 0) {
// NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
// http_response(aio, error_code.error, result_error);
//})
// return;
//}
if (http_get_param_str(aio, "node_name", node_name, sizeof(node_name)) <=
0) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
http_response(aio, error_code.error, result_error);
})
return;
}
// NEU_JSON_RESPONSE_ERROR(
// neu_plugin_get_node_state(plugin, node_id, &state), {
// if (error_code.error != NEU_ERR_SUCCESS) {
// http_response(aio, error_code.error, result_error);
//} else {
// res.running = state.running;
// res.link = state.link;
NEU_JSON_RESPONSE_ERROR(
neu_plugin_get_node_state(plugin, node_name, &state), {
if (error_code.error != NEU_ERR_SUCCESS) {
http_response(aio, error_code.error, result_error);
} else {
res.running = state.running;
res.link = state.link;
// neu_json_encode_by_fn(&res, neu_json_encode_get_node_state_resp,
//&result);
neu_json_encode_by_fn(&res, neu_json_encode_get_node_state_resp,
&result);
// http_ok(aio, result);
// free(result);
//}
//})
http_ok(aio, result);
free(result);
}
})
}
plugins/restful/http.c
+1
View File
@@ -323,6 +323,7 @@ int http_response(nng_aio *aio, neu_err_code_e code, char *content)
case NEU_ERR_LIBRARY_INFO_INVALID:
case NEU_ERR_LICENSE_INVALID:
case NEU_ERR_GROUP_PARAMETER_INVALID:
case NEU_ERR_LIBRARY_SYSTEM_NOT_ALLOW_DEL:
status = NNG_HTTP_STATUS_BAD_REQUEST;
break;
case NEU_ERR_LIBRARY_NOT_FOUND:
src/adapter/adapter.c
+52 -70
View File
@@ -186,37 +186,34 @@ static int persister_singleton_load_plugins(neu_adapter_t *adapter)
return rv;
}
// static int persister_singleton_load_setting(neu_adapter_t *adapter,
// const char * adapter_name,
// neu_node_id_t node_id)
//{
// neu_persister_t *persister = persister_singleton_get();
// const char * setting = NULL;
//(void) node_id;
// int rv =
// neu_persister_load_adapter_setting(persister, adapter_name, &setting);
// if (0 != rv) {
// const char *fail_or_ignore = "fail";
// if (NEU_ERR_ENOENT == rv) {
//// ignore this error, no setting ever set
// rv = 0;
// fail_or_ignore = "ignore";
//}
// nlog_info("%s %s load setting of %s", adapter->name, fail_or_ignore,
// adapter_name);
// return rv;
//}
static int persister_singleton_load_setting(neu_adapter_t *adapter,
const char * adapter_name)
{
neu_persister_t *persister = persister_singleton_get();
const char * setting = NULL;
int rv =
neu_persister_load_adapter_setting(persister, adapter_name, &setting);
if (0 != rv) {
const char *fail_or_ignore = "fail";
if (NEU_ERR_ENOENT == rv) {
// ignore this error, no setting ever set
rv = 0;
fail_or_ignore = "ignore";
}
nlog_info("%s %s load setting of %s", adapter->name, fail_or_ignore,
adapter_name);
return rv;
}
//// rv = neu_manager_adapter_set_setting(adapter->manager, node_id,
//// setting);
// const char *ok_or_err = (0 == rv) ? "success" : "fail";
// nlog_info("%s %s set setting of %s %s", adapter->name, ok_or_err,
// adapter_name, setting);
rv = neu_manager_node_setting(adapter->manager, adapter_name, setting);
const char *ok_or_err = (0 == rv) ? "success" : "fail";
nlog_info("%s %s set setting of %s %s", adapter->name, ok_or_err,
adapter_name, setting);
// free((char *) setting);
free((char *) setting);
// return rv;
//}
return rv;
}
// static int persister_singleton_load_datatags(neu_adapter_t * adapter,
// const char * adapter_name,
@@ -403,27 +400,24 @@ static int persister_singleton_load_data(neu_adapter_t *adapter)
continue;
}
rv = persister_singleton_load_grp_and_tags(adapter, adapter_info->name);
if (0 != rv) {
continue;
}
persister_singleton_load_setting(adapter, adapter_info->name);
persister_singleton_load_grp_and_tags(adapter, adapter_info->name);
if (ADAPTER_STATE_RUNNING == adapter_info->state) {
// rv = neu_manager_start_adapter_with_id(adapter->manager,
// node_id); if (0 != rv) { nlog_error("%s fail start adapter
// %s", adapter->name, adapter_info->name);
//}
rv = neu_manager_node_ctl(adapter->manager, adapter_info->name,
NEU_ADAPTER_CTL_START);
if (0 != rv) {
nlog_error("%s fail start adapter %s", adapter->name,
adapter_info->name);
}
}
}
VECTOR_FOR_EACH(adapter_infos, iter)
{
neu_persist_adapter_info_t *adapter_info = iterator_get(&iter);
rv =
persister_singleton_load_subscriptions(adapter, adapter_info->name);
if (0 != rv) {
continue;
}
persister_singleton_load_subscriptions(adapter, adapter_info->name);
}
neu_persist_adapter_infos_free(adapter_infos);
@@ -894,37 +888,27 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd,
}
case NEU_REQRESP_GET_NODE_STATE: {
ADAPTER_RESP_CMD(adapter, cmd, neu_reqresp_node_state_t,
neu_cmd_get_node_state_t, rv, NEU_REQRESP_NODE_STATE,
p_result, {
ret = calloc(1, sizeof(neu_reqresp_node_state_t));
// ret->result = neu_manager_adapter_get_state(
// adapter->manager, req_cmd->node_id,
// &ret->state);
});
ADAPTER_RESP_CMD(
adapter, cmd, neu_reqresp_node_state_t, neu_cmd_get_node_state_t,
rv, NEU_REQRESP_NODE_STATE, p_result, {
ret = calloc(1, sizeof(neu_reqresp_node_state_t));
ret->result = neu_manager_get_node_state(
adapter->manager, req_cmd->node_name, &ret->state);
});
break;
}
case NEU_REQRESP_NODE_CTL: {
ADAPTER_RESP_CODE(adapter, cmd, intptr_t, neu_cmd_node_ctl_t, rv,
NEU_REQRESP_ERR_CODE, p_result, {
ret = 0;
// ret =
// neu_manager_adapter_ctl(adapter->manager,
// req_cmd->node_id, req_cmd->ctl);
// if (0 == ret) {
// char *node_name = NULL;
// 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_ADD_NODE, node_name);
// free(node_name);
//}
//}
});
ADAPTER_RESP_CODE(
adapter, cmd, intptr_t, neu_cmd_node_ctl_t, rv,
NEU_REQRESP_ERR_CODE, p_result, {
ret = neu_manager_node_ctl(adapter->manager, req_cmd->node_name,
req_cmd->ctl);
if (0 == ret) {
ADAPTER_SEND_NODE_EVENT(adapter, rv, MSG_EVENT_ADD_NODE,
req_cmd->node_name);
}
});
break;
}
@@ -1274,8 +1258,6 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
case MSG_EVENT_ADD_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);
src/core/manager.c
+35 -2
View File
@@ -288,8 +288,7 @@ int neu_manager_add_plugin(neu_manager_t *manager, const char *plugin_library)
int neu_manager_del_plugin(neu_manager_t *manager, const char *plugin)
{
neu_plugin_manager_del(manager->plugin_manager, plugin);
return 0;
return neu_plugin_manager_del(manager->plugin_manager, plugin);
}
UT_array *neu_manager_get_plugins(neu_manager_t *manager)
@@ -313,6 +312,40 @@ int neu_manager_stop_node(neu_manager_t *manager, const char *node)
return 0;
}
int neu_manager_get_node_state(neu_manager_t *manager, const char *node,
neu_plugin_state_t *state)
{
neu_adapter_t *adapter = neu_node_manager_find(manager->node_manager, node);
if (adapter == NULL) {
return NEU_ERR_NODE_NOT_EXIST;
}
*state = neu_adapter_get_state(adapter);
return NEU_ERR_SUCCESS;
}
int neu_manager_node_ctl(neu_manager_t *manager, const char *node,
neu_adapter_ctl_e ctl)
{
int ret = NEU_ERR_SUCCESS;
neu_adapter_t *adapter = neu_node_manager_find(manager->node_manager, node);
if (adapter == NULL) {
return NEU_ERR_NODE_NOT_EXIST;
}
switch (ctl) {
case NEU_ADAPTER_CTL_START:
ret = neu_adapter_start(adapter);
break;
case NEU_ADAPTER_CTL_STOP:
ret = neu_adapter_stop(adapter);
break;
}
return ret;
}
int neu_manager_node_setting(neu_manager_t *manager, const char *node,
const char *setting)
{
src/core/manager.h
+4
View File
@@ -58,6 +58,10 @@ UT_array *neu_manager_get_plugins(neu_manager_t *manager);
int neu_manager_start_node(neu_manager_t *manager, const char *node);
int neu_manager_stop_node(neu_manager_t *manager, const char *node);
int neu_manager_get_node_state(neu_manager_t *manager, const char *node,
neu_plugin_state_t *state);
int neu_manager_node_ctl(neu_manager_t *manager, const char *node,
neu_adapter_ctl_e ctl);
int neu_manager_node_setting(neu_manager_t *manager, const char *node,
const char *setting);
src/core/pluginx_manager.c
+6 -4
View File
@@ -148,14 +148,16 @@ int neu_plugin_manager_add(neu_plugin_manager_t *mgr,
int neu_plugin_manager_del(neu_plugin_manager_t *mgr, const char *plugin_name)
{
plugin_entity_t *plugin = NULL;
int ret = -1;
int ret = NEU_ERR_LIBRARY_SYSTEM_NOT_ALLOW_DEL;
nng_mtx_lock(mgr->mtx);
HASH_FIND_STR(mgr->plugins, plugin_name, plugin);
if (plugin != NULL) {
HASH_DEL(mgr->plugins, plugin);
entity_free(plugin);
ret = 0;
if (plugin->kind != NEU_PLUGIN_KIND_SYSTEM) {
HASH_DEL(mgr->plugins, plugin);
entity_free(plugin);
ret = NEU_ERR_SUCCESS;
}
}
nng_mtx_unlock(mgr->mtx);
src/parser/neu_json_node.c
+5 -5
View File
@@ -296,8 +296,8 @@ int neu_json_decode_node_ctl_req(char *buf, neu_json_node_ctl_req_t **result)
json_obj = neu_json_decode_new(buf);
neu_json_elem_t req_elems[] = { {
.name = "id",
.t = NEU_JSON_INT,
.name = "name",
.t = NEU_JSON_STR,
},
{
.name = "cmd",
@@ -309,8 +309,8 @@ int neu_json_decode_node_ctl_req(char *buf, neu_json_node_ctl_req_t **result)
goto decode_fail;
}
req->id = req_elems[0].v.val_int;
req->cmd = req_elems[1].v.val_int;
req->name = req_elems[0].v.val_str;
req->cmd = req_elems[1].v.val_int;
*result = req;
goto decode_exit;
@@ -330,7 +330,7 @@ decode_exit:
void neu_json_decode_node_ctl_req_free(neu_json_node_ctl_req_t *req)
{
free(req->name);
free(req);
}