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

Merge pull request #991 from fengzeroz/main

Browse Source 
remove sub msg
This commit is contained in:
fengzeroz
2022-12-20 15:19:34 +08:00
committed by GitHub
parent 1ce151176c 6bba7d4ab6
commit 168e930664
9 changed files with 0 additions and 233 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CMakeLists.txt
-1
View File
@@ -103,7 +103,6 @@ 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
-4
View File
@@ -95,10 +95,6 @@ typedef enum neu_reqresp_type {
NEU_REQ_UPDATE_LOG_LEVEL,
} 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_REQ_UPDATE_LICENSE] = "NEU_REQ_UPDATE_LICENSE",
include/neuron/plugin.h
-1
View File
@@ -94,7 +94,6 @@ 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];
bool display;
bool single;
const char * single_name;
plugins/mqtt/mqtt_plugin.c
-1
View File
@@ -381,7 +381,6 @@ const neu_plugin_module_t neu_plugin_module = {
.intf_funs = &plugin_intf_funs,
.kind = NEU_PLUGIN_KIND_SYSTEM,
.type = NEU_NA_TYPE_APP,
.sub_msg = { 0 },
.display = true,
.single = false,
};
src/core/manager.c
-54
View File
@@ -37,7 +37,6 @@
#include "node_manager.h"
#include "plugin_manager.h"
#include "storage.h"
#include "sub_msg.h"
#include "subscribe.h"
#include "manager.h"
@@ -58,7 +57,6 @@ 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 update_timestamp(void *usr_data);
static int report_nodes_state(void *usr_data);
static void start_single_adapter(neu_manager_t *manager, const char *name,
const char *plugin_name, bool display);
@@ -77,12 +75,6 @@ neu_manager_t *neu_manager_create()
.cb = update_timestamp,
.type = NEU_EVENT_TIMER_BLOCK,
};
neu_event_timer_param_t timer_param = {
.second = 1,
.millisecond = 0,
.cb = report_nodes_state,
.type = NEU_EVENT_TIMER_BLOCK,
};
neu_event_timer_param_t timer_level = {
.second = 60,
@@ -95,7 +87,6 @@ neu_manager_t *neu_manager_create()
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();
rv = nng_pair1_open_poly(&manager->socket);
assert(rv == 0);
@@ -137,9 +128,6 @@ neu_manager_t *neu_manager_create()
timer_level.usr_data = (void *) manager;
manager->timer_lev = neu_event_add_timer(manager->events, timer_level);
timer_param.usr_data = (void *) manager;
manager->timer = neu_event_add_timer(manager->events, timer_param);
timestamp_timer_param.usr_data = (void *) manager;
manager->timer_timestamp =
neu_event_add_timer(manager->events, timestamp_timer_param);
@@ -154,7 +142,6 @@ 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);
neu_event_del_timer(manager->events, manager->timer_lev);
neu_event_del_timer(manager->events, manager->timer_timestamp);
strcpy(header.sender, "manager");
@@ -177,7 +164,6 @@ void neu_manager_destroy(neu_manager_t *manager)
}
}
neu_sub_msg_manager_destroy(manager->sub_msg_manager);
neu_subscribe_manager_destroy(manager->subscribe_manager);
neu_node_manager_destroy(manager->node_manager);
neu_plugin_manager_destroy(manager->plugin_manager);
@@ -732,46 +718,6 @@ inline static void reply(neu_manager_t *manager, neu_reqresp_head_t *header,
}
}
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);
neu_manager_count_tag(manager);
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;
}
static int update_timestamp(void *usr_data)
{
(void) usr_data;
src/core/manager_internal.c
-20
View File
@@ -78,16 +78,6 @@ 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, start);
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;
}
@@ -100,16 +90,6 @@ 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
-3
View File
@@ -28,7 +28,6 @@
#include "node_manager.h"
#include "plugin_manager.h"
#include "sub_msg.h"
#include "subscribe.h"
typedef struct neu_manager {
@@ -39,9 +38,7 @@ typedef struct neu_manager {
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_event_timer_t *timer;
neu_event_timer_t *timer_timestamp;
neu_event_timer_t *timer_lev;
src/core/sub_msg.c
-108
View File
@@ -1,108 +0,0 @@
/**
* 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
@@ -1,41 +0,0 @@
/**
* 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