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

feat(manager): send nodes state periodically

Browse Source
This commit is contained in:
fengzero
2022-08-01 08:48:26 +00:00
parent 1f09eeac53
commit 4f8263d8f3
13 changed files with 312 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CMakeLists.txt
+1
View File
@@ -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
include/neuron/adapter.h
+39 -2
View File
@@ -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;
include/neuron/define.h
+1
View File
@@ -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,
include/neuron/plugin.h
+2 -1
View File
@@ -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 *
plugins/mqtt/mqtt_plugin.c
+14 -3
View File
@@ -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,
};
src/adapter/adapter.c
+1
View File
@@ -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;
src/core/manager.c
+51
View File
@@ -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;
}
src/core/manager_internal.c
+20
View File
@@ -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);
src/core/manager_internal.h
+6 -3
View File
@@ -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;
src/core/node_manager.c
+25
View File
@@ -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;
}
src/core/node_manager.h
+3
View File
@@ -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
src/core/sub_msg.c
+108
View File
@@ -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 <stdlib.h>
#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;
}
}
}
src/core/sub_msg.h
+41
View File
@@ -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