From 4f8263d8f384666f73329be668de11bbeff142a8 Mon Sep 17 00:00:00 2001 From: fengzero Date: Mon, 1 Aug 2022 08:48:26 +0000 Subject: [PATCH] feat(manager): send nodes state periodically --- CMakeLists.txt | 1 + include/neuron/adapter.h | 41 +++++++++++++- include/neuron/define.h | 1 + include/neuron/plugin.h | 3 +- plugins/mqtt/mqtt_plugin.c | 17 +++++- src/adapter/adapter.c | 1 + src/core/manager.c | 51 +++++++++++++++++ src/core/manager_internal.c | 20 +++++++ src/core/manager_internal.h | 9 ++- src/core/node_manager.c | 25 +++++++++ src/core/node_manager.h | 3 + src/core/sub_msg.c | 108 ++++++++++++++++++++++++++++++++++++ src/core/sub_msg.h | 41 ++++++++++++++ 13 files changed, 312 insertions(+), 9 deletions(-) create mode 100644 src/core/sub_msg.c create mode 100644 src/core/sub_msg.h diff --git a/CMakeLists.txt b/CMakeLists.txt index 2f5264b0..0dff0ea0 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -94,6 +94,7 @@ set(NEURON_SOURCES src/core/manager_internal.c src/core/manager.c src/core/subscribe.c + src/core/sub_msg.c src/core/plugin_manager.c src/core/node_manager.c src/core/storage.c diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 838bc640..3391d993 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -39,7 +39,6 @@ typedef struct { typedef enum neu_reqresp_type { NEU_RESP_ERROR, - NEU_REQRESP_TRANS_DATA, NEU_REQ_UPDATE_LICENSE, NEU_REQ_READ_GROUP, @@ -67,6 +66,8 @@ typedef enum neu_reqresp_type { NEU_RESP_GET_NODE_SETTING, NEU_REQ_GET_NODE_STATE, NEU_RESP_GET_NODE_STATE, + NEU_REQ_GET_NODES_STATE, + NEU_RESP_GET_NODES_STATE, NEU_REQ_NODE_CTL, NEU_REQ_ADD_GROUP, @@ -87,11 +88,17 @@ typedef enum neu_reqresp_type { NEU_REQ_DEL_PLUGIN, NEU_REQ_GET_PLUGIN, NEU_RESP_GET_PLUGIN, + + NEU_REQRESP_TRANS_DATA, + NEU_REQRESP_NODES_STATE, } neu_reqresp_type_e; +typedef enum { + NEU_SUBSCRIBE_NODES_STATE = NEU_REQRESP_NODES_STATE, +} neu_subscribe_type_e; + static const char *neu_reqresp_type_string_t[] = { [NEU_RESP_ERROR] = "NEU_RESP_ERROR", - [NEU_REQRESP_TRANS_DATA] = "NEU_REQRESP_TRANS_DATA", [NEU_REQ_UPDATE_LICENSE] = "NEU_REQ_UPDATE_LICENSE", [NEU_REQ_READ_GROUP] = "NEU_REQ_READ_GROUP", @@ -119,6 +126,8 @@ static const char *neu_reqresp_type_string_t[] = { [NEU_RESP_GET_NODE_SETTING] = "NEU_RESP_GET_NODE_SETTING", [NEU_REQ_GET_NODE_STATE] = "NEU_REQ_GET_NODE_STATE", [NEU_RESP_GET_NODE_STATE] = "NEU_RESP_GET_NODE_STATE", + [NEU_REQ_GET_NODES_STATE] = "NEU_REQ_GET_NODES_STATE", + [NEU_RESP_GET_NODES_STATE] = "NEU_RESP_GET_NODES_STATE", [NEU_REQ_NODE_CTL] = "NEU_REQ_NODE_CTL", [NEU_REQ_ADD_GROUP] = "NEU_REQ_ADD_GROUP", @@ -139,6 +148,9 @@ static const char *neu_reqresp_type_string_t[] = { [NEU_REQ_DEL_PLUGIN] = "NEU_REQ_DEL_PLUGIN", [NEU_REQ_GET_PLUGIN] = "NEU_REQ_GET_PLUGIN", [NEU_RESP_GET_PLUGIN] = "NEU_RESP_GET_PLUGIN", + + [NEU_REQRESP_TRANS_DATA] = "NEU_REQRESP_TRANS_DATA", + [NEU_REQRESP_NODES_STATE] = "NEU_REQRESP_NODE_STATE", }; inline static const char *neu_reqresp_type_string(neu_reqresp_type_e type) @@ -322,6 +334,24 @@ typedef struct neu_resp_get_node_state { neu_node_state_t state; } neu_resp_get_node_state_t; +typedef struct neu_req_get_nodes_state { +} neu_req_get_nodes_state_t; + +typedef struct { + char node[NEU_NODE_NAME_LEN]; + neu_node_state_t state; +} neu_nodes_state_t; + +inline static UT_icd neu_nodes_state_t_icd() +{ + UT_icd icd = { sizeof(neu_nodes_state_t), NULL, NULL, NULL }; + + return icd; +} +typedef struct { + UT_array *states; // array of neu_node_state_t +} neu_resp_get_nodes_state_t, neu_reqresp_nodes_state_t; + typedef struct neu_req_update_license { } neu_req_update_license_t; @@ -459,6 +489,13 @@ inline static nng_msg *neu_msg_gen(neu_reqresp_head_t *header, void *data) case NEU_RESP_GET_NODE_STATE: data_size = sizeof(neu_resp_get_node_state_t); break; + case NEU_REQ_GET_NODES_STATE: + data_size = sizeof(neu_req_get_nodes_state_t); + break; + case NEU_REQRESP_NODES_STATE: + case NEU_RESP_GET_NODES_STATE: + data_size = sizeof(neu_resp_get_nodes_state_t); + break; case NEU_REQ_READ_GROUP: data_size = sizeof(neu_req_read_group_t); break; diff --git a/include/neuron/define.h b/include/neuron/define.h index 1a942c66..f29003c3 100644 --- a/include/neuron/define.h +++ b/include/neuron/define.h @@ -30,6 +30,7 @@ #define NEU_PLUGIN_LIBRARY_LEN 32 #define NEU_PLUGIN_DESCRIPTION_LEN 256 #define NEU_DRIVER_TAG_CACHE_EXPIRE_TIME 30 +#define NEU_APP_SUBSCRIBE_MSG_SIZE 4 typedef enum neu_plugin_kind { NEU_PLUGIN_KIND_STATIC = 0, diff --git a/include/neuron/plugin.h b/include/neuron/plugin.h index 2a76ba3e..21c7e77e 100644 --- a/include/neuron/plugin.h +++ b/include/neuron/plugin.h @@ -24,7 +24,7 @@ extern "C" { #endif -#include "utils/utarray.h" +#include "utils/utextend.h" #include "utils/zlog.h" #include "adapter.h" @@ -88,6 +88,7 @@ typedef struct neu_plugin_module { const neu_plugin_intf_funs_t *intf_funs; neu_node_type_e type; neu_plugin_kind_e kind; + neu_subscribe_type_e sub_msg[NEU_APP_SUBSCRIBE_MSG_SIZE]; } neu_plugin_module_t; inline static neu_plugin_common_t * diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index ed843d0c..033f2e24 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -702,6 +702,16 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, error = node_state_update(plugin, head, data); break; } + case NEU_REQRESP_NODES_STATE: { + neu_reqresp_nodes_state_t *states = (neu_reqresp_nodes_state_t *) data; + utarray_foreach(states->states, neu_nodes_state_t *, state) + { + nlog_info("node %s: link: %d, running: %d", state->node, + state->state.link, state->state.running); + } + utarray_free(states->states); + break; + } default: error = NEU_ERR_MQTT_FAILURE; break; @@ -729,7 +739,8 @@ const neu_plugin_module_t neu_plugin_module = { "by Neuron from the device can be transmitted to the MQTT Broker " "through the MQTT application, and users can also send commands to " "Neuron through the MQTT application.", - .intf_funs = &plugin_intf_funs, - .kind = NEU_PLUGIN_KIND_SYSTEM, - .type = NEU_NA_TYPE_APP, + .intf_funs = &plugin_intf_funs, + .kind = NEU_PLUGIN_KIND_SYSTEM, + .type = NEU_NA_TYPE_APP, + .sub_msg[0] = NEU_SUBSCRIBE_NODES_STATE, }; diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 65148634..056eef66 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -261,6 +261,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_RESP_ERROR: case NEU_REQRESP_TRANS_DATA: case NEU_RESP_APP_SUBSCRIBE_GROUP: + case NEU_REQRESP_NODES_STATE: adapter->module->intf_funs->request( adapter->plugin, (neu_reqresp_head_t *) header, &header[1]); break; diff --git a/src/core/manager.c b/src/core/manager.c index 68da429e..cfa64477 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -35,6 +35,7 @@ #include "node_manager.h" #include "plugin_manager.h" #include "storage.h" +#include "sub_msg.h" #include "subscribe.h" #include "manager.h" @@ -53,6 +54,7 @@ inline static void forward_msg_dup(neu_manager_t *manager, nng_msg *msg, inline static void forward_msg(neu_manager_t *manager, nng_msg *msg, const char *ndoe); static void start_static_adapter(neu_manager_t *manager, const char *name); +static int report_nodes_state(void *usr_data); neu_manager_t *neu_manager_create() { @@ -62,11 +64,17 @@ neu_manager_t *neu_manager_create() .usr_data = (void *) manager, .cb = manager_loop, }; + neu_event_timer_param_t timer_param = { + .second = 1, + .millisecond = 0, + .cb = report_nodes_state, + }; manager->events = neu_event_new(); manager->plugin_manager = neu_plugin_manager_create(); manager->node_manager = neu_node_manager_create(); manager->subscribe_manager = neu_subscribe_manager_create(); + manager->sub_msg_manager = neu_sub_msg_manager_create(); manager->persister = neu_persister_create("persistence"); rv = nng_pair1_open_poly(&manager->socket); @@ -93,6 +101,9 @@ neu_manager_t *neu_manager_create() manager_load_subscribe(manager); } + timer_param.usr_data = (void *) manager; + manager->timer = neu_event_add_timer(manager->events, timer_param); + return manager; } @@ -103,6 +114,7 @@ void neu_manager_destroy(neu_manager_t *manager) nng_msg * uninit_msg = NULL; UT_array *pipes = neu_node_manager_get_pipes_all(manager->node_manager); + neu_event_del_timer(manager->events, manager->timer); strcpy(header.sender, "manager"); utarray_foreach(pipes, nng_pipe *, pipe) { @@ -123,6 +135,7 @@ void neu_manager_destroy(neu_manager_t *manager) } } + neu_sub_msg_manager_destroy(manager->sub_msg_manager); neu_persister_destroy(manager->persister); neu_subscribe_manager_destroy(manager->subscribe_manager); neu_node_manager_destroy(manager->node_manager); @@ -521,4 +534,42 @@ inline static void reply(neu_manager_t *manager, neu_reqresp_head_t *header, nlog_warn("reply %s to %s, error: %d", neu_reqresp_type_string(header->type), header->receiver, ret); } +} + +static int report_nodes_state(void *usr_data) +{ + neu_manager_t *manager = (neu_manager_t *) usr_data; + UT_array * apps = neu_sub_msg_manager_get(manager->sub_msg_manager, + NEU_SUBSCRIBE_NODES_STATE); + if (apps == NULL) { + return 0; + } + + UT_array *states = neu_node_manager_get_state(manager->node_manager); + + utarray_foreach(apps, char *, app) + { + nng_msg * msg = NULL; + neu_reqresp_head_t header = { 0 }; + neu_resp_get_nodes_state_t node_states = { 0 }; + nng_pipe pipe = neu_node_manager_get_pipe(manager->node_manager, app); + + header.type = NEU_REQRESP_NODES_STATE; + strcpy(header.sender, "manager"); + strcpy(header.receiver, app); + node_states.states = utarray_clone(states); + + msg = neu_msg_gen(&header, (void *) &node_states); + nng_msg_set_pipe(msg, pipe); + + if (nng_sendmsg(manager->socket, msg, 0) != 0) { + nng_msg_free(msg); + nlog_warn("manager -> %s nodes state fail", app); + } + } + + utarray_free(apps); + utarray_free(states); + + return 0; } \ No newline at end of file diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c index 891b7832..fdacaba2 100644 --- a/src/core/manager_internal.c +++ b/src/core/manager_internal.c @@ -74,6 +74,16 @@ int neu_manager_add_node(neu_manager_t *manager, const char *node_name, neu_node_manager_add(manager->node_manager, adapter); neu_adapter_init(adapter); + if (instance.module->type == NEU_NA_TYPE_APP) { + for (int i = 0; i < NEU_APP_SUBSCRIBE_MSG_SIZE; i++) { + if (instance.module->sub_msg[i] != 0) { + neu_sub_msg_manager_add(manager->sub_msg_manager, + instance.module->sub_msg[i], + adapter->name); + } + } + } + return NEU_ERR_SUCCESS; } @@ -86,6 +96,16 @@ int neu_manager_del_node(neu_manager_t *manager, const char *node_name) return NEU_ERR_NODE_NOT_EXIST; } + if (adapter->module->type == NEU_NA_TYPE_APP) { + for (int i = 0; i < NEU_APP_SUBSCRIBE_MSG_SIZE; i++) { + if (adapter->module->sub_msg[i] != 0) { + neu_sub_msg_manager_del(manager->sub_msg_manager, + adapter->module->sub_msg[i], + adapter->name); + } + } + } + neu_adapter_destroy(adapter); neu_subscribe_manager_remove(manager->subscribe_manager, node_name, NULL); neu_node_manager_del(manager->node_manager, node_name); diff --git a/src/core/manager_internal.h b/src/core/manager_internal.h index 90b27c66..88ca93a5 100644 --- a/src/core/manager_internal.h +++ b/src/core/manager_internal.h @@ -28,16 +28,19 @@ #include "node_manager.h" #include "plugin_manager.h" +#include "sub_msg.h" #include "subscribe.h" typedef struct neu_manager { - nng_socket socket; - neu_events_t * events; - neu_event_io_t *loop; + nng_socket socket; + neu_events_t * events; + neu_event_io_t * loop; + neu_event_timer_t *timer; neu_plugin_manager_t *plugin_manager; neu_node_manager_t * node_manager; neu_subscribe_mgr_t * subscribe_manager; + neu_sub_msg_mgr_t * sub_msg_manager; neu_persister_t *persister; } neu_manager_t; diff --git a/src/core/node_manager.c b/src/core/node_manager.c index 3a1e6c60..1990c040 100644 --- a/src/core/node_manager.c +++ b/src/core/node_manager.c @@ -233,3 +233,28 @@ nng_pipe neu_node_manager_get_pipe(neu_node_manager_t *mgr, const char *name) return pipe; } + +UT_array *neu_node_manager_get_state(neu_node_manager_t *mgr) +{ + UT_icd icd = { sizeof(neu_nodes_state_t), NULL, NULL, NULL }; + UT_array * states = NULL; + node_entity_t *el = NULL, *tmp = NULL; + + utarray_new(states, &icd); + + HASH_ITER(hh, mgr->nodes, el, tmp) + { + neu_nodes_state_t state = { 0 }; + + if (!el->is_static) { + strcpy(state.node, el->adapter->name); + state.state.running = el->adapter->state; + state.state.link = + neu_plugin_to_plugin_common(el->adapter->plugin)->link_state; + + utarray_push_back(states, &state); + } + } + + return states; +} diff --git a/src/core/node_manager.h b/src/core/node_manager.h index 06180878..057420c0 100644 --- a/src/core/node_manager.h +++ b/src/core/node_manager.h @@ -52,4 +52,7 @@ UT_array *neu_node_manager_get_pipes(neu_node_manager_t *mgr, UT_array *neu_node_manager_get_pipes_all(neu_node_manager_t *mgr); nng_pipe neu_node_manager_get_pipe(neu_node_manager_t *mgr, const char *name); +// neu_nodes_state_t array +UT_array *neu_node_manager_get_state(neu_node_manager_t *mgr); + #endif \ No newline at end of file diff --git a/src/core/sub_msg.c b/src/core/sub_msg.c new file mode 100644 index 00000000..ee2d5023 --- /dev/null +++ b/src/core/sub_msg.c @@ -0,0 +1,108 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ +#include + +#include "define.h" + +#include "sub_msg.h" + +struct sub_app { + neu_subscribe_type_e type; + UT_array * apps; +}; + +struct neu_sub_msg_mgr { + struct sub_app *sub_apps; +}; + +static UT_icd app_icd = { NEU_NODE_NAME_LEN, NULL, NULL, NULL }; + +neu_sub_msg_mgr_t *neu_sub_msg_manager_create() +{ + neu_sub_msg_mgr_t *mgr = calloc(1, sizeof(neu_sub_msg_mgr_t)); + + mgr->sub_apps = calloc(NEU_APP_SUBSCRIBE_MSG_SIZE, sizeof(struct sub_app)); + for (int i = 0; i < NEU_APP_SUBSCRIBE_MSG_SIZE; i++) { + utarray_new(mgr->sub_apps[i].apps, &app_icd); + } + + mgr->sub_apps[0].type = NEU_SUBSCRIBE_NODES_STATE; + + return mgr; +} + +void neu_sub_msg_manager_destroy(neu_sub_msg_mgr_t *mgr) +{ + for (int i = 0; i < NEU_APP_SUBSCRIBE_MSG_SIZE; i++) { + utarray_free(mgr->sub_apps[i].apps); + } + + free(mgr->sub_apps); + free(mgr); +} + +// app name array +UT_array *neu_sub_msg_manager_get(neu_sub_msg_mgr_t * mgr, + neu_subscribe_type_e type) +{ + UT_array *result = NULL; + + for (int i = 0; i < NEU_APP_SUBSCRIBE_MSG_SIZE; i++) { + if (mgr->sub_apps[i].type == type) { + if (utarray_len(mgr->sub_apps[i].apps) > 0) { + result = utarray_clone(mgr->sub_apps[i].apps); + } + break; + } + } + + return result; +} + +void neu_sub_msg_manager_add(neu_sub_msg_mgr_t *mgr, neu_subscribe_type_e type, + const char *app) +{ + char app_name[NEU_NODE_NAME_LEN] = { 0 }; + + for (int i = 0; i < NEU_APP_SUBSCRIBE_MSG_SIZE; i++) { + if (mgr->sub_apps[i].type == type) { + strcpy(app_name, app); + utarray_push_back(mgr->sub_apps[i].apps, app_name); + } + } +} + +void neu_sub_msg_manager_del(neu_sub_msg_mgr_t *mgr, neu_subscribe_type_e type, + const char *app) +{ + for (int i = 0; i < NEU_APP_SUBSCRIBE_MSG_SIZE; i++) { + if (mgr->sub_apps[i].type == type) { + utarray_foreach(mgr->sub_apps[i].apps, char *, app_name) + { + if (strcmp(app_name, app) == 0) { + utarray_erase( + mgr->sub_apps[i].apps, + utarray_eltidx(mgr->sub_apps[i].apps, app_name), 1); + break; + } + } + break; + } + } +} diff --git a/src/core/sub_msg.h b/src/core/sub_msg.h new file mode 100644 index 00000000..b9dd14c0 --- /dev/null +++ b/src/core/sub_msg.h @@ -0,0 +1,41 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 3 of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public License + * along with this program; if not, write to the Free Software Foundation, + * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. + **/ + +#ifndef _NEU_SUB_MSG_H_ +#define _NEU_SUB_MSG_H_ + +#include "define.h" +#include "utils/utextend.h" + +#include "adapter.h" + +typedef struct neu_sub_msg_mgr neu_sub_msg_mgr_t; + +neu_sub_msg_mgr_t *neu_sub_msg_manager_create(); +void neu_sub_msg_manager_destroy(neu_sub_msg_mgr_t *mgr); + +// app name array +UT_array *neu_sub_msg_manager_get(neu_sub_msg_mgr_t * mgr, + neu_subscribe_type_e type); +void neu_sub_msg_manager_add(neu_sub_msg_mgr_t *mgr, neu_subscribe_type_e type, + const char *app); +void neu_sub_msg_manager_del(neu_sub_msg_mgr_t *mgr, neu_subscribe_type_e type, + const char *app); + +#endif