From bd5647257e93d5867fa4d63cc19ab7367b848bfa Mon Sep 17 00:00:00 2001 From: fengzero Date: Thu, 28 Oct 2021 08:18:07 +0000 Subject: [PATCH] Add node state: running state and link state & api: start/stop node note start/stop node operation --- include/neu_adapter.h | 24 +++++++++ include/neu_plugin.h | 13 ++--- include/neu_plugin_info.h | 18 +++++++ plugins/modbus/modbus_tcp.c | 10 +++- plugins/restful/adapter_handle.c | 48 +++++++++++++++++ plugins/restful/adapter_handle.h | 2 + plugins/restful/handle.c | 18 +++++++ src/adapter/adapter.c | 47 ++++++++++++++--- src/adapter/adapter_internal.h | 1 + src/base/neu_plugin_common.c | 30 +++++++++++ src/core/neu_manager.c | 88 ++++++++++++++++++++++++++++++-- src/core/neu_manager.h | 6 +++ src/parser/neu_json_node.c | 52 +++++++++++++++++++ src/parser/neu_json_node.h | 13 +++++ 14 files changed, 348 insertions(+), 22 deletions(-) diff --git a/include/neu_adapter.h b/include/neu_adapter.h index 122d8f6f..2e201f15 100644 --- a/include/neu_adapter.h +++ b/include/neu_adapter.h @@ -75,6 +75,9 @@ typedef enum neu_reqresp_type { NEU_REQRESP_SET_NODE_SETTING, NEU_REQRESP_GET_NODE_SETTING, NEU_REQRESP_GET_NODE_SETTING_RESP, + NEU_REQRESP_NODE_STATE, + NEU_REQRESP_NODE_STATE_RESP, + NEU_REQRESP_NODE_CTL, } neu_reqresp_type_e; typedef uint32_t neu_node_id_t; @@ -279,6 +282,27 @@ typedef struct neu_reqresp_node_setting { char * setting; } neu_reqresp_node_setting_t; +/* NEU_REQRESP_GET_NODE_STATE */ +typedef struct neu_cmd_get_node_state { + neu_node_id_t node_id; +} neu_cmd_get_node_state_t; + +typedef struct neu_reqresp_node_state { + int32_t result; + neu_plugin_state_t state; +} neu_reqresp_node_state_t; + +typedef enum neu_adapter_ctl { + NEU_ADAPTER_CTL_START = 0, + NEU_ADAPTER_CTL_STOP, +} neu_adapter_ctl_e; + +/* NEU_REQRESP_NODE_CTL */ +typedef struct neu_cmd_node_ctl { + neu_node_id_t node_id; + neu_adapter_ctl_e ctl; +} neu_cmd_node_ctl_t; + typedef struct neu_request { uint32_t req_id; neu_reqresp_type_e req_type; diff --git a/include/neu_plugin.h b/include/neu_plugin.h index cc631bca..f206f85a 100644 --- a/include/neu_plugin.h +++ b/include/neu_plugin.h @@ -29,16 +29,9 @@ extern "C" { #define NEURON_PLUGIN_VER_1_0 100 #define NEURON_PLUGIN_VER_2_0 200 -typedef enum neu_plugin_state { - NEURON_PLUGIN_STATE_NULL, - NEURON_PLUGIN_STATE_IDLE, - NEURON_PLUGIN_STATE_READY, - NEURON_PLUGIN_STATE_RUNNING, -} neu_plugin_state_e; - typedef struct neu_plugin_common { uint32_t magic; - neu_plugin_state_e state; + neu_plugin_state_t state; neu_adapter_t * adapter; const adapter_callbacks_t *adapter_callbacks; uint32_t event_id; @@ -138,6 +131,10 @@ intptr_t neu_plugin_set_node_setting(neu_plugin_t *plugin, const char * setting); int32_t neu_plugin_get_node_setting(neu_plugin_t *plugin, neu_node_id_t node_id, char **setting); +int32_t neu_plugin_get_node_state(neu_plugin_t *plugin, neu_node_id_t node_id, + neu_plugin_state_t *state); +intptr_t neu_plugin_node_ctl(neu_plugin_t *plugin, neu_node_id_t node_id, + neu_adapter_ctl_e ctl); #ifdef __cplusplus } #endif diff --git a/include/neu_plugin_info.h b/include/neu_plugin_info.h index 500cab13..896fbfd1 100644 --- a/include/neu_plugin_info.h +++ b/include/neu_plugin_info.h @@ -34,6 +34,24 @@ typedef enum plugin_kind { PLUGIN_KIND_CUSTOM } plugin_kind_e; +typedef enum neu_plugin_link_state { + NEU_PLUGIN_LINK_STATE_DISCONNECTED = 0, + NEU_PLUGIN_LINK_STATE_CONNECTING, + NEU_PLUGIN_LINK_STATE_CONNECTED, +} neu_plugin_link_state_e; + +typedef enum neu_plugin_running_state { + NEU_PLUGIN_RUNNING_STATE_IDLE = 0, + NEU_PLUGIN_RUNNING_STATE_READY, + NEU_PLUGIN_RUNNING_STATE_RUNNING, + NEU_PLUGIN_RUNNING_STATE_STOPPED, +} neu_plugin_running_state_e; + +typedef struct neu_plugin_state { + neu_plugin_link_state_e link; + neu_plugin_running_state_e running; +} neu_plugin_state_t; + #ifdef __cplusplus } #endif diff --git a/plugins/modbus/modbus_tcp.c b/plugins/modbus/modbus_tcp.c index 0d2786d5..fc316dfd 100644 --- a/plugins/modbus/modbus_tcp.c +++ b/plugins/modbus/modbus_tcp.c @@ -482,7 +482,9 @@ static neu_plugin_t *modbus_tcp_open(neu_adapter_t * adapter, plugin->common.adapter = adapter; plugin->common.adapter_callbacks = callbacks; - plugin->common.state = NEURON_PLUGIN_STATE_NULL; + // plugin->common.state = NEURON_PLUGIN_STATE_NULL; + plugin->common.state.running = NEU_PLUGIN_RUNNING_STATE_IDLE; + plugin->common.state.link = NEU_PLUGIN_LINK_STATE_CONNECTING; TAILQ_INIT(&plugin->sub_instances); return plugin; @@ -499,8 +501,10 @@ static int modbus_tcp_init(neu_plugin_t *plugin) pthread_mutex_init(&plugin->mtx, NULL); plugin->client = neu_tcp_client_create("192.168.50.68", 502); - plugin->common.state = NEURON_PLUGIN_STATE_READY; + // plugin->common.state = NEURON_PLUGIN_STATE_READY; log_info("modbus tcp init....."); + plugin->common.state.running = NEU_PLUGIN_RUNNING_STATE_RUNNING; + return 0; } @@ -529,6 +533,7 @@ static int modbus_tcp_uninit(neu_plugin_t *plugin) pthread_mutex_destroy(&plugin->mtx); log_info("modbus uninit end..."); + plugin->common.state.running = NEU_PLUGIN_RUNNING_STATE_STOPPED; return 0; } @@ -542,6 +547,7 @@ static int modbus_tcp_config(neu_plugin_t *plugin, neu_config_t *configs) log_info("modbus config............. type: %d, config: %s", configs->type, configs->buf); + // plugin->common.state.running = NEU_PLUGIN_RUNNING_STATE_READY; return ret; } diff --git a/plugins/restful/adapter_handle.c b/plugins/restful/adapter_handle.c index 21069050..863a19a7 100644 --- a/plugins/restful/adapter_handle.c +++ b/plugins/restful/adapter_handle.c @@ -166,4 +166,52 @@ void handle_get_node_setting(nng_aio *aio) http_ok(aio, setting); free(setting); +} + +void handle_node_ctl(nng_aio *aio) +{ + neu_plugin_t *plugin = neu_rest_get_plugin(); + + REST_PROCESS_HTTP_REQUEST( + aio, neu_parse_node_ctl_req_t, neu_parse_decode_node_ctl, { + intptr_t err = 0; + + err = neu_plugin_node_ctl(plugin, req->node_id, req->cmd); + + if (err != 0) { + http_bad_request(aio, "{\"error\": 1}"); + } else { + http_ok(aio, "{\"error\": 0}"); + } + }) +} + +void handle_get_node_state(nng_aio *aio) +{ + neu_plugin_t * plugin = neu_rest_get_plugin(); + char * s_node_id = http_get_param(aio, "node_id"); + neu_node_id_t node_id = 0; + neu_parse_node_state_res_t res = { 0 }; + neu_plugin_state_t state = { 0 }; + char * result = NULL; + + if (s_node_id == NULL) { + http_bad_request(aio, "{\"error\": 1}"); + return; + } + + node_id = (neu_node_id_t) atoi(s_node_id); + + if (neu_plugin_get_node_state(plugin, node_id, &state) != 0) { + http_not_found(aio, "{\"error\": 1}"); + return; + } + + res.running = state.running; + res.link = state.link; + + neu_json_encode_by_fn(&res, neu_parse_encode_get_node_state, &result); + + http_ok(aio, result); + free(result); } \ No newline at end of file diff --git a/plugins/restful/adapter_handle.h b/plugins/restful/adapter_handle.h index 1fba34f8..25ddb351 100644 --- a/plugins/restful/adapter_handle.h +++ b/plugins/restful/adapter_handle.h @@ -27,5 +27,7 @@ void handle_update_adapter(nng_aio *aio); void handle_get_adapter(nng_aio *aio); void handle_set_node_setting(nng_aio *aio); void handle_get_node_setting(nng_aio *aio); +void handle_node_ctl(nng_aio *aio); +void handle_get_node_state(nng_aio *aio); #endif \ No newline at end of file diff --git a/plugins/restful/handle.c b/plugins/restful/handle.c index 71d271d4..f32758db 100644 --- a/plugins/restful/handle.c +++ b/plugins/restful/handle.c @@ -95,6 +95,12 @@ struct neu_rest_handler cors_handler[] = { { .url = "/api/v2/node/setting", }, + { + .url = "/api/v2/node/ctl", + }, + { + .url = "/api/v2/node/state", + }, }; struct neu_rest_handler api_handlers[] = { @@ -268,6 +274,18 @@ struct neu_rest_handler api_handlers[] = { .url = "/api/v2/node/setting", .value.handler = handle_get_node_setting, }, + { + .method = NEU_REST_METHOD_POST, + .type = NEU_REST_HANDLER_FUNCTION, + .url = "/api/v2/node/ctl", + .value.handler = handle_node_ctl, + }, + { + .method = NEU_REST_METHOD_GET, + .type = NEU_REST_HANDLER_FUNCTION, + .url = "/api/v2/node/state", + .value.handler = handle_get_node_state, + }, }; void neu_rest_web_handler(const struct neu_rest_handler **handlers, diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 00c4bf57..8db621a3 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -104,11 +104,10 @@ } typedef enum adapter_state { - ADAPTER_STATE_NULL, - ADAPTER_STATE_IDLE, - ADAPTER_STATE_STARTING, + ADAPTER_STATE_IDLE = 0, + ADAPTER_STATE_READY, ADAPTER_STATE_RUNNING, - ADAPTER_STATE_STOPPING, + ADAPTER_STATE_STOPPED, } adapter_state_e; struct neu_adapter { @@ -149,11 +148,12 @@ static void adapter_loop(void *arg) { int rv; neu_adapter_t *adapter; + nng_dialer dialer = { 0 }; adapter = (neu_adapter_t *) arg; const char *manager_url; manager_url = neu_manager_get_url(adapter->manager); - rv = nng_dial(adapter->sock, manager_url, NULL, 0); + rv = nng_dial(adapter->sock, manager_url, &dialer, 0); if (rv != 0) { neu_panic("The adapter can't dial to %s", manager_url); } @@ -183,7 +183,7 @@ static void adapter_loop(void *arg) nng_mtx_lock(adapter->mtx); if (adapter->stop) { - adapter->state = ADAPTER_STATE_NULL; + adapter->state = ADAPTER_STATE_STOPPED; nng_mtx_unlock(adapter->mtx); log_info("Exit loop of the adapter(%s)", adapter->name); break; @@ -396,7 +396,7 @@ static void adapter_loop(void *arg) log_info("adapter(%s) exit loop by exit_code=%d", adapter->name, exit_code); nng_mtx_lock(adapter->mtx); - adapter->state = ADAPTER_STATE_NULL; + adapter->state = ADAPTER_STATE_STOPPED; adapter->stop = true; nng_mtx_unlock(adapter->mtx); break; @@ -411,6 +411,7 @@ static void adapter_loop(void *arg) nng_msg_free(msg); } + nng_dialer_close(dialer); return; } @@ -729,6 +730,28 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd, break; } + case NEU_REQRESP_NODE_STATE: { + ADAPTER_RESP_CMD( + adapter, cmd, neu_reqresp_node_state_t, neu_cmd_get_node_state_t, + rv, NEU_REQRESP_NODE_STATE_RESP, 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); + }); + 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 = neu_manager_adapter_ctl(adapter->manager, + req_cmd->node_id, + req_cmd->ctl); + }); + break; + } + default: rv = -1; break; @@ -896,7 +919,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, } int rv; - adapter->state = ADAPTER_STATE_NULL; + adapter->state = ADAPTER_STATE_IDLE; adapter->stop = false; if ((rv = nng_mtx_alloc(&adapter->mtx)) != 0) { log_error("Can't allocate mutex for adapter"); @@ -994,6 +1017,7 @@ int neu_adapter_start(neu_adapter_t *adapter) intf_funs->init(adapter->plugin); } + adapter->stop = false; nng_thread_create(&adapter->thrd, adapter_loop, adapter); return rv; } @@ -1120,4 +1144,11 @@ int neu_adapter_get_setting(neu_adapter_t *adapter, char **config) } return -1; +} + +neu_plugin_state_t neu_adapter_get_state(neu_adapter_t *adapter) +{ + neu_plugin_common_t *common = neu_plugin_to_plugin_common(adapter->plugin); + + return common->state; } \ No newline at end of file diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h index 3e03cbbd..135556de 100644 --- a/src/adapter/adapter_internal.h +++ b/src/adapter/adapter_internal.h @@ -40,5 +40,6 @@ adapter_type_e neu_adapter_get_type(neu_adapter_t *adapter); plugin_id_t neu_adapter_get_plugin_id(neu_adapter_t *adapter); int neu_adapter_set_setting(neu_adapter_t *adapter, neu_config_t *config); int neu_adapter_get_setting(neu_adapter_t *adapter, char **config); +neu_plugin_state_t neu_adapter_get_state(neu_adapter_t *adapter); #endif diff --git a/src/base/neu_plugin_common.c b/src/base/neu_plugin_common.c index 99a9b493..b4b6b456 100644 --- a/src/base/neu_plugin_common.c +++ b/src/base/neu_plugin_common.c @@ -452,4 +452,34 @@ int32_t neu_plugin_get_node_setting(neu_plugin_t *plugin, neu_node_id_t node_id, *setting = resp->setting; }) return ret; +} + +int32_t neu_plugin_get_node_state(neu_plugin_t *plugin, neu_node_id_t node_id, + neu_plugin_state_t *state) +{ + neu_cmd_get_node_state_t get_node_state = { 0 }; + int ret = -1; + + get_node_state.node_id = node_id; + + PLUGIN_CALL_CMD(plugin, NEU_REQRESP_NODE_STATE, get_node_state, + neu_reqresp_node_state_t, { + ret = resp->result; + *state = resp->state; + }) + return ret; +} + +intptr_t neu_plugin_node_ctl(neu_plugin_t *plugin, neu_node_id_t node_id, + neu_adapter_ctl_e ctl) +{ + intptr_t errorcode = -1; + neu_cmd_node_ctl_t node_ctl = { 0 }; + + node_ctl.node_id = node_id; + node_ctl.ctl = ctl; + + PLUGIN_CALL_CMD(plugin, NEU_REQRESP_NODE_CTL, node_ctl, intptr_t, + { errorcode = (intptr_t) resp; }) + return errorcode; } \ No newline at end of file diff --git a/src/core/neu_manager.c b/src/core/neu_manager.c index c0cd2dcb..3a59e1fa 100644 --- a/src/core/neu_manager.c +++ b/src/core/neu_manager.c @@ -578,7 +578,7 @@ static int manager_unreg_adapter(neu_manager_t *manager, adapter_id_t id, return rv; } -static int manager_start_adapter(neu_manager_t *manager, neu_adapter_t *adapter) +int neu_manager_start_adapter(neu_manager_t *manager, neu_adapter_t *adapter) { int rv = 0; @@ -602,7 +602,7 @@ static int manager_start_adapter(neu_manager_t *manager, neu_adapter_t *adapter) return rv; } -static int manager_stop_adapter(neu_manager_t *manager, neu_adapter_t *adapter) +int neu_manager_stop_adapter(neu_manager_t *manager, neu_adapter_t *adapter) { int rv = 0; size_t msg_size; @@ -681,7 +681,7 @@ static void reg_and_start_default_adapters(neu_manager_t *manager) manager, (adapter_reg_cmd_t *) &default_adapter_reg_cmds[i], &p_adapter); if (id != 0 && p_adapter != NULL) { - manager_start_adapter(manager, p_adapter); + neu_manager_start_adapter(manager, p_adapter); } } return; @@ -701,7 +701,7 @@ static void stop_and_unreg_bind_adapters(neu_manager_t *manager) reg_entity = (adapter_reg_entity_t *) iterator_get(&iter); adapter = reg_entity->adapter; if (adapter != NULL) { - manager_stop_adapter(manager, adapter); + neu_manager_stop_adapter(manager, adapter); manager_unreg_adapter(manager, reg_entity->adapter_id, false); } } @@ -1828,5 +1828,85 @@ int neu_manager_adapter_get_setting(neu_manager_t *manager, ret = neu_adapter_get_setting(reg_entity->adapter, config); nng_mtx_unlock(manager->adapters_mtx); + return ret; +} + +int neu_manager_adapter_get_state(neu_manager_t *manager, neu_node_id_t node_id, + neu_plugin_state_t *state) +{ + adapter_id_t adapter_id = { 0 }; + adapter_reg_entity_t *reg_entity = NULL; + + nng_mtx_lock(manager->adapters_mtx); + + adapter_id = neu_manager_adapter_id_from_node_id(manager, node_id); + reg_entity = find_reg_adapter_by_id(&manager->reg_adapters, adapter_id); + + if (reg_entity == NULL) { + log_error("Can't find matched src registered adapter"); + nng_mtx_unlock(manager->adapters_mtx); + return -1; + } + + *state = neu_adapter_get_state(reg_entity->adapter); + nng_mtx_unlock(manager->adapters_mtx); + + return 0; +} + +int neu_manager_adapter_ctl(neu_manager_t *manager, neu_node_id_t node_id, + neu_adapter_ctl_e ctl) +{ + adapter_id_t adapter_id = { 0 }; + adapter_reg_entity_t *reg_entity = NULL; + int ret = -1; + neu_plugin_state_t state = { 0 }; + + nng_mtx_lock(manager->adapters_mtx); + + adapter_id = neu_manager_adapter_id_from_node_id(manager, node_id); + reg_entity = find_reg_adapter_by_id(&manager->reg_adapters, adapter_id); + + if (reg_entity == NULL) { + log_error("Can't find matched src registered adapter"); + nng_mtx_unlock(manager->adapters_mtx); + return -1; + } + + nng_mtx_unlock(manager->adapters_mtx); + state = neu_adapter_get_state(reg_entity->adapter); + + switch (ctl) { + case NEU_ADAPTER_CTL_START: + switch (state.running) { + case NEU_PLUGIN_RUNNING_STATE_IDLE: + ret = -1; + break; + case NEU_PLUGIN_RUNNING_STATE_READY: + case NEU_PLUGIN_RUNNING_STATE_STOPPED: + // ret = neu_manager_start_adapter(manager, reg_entity->adapter); + ret = 0; + break; + case NEU_PLUGIN_RUNNING_STATE_RUNNING: + ret = 0; + break; + } + break; + case NEU_ADAPTER_CTL_STOP: + switch (state.running) { + case NEU_PLUGIN_RUNNING_STATE_IDLE: + case NEU_PLUGIN_RUNNING_STATE_READY: + ret = -1; + break; + case NEU_PLUGIN_RUNNING_STATE_RUNNING: + // ret = neu_manager_stop_adapter(manager, reg_entity->adapter); + ret = 0; + break; + case NEU_PLUGIN_RUNNING_STATE_STOPPED: + ret = 0; + break; + } + break; + } return ret; } \ No newline at end of file diff --git a/src/core/neu_manager.h b/src/core/neu_manager.h index 34c159da..a1432650 100644 --- a/src/core/neu_manager.h +++ b/src/core/neu_manager.h @@ -68,4 +68,10 @@ int neu_manager_adapter_set_setting(neu_manager_t *manager, neu_node_id_t node_id, const char *config); int neu_manager_adapter_get_setting(neu_manager_t *manager, neu_node_id_t node_id, char **config); +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_start_adapter(neu_manager_t *manager, neu_adapter_t *adapter); +int neu_manager_stop_adapter(neu_manager_t *manager, neu_adapter_t *adapter); #endif diff --git a/src/parser/neu_json_node.c b/src/parser/neu_json_node.c index 9a152fec..90934da7 100644 --- a/src/parser/neu_json_node.c +++ b/src/parser/neu_json_node.c @@ -210,6 +210,58 @@ int neu_parse_decode_node_setting(char * buf, } void neu_parse_decode_node_setting_free(neu_parse_node_setting_req_t *req) +{ + free(req); +} + +int neu_parse_encode_get_node_state(void *json_object, void *param) +{ + neu_parse_node_state_res_t *res = (neu_parse_node_state_res_t *) param; + + neu_json_elem_t elems[] = { + { + .name = "running", + .t = NEU_JSON_INT, + .v.val_int = res->running, + }, + { + .name = "link", + .t = NEU_JSON_INT, + .v.val_int = res->link, + }, + }; + + return neu_json_encode_field(json_object, elems, NEU_JSON_ELEM_SIZE(elems)); +} + +int neu_parse_decode_node_ctl(char *buf, neu_parse_node_ctl_req_t **result) +{ + neu_parse_node_ctl_req_t *req = calloc(1, sizeof(neu_parse_node_ctl_req_t)); + neu_json_elem_t elem[] = { + { + .name = "id", + .t = NEU_JSON_INT, + }, + { + .name = "cmd", + .t = NEU_JSON_INT, + }, + }; + + int ret = neu_json_decode(buf, NEU_JSON_ELEM_SIZE(elem), elem); + if (ret != 0) { + free(req); + return -1; + } + req->node_id = elem[0].v.val_int; + req->cmd = elem[1].v.val_int; + + *result = req; + + return 0; +} + +void neu_parse_decode_node_ctl_free(neu_parse_node_ctl_req_t *req) { free(req); } \ No newline at end of file diff --git a/src/parser/neu_json_node.h b/src/parser/neu_json_node.h index a4d43567..f3a288fd 100644 --- a/src/parser/neu_json_node.h +++ b/src/parser/neu_json_node.h @@ -80,6 +80,19 @@ int neu_parse_decode_node_setting(char * buf, neu_parse_node_setting_req_t **result); void neu_parse_decode_node_setting_free(neu_parse_node_setting_req_t *req); +typedef struct { + uint8_t running; + uint8_t link; +} neu_parse_node_state_res_t; +int neu_parse_encode_get_node_state(void *json_object, void *param); + +typedef struct { + uint32_t node_id; + uint8_t cmd; +} neu_parse_node_ctl_req_t; +int neu_parse_decode_node_ctl(char *buf, neu_parse_node_ctl_req_t **result); +void neu_parse_decode_node_ctl_free(neu_parse_node_ctl_req_t *req); + #ifdef __cplusplus } #endif