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

refactor persist & adapter msg

Browse Source
This commit is contained in:
fengzero
2022-06-20 08:08:40 +00:00
parent 13f103ad42
commit 993530f86b
21 changed files with 1112 additions and 1446 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CMakeLists.txt
+2 -1
View File
@@ -91,6 +91,8 @@ set(NEURON_SOURCES
src/core/subscribe.c
src/core/plugin_manager.c
src/core/node_manager.c
src/core/storage.c
src/adapter/storage.c
src/adapter/adapter.c
src/adapter/driver/cache.c
src/adapter/driver/driver.c
@@ -107,7 +109,6 @@ set(NEURON_SOURCES
plugins/restful/version_handle.c
plugins/restful/rest.c
plugins/restful/http.c
plugins/dummy/dummy.c
${PERSIST_SOURCES})
set(CMAKE_BUILD_RPATH ./)
ft/tag_test.robot
+1 -1
View File
@@ -17,7 +17,7 @@ ${tag_desc} {"name": "tag_desc", "description": "description info",
Add tags to non-existent node, it should return failure
${res}= Add Tags modbus-node group ${tag1},${tag2}
Check Response Status ${res} 200
Check Response Status ${res} 404
Check Error Code ${res} ${NEU_ERR_NODE_NOT_EXIST}
Add tags to the non-existent group, it should return failure
include/neuron/adapter.h
+6 -5
View File
@@ -41,7 +41,6 @@ typedef enum neu_reqresp_type {
NEU_RESP_ERROR,
NEU_REQRESP_TRANS_DATA,
NEU_REQ_UPDATE_LICENSE,
NEU_REQRESP_PERSISTENCE_LOAD,
NEU_REQ_READ_GROUP,
NEU_RESP_READ_GROUP,
@@ -56,6 +55,8 @@ typedef enum neu_reqresp_type {
NEU_REQ_APP_UNSUBSCRIBE_GROUP,
NEU_REQ_NODE_INIT,
NEU_REQ_NODE_UNINIT,
NEU_RESP_NODE_UNINIT,
NEU_REQ_ADD_NODE,
NEU_REQ_DEL_NODE,
NEU_REQ_GET_NODE,
@@ -98,9 +99,9 @@ typedef struct neu_resp_error {
int error;
} neu_resp_error_t;
typedef struct neu_req_node_init {
typedef struct {
char node[NEU_NODE_NAME_LEN];
} neu_req_node_init_t;
} neu_req_node_init_t, neu_req_node_uninit_t, neu_resp_node_uninit_t;
typedef struct neu_req_add_plugin {
char library[NEU_PLUGIN_LIBRARY_LEN];
@@ -304,9 +305,9 @@ inline static nng_msg *neu_msg_gen(neu_reqresp_head_t *header, void *data)
size_t data_size = 0;
switch (header->type) {
case NEU_REQRESP_PERSISTENCE_LOAD:
break;
case NEU_REQ_NODE_INIT:
case NEU_REQ_NODE_UNINIT:
case NEU_RESP_NODE_UNINIT:
data_size = sizeof(neu_req_node_init_t);
break;
case NEU_RESP_ERROR:
plugins/dummy/dummy.c
-126
View File
@@ -1,126 +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 <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <nng/nng.h>
#include <nng/supplemental/http/http.h>
#include "dummy.h"
#include "plugin.h"
#include "utils/log.h"
#define neu_plugin_module default_dummy_plugin_module
struct neu_plugin {
neu_plugin_common_t common;
};
static neu_plugin_t *dummy_plugin_open()
{
neu_plugin_t *plugin = calloc(1, sizeof(neu_plugin_t));
neu_plugin_common_init(&plugin->common);
nlog_info("Success to create plugin: %s", neu_plugin_module.module_name);
return plugin;
}
static int dummy_plugin_close(neu_plugin_t *plugin)
{
free(plugin);
nlog_info("Success to free plugin: %s", neu_plugin_module.module_name);
return 0;
}
static int dummy_plugin_init(neu_plugin_t *plugin)
{
int rv = 0;
(void) plugin;
nlog_info("Initialize plugin: %s", neu_plugin_module.module_name);
return rv;
}
static int dummy_plugin_uninit(neu_plugin_t *plugin)
{
int rv = 0;
(void) plugin;
nlog_info("Uninitialize plugin: %s", neu_plugin_module.module_name);
return rv;
}
static int dummy_plugin_config(neu_plugin_t *plugin, const char *configs)
{
int rv = 0;
(void) plugin;
(void) configs;
nlog_info("config plugin: %s", neu_plugin_module.module_name);
return rv;
}
static int dummy_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
void *data)
{
(void) plugin;
(void) head;
(void) data;
return 0;
}
static int dummy_plugin_start(neu_plugin_t *plugin)
{
(void) plugin;
return 0;
}
static int dummy_plugin_stop(neu_plugin_t *plugin)
{
(void) plugin;
return 0;
}
static const neu_plugin_intf_funs_t plugin_intf_funs = {
.open = dummy_plugin_open,
.close = dummy_plugin_close,
.init = dummy_plugin_init,
.uninit = dummy_plugin_uninit,
.start = dummy_plugin_start,
.stop = dummy_plugin_stop,
.setting = dummy_plugin_config,
.request = dummy_plugin_request,
};
#define DEFAULT_DUMMY_PLUGIN_DESCR "A plugin that does nothing"
const neu_plugin_module_t default_dummy_plugin_module = {
.version = NEURON_PLUGIN_VER_1_0,
.module_name = "neuron-default-dummy",
.module_descr = DEFAULT_DUMMY_PLUGIN_DESCR,
.intf_funs = &plugin_intf_funs,
.kind = NEU_PLUGIN_KIND_STATIC,
.type = NEU_NA_TYPE_APP,
};
plugins/mqtt/mqtt.c
+3 -3
View File
@@ -519,10 +519,10 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
error = trans_data(plugin, data);
break;
case NEU_REQ_APP_SUBSCRIBE_GROUP: {
neu_req_app_subscribe_group_t *sub =
(neu_req_app_subscribe_group_t *) data;
// neu_req_app_subscribe_group_t *sub =
//(neu_req_app_subscribe_group_t *) data;
utarray_free(sub->tags);
// utarray_free(sub->tags);
break;
}
default:
src/adapter/adapter.c
+498 -678
View File
File diff suppressed because it is too large Load Diff
src/adapter/adapter_internal.h
+25 -36
View File
@@ -30,62 +30,51 @@
#include "adapter_info.h"
#include "core/manager.h"
typedef enum adapter_state {
ADAPTER_STATE_IDLE = 0,
ADAPTER_STATE_INIT,
ADAPTER_STATE_READY,
ADAPTER_STATE_RUNNING,
ADAPTER_STATE_STOPPED,
} adapter_state_e;
struct neu_adapter {
adapter_state_e state;
char * name;
neu_manager_t * manager;
char *name;
char *setting;
uint32_t req_id;
neu_node_running_state_e state;
neu_persister_t * persister;
adapter_callbacks_t cb_funs;
char * setting;
struct {
void * handle; // handle of dynamic lib
neu_plugin_module_t *module;
neu_plugin_t * plugin;
} plugin_info;
void * handle;
neu_plugin_module_t *module;
neu_plugin_t * plugin;
struct {
nng_socket sock;
nng_dialer dialer;
nng_mtx * mtx;
nng_mtx * sub_grp_mtx;
} nng;
nng_socket sock;
nng_dialer dialer;
neu_events_t * events;
neu_event_io_t *nng_io;
int recv_fd;
};
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info,
neu_manager_t * manager);
void neu_adapter_destroy(neu_adapter_t *adapter);
int neu_adapter_start(neu_adapter_t *adapter);
int neu_adapter_stop(neu_adapter_t *adapter);
int neu_adapter_init(neu_adapter_t *adapter);
int neu_adapter_uninit(neu_adapter_t *adapter);
typedef void (*adapter_handler)(neu_adapter_t * adapter,
neu_reqresp_head_t *header);
typedef struct adapter_msg_handler {
neu_reqresp_type_e type;
adapter_handler handler;
} adapter_msg_handler_t;
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool start);
void neu_adapter_init(neu_adapter_t *adapter);
int neu_adapter_start(neu_adapter_t *adapter);
int neu_adapter_stop(neu_adapter_t *adapter);
int neu_adapter_uninit(neu_adapter_t *adapter);
void neu_adapter_destroy(neu_adapter_t *adapter);
neu_event_timer_t *neu_adapter_add_timer(neu_adapter_t * adapter,
neu_event_timer_param_t param);
void neu_adapter_del_timer(neu_adapter_t *adapter, neu_event_timer_t *timer);
const char * neu_adapter_get_name(neu_adapter_t *adapter);
neu_manager_t * neu_adapter_get_manager(neu_adapter_t *adapter);
neu_adapter_type_e neu_adapter_get_type(neu_adapter_t *adapter);
int neu_adapter_set_setting(neu_adapter_t *adapter, const char *config);
int neu_adapter_get_setting(neu_adapter_t *adapter, char **config);
neu_node_state_t neu_adapter_get_state(neu_adapter_t *adapter);
neu_node_running_state_e
neu_adapter_state_to_plugin_state(neu_adapter_t *adapter);
int neu_adapter_validate_tag(neu_adapter_t *adapter, neu_datatag_t *tag);
#endif
src/adapter/driver/driver.c
+16 -25
View File
@@ -128,13 +128,10 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag,
}
}
neu_adapter_driver_t *neu_adapter_driver_create(neu_adapter_t *adapter)
neu_adapter_driver_t *neu_adapter_driver_create()
{
neu_adapter_driver_t *driver = calloc(1, sizeof(neu_adapter_driver_t));
driver->adapter = *adapter;
free(adapter);
nng_mtx_alloc(&driver->mtx);
driver->cache = neu_driver_cache_new();
driver->driver_events = neu_event_new();
@@ -265,9 +262,8 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
break;
}
driver->adapter.plugin_info.module->intf_funs->driver.write_tag(
driver->adapter.plugin_info.plugin, (void *) req, tag,
cmd->value.value);
driver->adapter.module->intf_funs->driver.write_tag(
driver->adapter.plugin, (void *) req, tag, cmd->value.value);
free(tag->address);
free(tag->name);
free(tag->description);
@@ -341,8 +337,8 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
neu_group_destroy(find->group);
free(find);
neu_plugin_to_plugin_common(driver->adapter.plugin_info.plugin)
->tag_size -= tag_size;
neu_plugin_to_plugin_common(driver->adapter.plugin)->tag_size -=
tag_size;
ret = NEU_ERR_SUCCESS;
}
@@ -406,8 +402,8 @@ int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
int ret = NEU_ERR_SUCCESS;
group_t *find = NULL;
ret = driver->adapter.plugin_info.module->intf_funs->driver.validate_tag(
driver->adapter.plugin_info.plugin, tag);
ret = driver->adapter.module->intf_funs->driver.validate_tag(
driver->adapter.plugin, tag);
if (ret != NEU_ERR_SUCCESS) {
return ret;
}
@@ -422,8 +418,7 @@ int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
nng_mtx_unlock(driver->mtx);
if (ret == NEU_ERR_SUCCESS) {
neu_plugin_to_plugin_common(driver->adapter.plugin_info.plugin)
->tag_size += 1;
neu_plugin_to_plugin_common(driver->adapter.plugin)->tag_size += 1;
}
return ret;
@@ -445,8 +440,7 @@ int neu_adapter_driver_del_tag(neu_adapter_driver_t *driver, const char *group,
nng_mtx_unlock(driver->mtx);
if (ret == NEU_ERR_SUCCESS) {
neu_plugin_to_plugin_common(driver->adapter.plugin_info.plugin)
->tag_size -= 1;
neu_plugin_to_plugin_common(driver->adapter.plugin)->tag_size -= 1;
}
return ret;
@@ -458,8 +452,8 @@ int neu_adapter_driver_update_tag(neu_adapter_driver_t *driver,
int ret = NEU_ERR_SUCCESS;
group_t *find = NULL;
ret = driver->adapter.plugin_info.module->intf_funs->driver.validate_tag(
driver->adapter.plugin_info.plugin, tag);
ret = driver->adapter.module->intf_funs->driver.validate_tag(
driver->adapter.plugin, tag);
if (ret != NEU_ERR_SUCCESS) {
return ret;
}
@@ -514,8 +508,7 @@ static int report_callback(void *usr_data)
{
group_t * group = (group_t *) usr_data;
neu_reqresp_head_t header = { 0 };
neu_node_running_state_e state =
neu_adapter_state_to_plugin_state((neu_adapter_t *) group->driver);
neu_node_running_state_e state = group->driver->adapter.state;
if (state != NEU_NODE_RUNNING_STATE_RUNNING) {
return 0;
}
@@ -531,7 +524,7 @@ static int report_callback(void *usr_data)
sizeof(neu_reqresp_trans_data_t) +
utarray_len(tags) * sizeof(neu_resp_tag_value_t));
strcpy(data->driver, neu_adapter_get_name((neu_adapter_t *) group->driver));
strcpy(data->driver, group->driver->adapter.name);
strcpy(data->group, group->name);
data->n_tag = utarray_len(tags);
read_group(group->driver->cache, group->name, tags,
@@ -580,8 +573,7 @@ static void group_change(void *arg, int64_t timestamp, UT_array *tags,
static int read_callback(void *usr_data)
{
group_t * group = (group_t *) usr_data;
neu_node_running_state_e state =
neu_adapter_state_to_plugin_state((neu_adapter_t *) group->driver);
neu_node_running_state_e state = group->driver->adapter.state;
if (state != NEU_NODE_RUNNING_STATE_RUNNING) {
return 0;
}
@@ -589,9 +581,8 @@ static int read_callback(void *usr_data)
nng_mtx_lock(group->driver->mtx);
if (!neu_group_is_change(group->group, group->timestamp)) {
if (group->grp.tags != NULL) {
group->driver->adapter.plugin_info.module->intf_funs->driver
.group_timer(group->driver->adapter.plugin_info.plugin,
&group->grp);
group->driver->adapter.module->intf_funs->driver.group_timer(
group->driver->adapter.plugin, &group->grp);
}
} else {
neu_group_change_test(group->group, group->timestamp, (void *) group,
src/adapter/driver/driver_internal.h
+1 -1
View File
@@ -22,7 +22,7 @@
#include "adapter.h"
neu_adapter_driver_t *neu_adapter_driver_create(neu_adapter_t *adapter);
neu_adapter_driver_t *neu_adapter_driver_create();
void neu_adapter_driver_destroy(neu_adapter_driver_t *driver);
int neu_adapter_driver_start(neu_adapter_driver_t *driver);
src/adapter/storage.c
+120
View File
@@ -0,0 +1,120 @@
/**
* 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 "utils/log.h"
#include "adapter_internal.h"
#include "driver/driver_internal.h"
#include "storage.h"
void adapter_storage_setting(neu_persister_t *persister, const char *node,
const char *setting)
{
neu_persister_store_adapter_setting(persister, node, setting);
}
void adapter_storage_add_group(neu_persister_t *persister, const char *node,
const char *group, uint32_t interval)
{
neu_persist_group_config_info_t info = {
.group_config_name = (char *) group,
.read_interval = interval,
.adapter_name = (char *) node,
};
neu_persister_store_group_config(persister, node, &info);
}
void adapter_storage_del_group(neu_persister_t *persister, const char *node,
const char *group)
{
neu_persister_delete_group_config(persister, node, group);
}
void adapter_storage_tag(neu_persister_t *persister, neu_adapter_t *adapter,
const char *node, const char *group)
{
UT_array *tags = NULL;
int rv = neu_adapter_driver_get_tag((neu_adapter_driver_t *) adapter, group,
&tags);
if (rv != 0) {
nlog_warn("fail get tag infos");
return;
}
rv = neu_persister_store_datatags(persister, node, group, tags);
if (0 != rv) {
nlog_error("fail store adapter:%s grp:%s datatag infos", node, group);
}
neu_persist_datatag_infos_free(tags);
}
int adapter_load_setting(neu_persister_t *persister, const char *node,
char **setting)
{
int rv = neu_persister_load_adapter_setting(persister, node,
(const char **) setting);
if (0 != rv) {
nlog_info("load %s setting fail", node);
return -1;
}
return 0;
}
int adapter_load_group_and_tag(neu_persister_t * persister,
neu_adapter_driver_t *driver)
{
UT_array * group_infos = NULL;
neu_adapter_t *adapter = (neu_adapter_t *) driver;
int rv = neu_persister_load_group_configs(persister, adapter->name,
&group_infos);
if (0 != rv) {
nlog_warn("load %s group fail", adapter->name);
return rv;
}
utarray_foreach(group_infos, neu_persist_group_config_info_t *, p)
{
UT_array *tags = NULL;
neu_adapter_driver_add_group(driver, p->group_config_name,
p->read_interval);
rv = neu_persister_load_datatags(persister, adapter->name,
p->group_config_name, &tags);
if (0 != rv) {
nlog_warn("load %s:%s tags fail", adapter->name,
p->group_config_name);
continue;
}
utarray_foreach(tags, neu_datatag_t *, tag)
{
neu_adapter_driver_add_tag(driver, p->group_config_name, tag);
}
neu_persist_datatag_infos_free(tags);
}
neu_persist_group_config_infos_free(group_infos);
return rv;
}
src/adapter/storage.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 ADAPTER_STORAGE_H
#define ADAPTER_STORAGE_H
#include "persist/persist.h"
#include "adapter_internal.h"
void adapter_storage_setting(neu_persister_t *persister, const char *node,
const char *setting);
void adapter_storage_add_group(neu_persister_t *persister, const char *node,
const char *group, uint32_t interval);
void adapter_storage_del_group(neu_persister_t *persister, const char *node,
const char *group);
void adapter_storage_tag(neu_persister_t *persister, neu_adapter_t *adapter,
const char *node, const char *group);
int adapter_load_setting(neu_persister_t *persister, const char *node,
char **setting);
int adapter_load_group_and_tag(neu_persister_t * persister,
neu_adapter_driver_t *driver);
#endif
src/core/manager.c
+189 -277
View File
@@ -17,6 +17,7 @@
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#include <stdlib.h>
#include <unistd.h>
#include <nng/nng.h>
#include <nng/protocol/pair1/pair.h>
@@ -33,6 +34,7 @@
#include "node_manager.h"
#include "plugin_manager.h"
#include "storage.h"
#include "subscribe.h"
#include "manager.h"
@@ -40,7 +42,6 @@
// definition for adapter names
#define DEFAULT_DASHBOARD_ADAPTER_NAME DEFAULT_DASHBOARD_PLUGIN_NAME
#define DEFAULT_PERSIST_ADAPTER_NAME DEFAULT_DUMMY_PLUGIN_NAME
static const char *const url = "inproc://neu_manager";
@@ -52,7 +53,6 @@ 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 void stop_static_adapter(neu_manager_t *manager, const char *name);
neu_manager_t *neu_manager_create()
{
@@ -67,7 +67,7 @@ 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->stop = false;
manager->persister = neu_persister_create("persistence");
nng_pair1_open_poly(&manager->socket);
@@ -78,17 +78,41 @@ neu_manager_t *neu_manager_create()
manager->loop = neu_event_add_io(manager->events, param);
start_static_adapter(manager, DEFAULT_DASHBOARD_PLUGIN_NAME);
start_static_adapter(manager, DEFAULT_DUMMY_PLUGIN_NAME);
if (manager_load_plugin(manager) != 0) {
nlog_warn("load plugin error");
} else {
manager_load_node(manager);
manager_load_subscribe(manager);
}
return manager;
}
void neu_manager_destroy(neu_manager_t *manager)
{
stop_static_adapter(manager, DEFAULT_DASHBOARD_PLUGIN_NAME);
stop_static_adapter(manager, DEFAULT_DUMMY_PLUGIN_NAME);
neu_reqresp_head_t header = { .type = NEU_REQ_NODE_UNINIT };
neu_req_node_init_t uninit = { 0 };
nng_msg * uninit_msg = NULL;
UT_array *pipes = neu_node_manager_get_pipes_all(manager->node_manager);
manager->stop = true;
utarray_foreach(pipes, nng_pipe *, pipe)
{
uninit_msg = neu_msg_gen(&header, &uninit);
nng_msg_set_pipe(uninit_msg, *pipe);
nng_sendmsg(manager->socket, uninit_msg, 0);
}
utarray_free(pipes);
while (1) {
usleep(1000 * 100);
if (neu_node_manager_size(manager->node_manager) == 0) {
break;
}
}
neu_persister_destroy(manager->persister);
neu_subscribe_manager_destroy(manager->subscribe_manager);
neu_node_manager_destroy(manager->node_manager);
neu_plugin_manager_destroy(manager->plugin_manager);
@@ -101,9 +125,8 @@ void neu_manager_destroy(neu_manager_t *manager)
nlog_warn("manager exit");
}
const char *neu_manager_get_url(neu_manager_t *manager)
const char *neu_manager_get_url()
{
(void) manager;
return url;
}
@@ -127,44 +150,52 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
header = (neu_reqresp_head_t *) nng_msg_body(msg);
if (manager->stop) {
nlog_warn("manager stopped, drop from %s msg: %d", header->sender,
header->type);
nng_msg_free(msg);
return 0;
}
nlog_info("manager recv msg from: %s, type: %d", header->sender,
header->type);
switch (header->type) {
case NEU_REQRESP_TRANS_DATA: {
neu_reqresp_trans_data_t *cmd = (neu_reqresp_trans_data_t *) &header[1];
UT_array *apps = neu_subscribe_manager_find(manager->subscribe_manager,
cmd->driver, cmd->group);
if (apps != NULL) {
utarray_foreach(apps, neu_app_subscribe_t *, app)
{
forward_msg_dup(manager, msg, app->pipe);
nlog_debug("forward trans data to pipe: %d", app->pipe.id);
}
utarray_free(apps);
break;
}
break;
}
case NEU_REQ_UPDATE_LICENSE: {
UT_array *pipes = neu_node_manager_get_pipes(manager->node_manager,
NEU_NA_TYPE_DRIVER);
utarray_foreach(pipes, nng_pipe *, pipe)
{
forward_msg_dup(manager, msg, *pipe);
nlog_info("forward license update to pipe: %d", pipe->id);
}
utarray_free(pipes);
break;
}
case NEU_REQ_NODE_INIT: {
neu_req_node_init_t *init = (neu_req_node_init_t *) &header[1];
nng_pipe pipe = nng_msg_get_pipe(msg);
neu_node_manager_update(manager->node_manager, init->node, pipe);
nlog_notice("bind node %s to pipe(%d)", init->node, pipe.id);
if (strcmp(DEFAULT_DUMMY_PLUGIN_NAME, init->node) == 0) {
header->type = NEU_REQRESP_PERSISTENCE_LOAD;
nng_msg *out_msg = neu_msg_gen(header, NULL);
manager->persist_pipe = pipe;
nng_msg_set_pipe(out_msg, manager->persist_pipe);
nng_sendmsg(manager->socket, out_msg, 0);
nlog_info("send persistence load to pipe: %d", pipe.id);
}
break;
}
case NEU_RESP_ERROR:
forward_msg(manager, msg, header->receiver);
break;
case NEU_REQ_ADD_PLUGIN: {
neu_req_add_plugin_t *cmd = (neu_req_add_plugin_t *) &header[1];
int error = neu_manager_add_plugin(manager, cmd->library);
neu_resp_error_t e = { .error = error };
if (error == NEU_ERR_SUCCESS) {
forward_msg_dup(manager, msg, manager->persist_pipe);
manager_strorage_plugin(manager);
}
header->type = NEU_RESP_ERROR;
@@ -177,6 +208,10 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
int error = neu_manager_del_plugin(manager, cmd->plugin);
neu_resp_error_t e = { .error = error };
if (error == NEU_ERR_SUCCESS) {
manager_strorage_plugin(manager);
}
header->type = NEU_RESP_ERROR;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
@@ -197,7 +232,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
neu_resp_error_t e = { .error = error };
if (error == NEU_ERR_SUCCESS) {
forward_msg_dup(manager, msg, manager->persist_pipe);
manager_storage_add_node(manager, cmd->node);
}
header->type = NEU_RESP_ERROR;
@@ -207,16 +242,31 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
case NEU_REQ_DEL_NODE: {
neu_req_del_node_t *cmd = (neu_req_del_node_t *) &header[1];
int error = neu_manager_del_node(manager, cmd->node);
neu_resp_error_t e = { .error = error };
neu_resp_error_t error = { 0 };
if (error == NEU_ERR_SUCCESS) {
forward_msg_dup(manager, msg, manager->persist_pipe);
strcpy(header->receiver, cmd->node);
if (neu_node_manager_find(manager->node_manager, header->receiver) !=
NULL) {
header->type = NEU_REQ_NODE_UNINIT;
forward_msg(manager, msg, header->receiver);
} else {
error.error = NEU_ERR_NODE_NOT_EXIST;
header->type = NEU_RESP_ERROR;
neu_msg_exchange(header);
reply(manager, header, &error);
}
break;
}
case NEU_RESP_NODE_UNINIT: {
neu_resp_node_uninit_t *cmd = (neu_resp_node_uninit_t *) &header[1];
header->type = NEU_RESP_ERROR;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
neu_manager_del_node(manager, cmd->node);
manager_storage_del_node(manager, cmd->node);
if (strlen(header->receiver) > 0) {
neu_resp_error_t error = { 0 };
header->type = NEU_RESP_ERROR;
reply(manager, header, &error);
}
break;
}
case NEU_REQ_GET_NODE: {
@@ -229,137 +279,6 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
reply(manager, header, &resp);
break;
}
case NEU_REQ_ADD_GROUP: {
neu_req_add_group_t *cmd = (neu_req_add_group_t *) &header[1];
int error = neu_manager_add_group(manager, cmd->driver, cmd->group,
cmd->interval);
neu_resp_error_t e = { .error = error };
if (error == NEU_ERR_SUCCESS) {
forward_msg_dup(manager, msg, manager->persist_pipe);
}
header->type = NEU_RESP_ERROR;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
break;
}
case NEU_REQ_DEL_GROUP: {
neu_req_del_group_t *cmd = (neu_req_del_group_t *) &header[1];
int error = neu_manager_del_group(manager, cmd->driver, cmd->group);
neu_resp_error_t e = { .error = error };
if (error == NEU_ERR_SUCCESS) {
forward_msg_dup(manager, msg, manager->persist_pipe);
}
header->type = NEU_RESP_ERROR;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
break;
}
case NEU_REQ_GET_GROUP: {
neu_req_get_group_t *cmd = (neu_req_get_group_t *) &header[1];
UT_array * groups = NULL;
int ret = neu_manager_get_group(manager, cmd->driver, &groups);
neu_resp_error_t e = { .error = ret };
neu_resp_get_group_t resp = { .groups = groups };
strcpy(header->receiver, header->sender);
if (ret != NEU_ERR_SUCCESS) {
header->type = NEU_RESP_ERROR;
reply(manager, header, &e);
} else {
header->type = NEU_RESP_GET_GROUP;
reply(manager, header, &resp);
}
break;
}
case NEU_REQ_ADD_TAG: {
neu_req_add_tag_t *cmd = (neu_req_add_tag_t *) &header[1];
neu_resp_add_tag_t resp = { 0 };
resp.error = neu_manager_add_tag(manager, cmd->driver, cmd->group,
cmd->n_tag, cmd->tags, &resp.index);
if (resp.error == NEU_ERR_SUCCESS) {
forward_msg_dup(manager, msg, manager->persist_pipe);
neu_manager_notify_app_sub_update(manager, cmd->driver, cmd->group);
}
for (int i = 0; i < cmd->n_tag; i++) {
free(cmd->tags[i].address);
free(cmd->tags[i].name);
free(cmd->tags[i].description);
}
free(cmd->tags);
header->type = NEU_RESP_ADD_TAG;
strcpy(header->receiver, header->sender);
reply(manager, header, &resp);
break;
}
case NEU_REQ_DEL_TAG: {
neu_req_del_tag_t *cmd = (neu_req_del_tag_t *) &header[1];
neu_resp_error_t error = { 0 };
error.error = neu_manager_del_tag(manager, cmd->driver, cmd->group,
cmd->n_tag, cmd->tags);
if (error.error == NEU_ERR_SUCCESS) {
forward_msg_dup(manager, msg, manager->persist_pipe);
neu_manager_notify_app_sub_update(manager, cmd->driver, cmd->group);
}
for (int i = 0; i < cmd->n_tag; i++) {
free(cmd->tags[i]);
}
free(cmd->tags);
header->type = NEU_RESP_ERROR;
strcpy(header->receiver, header->sender);
reply(manager, header, &error);
break;
}
case NEU_REQ_UPDATE_TAG: {
neu_req_update_tag_t *cmd = (neu_req_update_tag_t *) &header[1];
neu_resp_update_tag_t resp = { 0 };
resp.error = neu_manager_update_tag(manager, cmd->driver, cmd->group,
cmd->n_tag, cmd->tags, &resp.index);
if (resp.error == NEU_ERR_SUCCESS) {
forward_msg_dup(manager, msg, manager->persist_pipe);
neu_manager_notify_app_sub_update(manager, cmd->driver, cmd->group);
}
for (int i = 0; i < cmd->n_tag; i++) {
free(cmd->tags[i].address);
free(cmd->tags[i].name);
free(cmd->tags[i].description);
}
free(cmd->tags);
header->type = NEU_RESP_UPDATE_TAG;
strcpy(header->receiver, header->sender);
reply(manager, header, &resp);
break;
}
case NEU_REQ_GET_TAG: {
neu_req_get_tag_t *cmd = (neu_req_get_tag_t *) &header[1];
UT_array * tags = NULL;
int ret = neu_manager_get_tag(manager, cmd->driver, cmd->group, &tags);
neu_resp_error_t e = { .error = ret };
neu_resp_get_tag_t resp = { .tags = tags };
strcpy(header->receiver, header->sender);
if (ret != NEU_ERR_SUCCESS) {
header->type = NEU_RESP_ERROR;
reply(manager, header, &e);
} else {
header->type = NEU_RESP_GET_TAG;
reply(manager, header, &resp);
}
break;
}
case NEU_REQ_SUBSCRIBE_GROUP: {
neu_req_subscribe_t *cmd = (neu_req_subscribe_t *) &header[1];
neu_resp_error_t error = { 0 };
@@ -368,7 +287,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
neu_manager_subscribe(manager, cmd->app, cmd->driver, cmd->group);
if (error.error == NEU_ERR_SUCCESS) {
forward_msg_dup(manager, msg, manager->persist_pipe);
manager_storage_subscribe(manager, cmd->app);
neu_manager_notify_app_sub(manager, cmd->app, cmd->driver,
cmd->group);
}
@@ -386,7 +305,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
neu_manager_unsubscribe(manager, cmd->app, cmd->driver, cmd->group);
if (error.error == NEU_ERR_SUCCESS) {
forward_msg_dup(manager, msg, manager->persist_pipe);
manager_storage_subscribe(manager, cmd->app);
neu_manager_notify_app_unsub(manager, cmd->app, cmd->driver,
cmd->group);
}
@@ -407,109 +326,111 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
reply(manager, header, &resp);
break;
}
case NEU_REQ_NODE_SETTING: {
neu_req_node_setting_t *cmd = (neu_req_node_setting_t *) &header[1];
neu_resp_error_t error = { 0 };
error.error =
neu_manager_node_setting(manager, cmd->node, cmd->setting);
if (error.error == NEU_ERR_SUCCESS) {
forward_msg_dup(manager, msg, manager->persist_pipe);
} else {
free(cmd->setting);
}
header->type = NEU_RESP_ERROR;
strcpy(header->receiver, header->sender);
reply(manager, header, &error);
break;
}
case NEU_REQ_GET_NODE_SETTING: {
neu_req_get_node_setting_t *cmd =
(neu_req_get_node_setting_t *) &header[1];
neu_resp_get_node_setting_t resp = { 0 };
neu_resp_error_t error = { 0 };
error.error =
neu_manager_node_get_setting(manager, cmd->ndoe, &resp.setting);
strcpy(header->receiver, header->sender);
if (error.error != NEU_ERR_SUCCESS) {
header->type = NEU_RESP_ERROR;
reply(manager, header, &error);
} else {
header->type = NEU_RESP_GET_NODE_SETTING;
reply(manager, header, &resp);
}
break;
}
case NEU_REQ_GET_NODE_STATE: {
neu_req_get_node_state_t *cmd = (neu_req_get_node_state_t *) &header[1];
neu_resp_get_node_state_t resp = { 0 };
neu_resp_error_t error = { 0 };
error.error =
neu_manager_get_node_state(manager, cmd->node, &resp.state);
strcpy(header->receiver, header->sender);
if (error.error != NEU_ERR_SUCCESS) {
header->type = NEU_RESP_ERROR;
reply(manager, header, &error);
} else {
header->type = NEU_RESP_GET_NODE_STATE;
reply(manager, header, &resp);
}
break;
}
case NEU_REQ_NODE_CTL: {
neu_req_node_ctl_t *cmd = (neu_req_node_ctl_t *) &header[1];
neu_resp_error_t error = { 0 };
error.error = neu_manager_node_ctl(manager, cmd->node, cmd->ctl);
if (error.error == NEU_ERR_SUCCESS) {
forward_msg_dup(manager, msg, manager->persist_pipe);
}
strcpy(header->receiver, header->sender);
header->type = NEU_RESP_ERROR;
reply(manager, header, &error);
break;
}
case NEU_REQ_GET_NODE_SETTING:
case NEU_REQ_READ_GROUP:
case NEU_REQ_WRITE_TAG:
case NEU_REQ_GET_NODE_STATE:
case NEU_REQ_GET_GROUP:
case NEU_REQ_GET_TAG:
case NEU_REQ_NODE_CTL:
case NEU_REQ_ADD_GROUP: {
if (neu_node_manager_find(manager->node_manager, header->receiver) ==
NULL) {
neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST };
header->type = NEU_RESP_ERROR;
neu_msg_exchange(header);
reply(manager, header, &e);
} else {
forward_msg(manager, msg, header->receiver);
}
break;
}
case NEU_REQ_DEL_GROUP: {
neu_req_del_group_t *cmd = (neu_req_del_group_t *) &header[1];
if (neu_node_manager_find(manager->node_manager, header->receiver) ==
NULL) {
neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST };
header->type = NEU_RESP_ERROR;
neu_msg_exchange(header);
reply(manager, header, &e);
} else {
forward_msg(manager, msg, header->receiver);
neu_subscribe_manager_remove(manager->subscribe_manager,
cmd->driver, cmd->group);
}
break;
}
case NEU_REQ_DEL_TAG: {
neu_req_del_tag_t *cmd = (neu_req_del_tag_t *) &header[1];
if (neu_node_manager_find(manager->node_manager, header->receiver) ==
NULL) {
neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST };
header->type = NEU_RESP_ERROR;
neu_msg_exchange(header);
reply(manager, header, &e);
for (int i = 0; i < cmd->n_tag; i++) {
free(cmd->tags[i]);
}
free(cmd->tags);
} else {
forward_msg(manager, msg, header->receiver);
}
break;
}
case NEU_REQ_UPDATE_TAG:
case NEU_REQ_ADD_TAG: {
neu_req_add_tag_t *cmd = (neu_req_add_tag_t *) &header[1];
if (neu_node_manager_find(manager->node_manager, header->receiver) ==
NULL) {
neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST };
header->type = NEU_RESP_ERROR;
neu_msg_exchange(header);
reply(manager, header, &e);
for (int i = 0; i < cmd->n_tag; i++) {
free(cmd->tags[i].address);
free(cmd->tags[i].name);
free(cmd->tags[i].description);
}
free(cmd->tags);
} else {
forward_msg(manager, msg, header->receiver);
}
break;
}
case NEU_REQ_NODE_SETTING: {
neu_req_node_setting_t *cmd = (neu_req_node_setting_t *) &header[1];
if (neu_node_manager_find(manager->node_manager, header->receiver) ==
NULL) {
neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST };
header->type = NEU_RESP_ERROR;
neu_msg_exchange(header);
reply(manager, header, &e);
free(cmd->setting);
} else {
forward_msg(manager, msg, header->receiver);
}
break;
}
case NEU_RESP_ADD_TAG:
case NEU_RESP_UPDATE_TAG:
case NEU_RESP_GET_TAG:
case NEU_RESP_GET_GROUP:
case NEU_RESP_GET_NODE_SETTING:
case NEU_RESP_GET_NODE_STATE:
case NEU_RESP_ERROR:
case NEU_RESP_READ_GROUP:
forward_msg(manager, msg, header->receiver);
break;
case NEU_REQ_UPDATE_LICENSE: {
UT_array *pipes = neu_node_manager_get_pipes(manager->node_manager,
NEU_NA_TYPE_DRIVER);
utarray_foreach(pipes, nng_pipe *, pipe)
{
forward_msg_dup(manager, msg, *pipe);
nlog_info("forward license update to pipe: %d", pipe->id);
}
utarray_free(pipes);
break;
}
case NEU_REQRESP_TRANS_DATA: {
neu_reqresp_trans_data_t *cmd = (neu_reqresp_trans_data_t *) &header[1];
UT_array *apps = neu_subscribe_manager_find(manager->subscribe_manager,
cmd->driver, cmd->group);
if (apps != NULL) {
utarray_foreach(apps, neu_app_subscribe_t *, app)
{
forward_msg_dup(manager, msg, app->pipe);
nlog_debug("forward trans data to pipe: %d", app->pipe.id);
}
utarray_free(apps);
break;
}
break;
}
default:
assert(false);
break;
@@ -560,15 +481,6 @@ static void start_static_adapter(neu_manager_t *manager, const char *name)
neu_adapter_start(adapter);
}
static void stop_static_adapter(neu_manager_t *manager, const char *name)
{
neu_adapter_t *adapter = neu_node_manager_find(manager->node_manager, name);
neu_adapter_uninit(adapter);
neu_adapter_destroy(adapter);
neu_node_manager_del(manager->node_manager, name);
}
inline static void reply(neu_manager_t *manager, neu_reqresp_head_t *header,
void *data)
{
src/core/manager.h
+1 -1
View File
@@ -29,6 +29,6 @@
neu_manager_t *neu_manager_create();
void neu_manager_destroy(neu_manager_t *manager);
const char *neu_manager_get_url(neu_manager_t *manager);
const char *neu_manager_get_url(void);
#endif
src/core/manager_internal.c
+5 -237
View File
@@ -66,7 +66,7 @@ int neu_manager_add_node(neu_manager_t *manager, const char *node_name,
adapter_info.handle = instance.handle;
adapter_info.module = instance.module;
adapter = neu_adapter_create(&adapter_info, manager);
adapter = neu_adapter_create(&adapter_info, true);
neu_node_manager_add(manager->node_manager, adapter);
neu_adapter_init(adapter);
@@ -82,9 +82,7 @@ int neu_manager_del_node(neu_manager_t *manager, const char *node_name)
return NEU_ERR_NODE_NOT_EXIST;
}
neu_adapter_uninit(adapter);
neu_adapter_destroy(adapter);
neu_subscribe_manager_remove(manager->subscribe_manager, node_name, NULL);
neu_node_manager_del(manager->node_manager, node_name);
return NEU_ERR_SUCCESS;
@@ -95,180 +93,6 @@ UT_array *neu_manager_get_nodes(neu_manager_t *manager, neu_node_type_e type)
return neu_node_manager_get(manager->node_manager, type);
}
int neu_manager_add_group(neu_manager_t *manager, const char *driver,
const char *group, uint32_t interval)
{
if (interval < NEU_GROUP_INTERVAL_LIMIT) {
return NEU_ERR_GROUP_PARAMETER_INVALID;
}
if (strlen(group) > NEU_GROUP_NAME_LEN) {
return NEU_ERR_GROUP_PARAMETER_INVALID;
}
neu_adapter_t *adapter =
neu_node_manager_find(manager->node_manager, driver);
if (adapter == NULL) {
return NEU_ERR_NODE_NOT_EXIST;
}
if (adapter->plugin_info.module->type != NEU_NA_TYPE_DRIVER) {
return NEU_ERR_GROUP_NOT_ALLOW;
}
return neu_adapter_driver_add_group((neu_adapter_driver_t *) adapter, group,
interval);
}
int neu_manager_del_group(neu_manager_t *manager, const char *driver,
const char *group)
{
neu_adapter_t *adapter =
neu_node_manager_find(manager->node_manager, driver);
if (adapter == NULL) {
return NEU_ERR_NODE_NOT_EXIST;
}
if (adapter->plugin_info.module->type != NEU_NA_TYPE_DRIVER) {
return NEU_ERR_GROUP_NOT_ALLOW;
}
neu_subscribe_manager_remove(manager->subscribe_manager, driver, group);
return neu_adapter_driver_del_group((neu_adapter_driver_t *) adapter,
group);
}
int neu_manager_update_group(neu_manager_t *manager, const char *driver,
const char *group, uint32_t interval)
{
if (interval < NEU_GROUP_INTERVAL_LIMIT) {
return NEU_ERR_GROUP_PARAMETER_INVALID;
}
(void) manager;
(void) driver;
(void) group;
(void) interval;
return 0;
}
int neu_manager_get_group(neu_manager_t *manager, const char *driver,
UT_array **groups)
{
neu_adapter_t *adapter =
neu_node_manager_find(manager->node_manager, driver);
if (adapter == NULL) {
return NEU_ERR_NODE_NOT_EXIST;
}
if (adapter->plugin_info.module->type != NEU_NA_TYPE_DRIVER) {
return NEU_ERR_GROUP_NOT_ALLOW;
}
*groups = neu_adapter_driver_get_group((neu_adapter_driver_t *) adapter);
return NEU_ERR_SUCCESS;
}
int neu_manager_add_tag(neu_manager_t *manager, const char *driver,
const char *group, uint16_t n_tag, neu_datatag_t *tags,
uint16_t *index)
{
neu_adapter_t *adapter =
neu_node_manager_find(manager->node_manager, driver);
*index = 0;
if (adapter == NULL) {
return NEU_ERR_NODE_NOT_EXIST;
}
if (adapter->plugin_info.module->type != NEU_NA_TYPE_DRIVER) {
return NEU_ERR_GROUP_NOT_ALLOW;
}
for (int i = 0; i < n_tag; i++) {
int ret = neu_adapter_driver_add_tag((neu_adapter_driver_t *) adapter,
group, &tags[i]);
if (ret == NEU_ERR_SUCCESS) {
*index += 1;
} else {
return ret;
}
}
return NEU_ERR_SUCCESS;
}
int neu_manager_del_tag(neu_manager_t *manager, const char *driver,
const char *group, uint16_t n_tag, char **tags)
{
neu_adapter_t *adapter =
neu_node_manager_find(manager->node_manager, driver);
if (adapter == NULL) {
return NEU_ERR_NODE_NOT_EXIST;
}
if (adapter->plugin_info.module->type != NEU_NA_TYPE_DRIVER) {
return NEU_ERR_GROUP_NOT_ALLOW;
}
for (int i = 0; i < n_tag; i++) {
neu_adapter_driver_del_tag((neu_adapter_driver_t *) adapter, group,
tags[i]);
}
return NEU_ERR_SUCCESS;
}
int neu_manager_update_tag(neu_manager_t *manager, const char *driver,
const char *group, uint16_t n_tag,
neu_datatag_t *tags, uint16_t *index)
{
neu_adapter_t *adapter =
neu_node_manager_find(manager->node_manager, driver);
*index = 0;
if (adapter == NULL) {
return NEU_ERR_NODE_NOT_EXIST;
}
if (adapter->plugin_info.module->type != NEU_NA_TYPE_DRIVER) {
return NEU_ERR_GROUP_NOT_ALLOW;
}
for (int i = 0; i < n_tag; i++) {
int ret = neu_adapter_driver_update_tag(
(neu_adapter_driver_t *) adapter, group, &tags[i]);
if (ret == NEU_ERR_SUCCESS) {
*index += 1;
} else {
return ret;
}
}
return NEU_ERR_SUCCESS;
}
int neu_manager_get_tag(neu_manager_t *manager, const char *driver,
const char *group, UT_array **tags)
{
neu_adapter_t *adapter =
neu_node_manager_find(manager->node_manager, driver);
if (adapter == NULL) {
return NEU_ERR_NODE_NOT_EXIST;
}
if (adapter->plugin_info.module->type != NEU_NA_TYPE_DRIVER) {
return NEU_ERR_GROUP_NOT_ALLOW;
}
return neu_adapter_driver_get_tag((neu_adapter_driver_t *) adapter, group,
tags);
}
int neu_manager_subscribe(neu_manager_t *manager, const char *app,
const char *driver, const char *group)
{
@@ -309,62 +133,6 @@ UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app)
return neu_subscribe_manager_get(manager->subscribe_manager, app);
}
int neu_manager_get_node_state(neu_manager_t *manager, const char *node,
neu_node_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)
{
neu_adapter_t *adapter = neu_node_manager_find(manager->node_manager, node);
if (adapter == NULL) {
return NEU_ERR_NODE_NOT_EXIST;
}
return neu_adapter_set_setting(adapter, setting);
}
int neu_manager_node_get_setting(neu_manager_t *manager, const char *node,
char **setting)
{
neu_adapter_t *adapter = neu_node_manager_find(manager->node_manager, node);
if (adapter == NULL) {
return NEU_ERR_NODE_NOT_EXIST;
}
return neu_adapter_get_setting(adapter, setting);
}
int neu_manager_get_adapter_info(neu_manager_t *manager, const char *name,
neu_persist_adapter_info_t *info)
{
@@ -372,8 +140,8 @@ int neu_manager_get_adapter_info(neu_manager_t *manager, const char *name,
if (adapter != NULL) {
info->name = strdup(name);
info->type = adapter->plugin_info.module->type;
info->plugin_name = strdup(adapter->plugin_info.module->module_name);
info->type = adapter->module->type;
info->plugin_name = strdup(adapter->module->module_name);
info->state = adapter->state;
return 0;
}
@@ -392,8 +160,8 @@ void neu_manager_notify_app_sub(neu_manager_t *manager, const char *app,
header.type = NEU_REQ_APP_SUBSCRIBE_GROUP;
strcpy(sub.driver, driver);
strcpy(sub.group, group);
int ret = neu_manager_get_tag(manager, driver, group, &sub.tags);
assert(ret == 0);
// int ret = neu_manager_get_tag(manager, driver, group, &sub.tags);
// assert(ret == 0);
msg = neu_msg_gen(&header, (void *) &sub);
nng_msg_set_pipe(msg, pipe);
src/core/manager_internal.h
+2 -33
View File
@@ -24,6 +24,7 @@
#include <nng/supplemental/util/platform.h>
#include "event/event.h"
#include "persist/persist.h"
#include "node_manager.h"
#include "plugin_manager.h"
@@ -34,13 +35,11 @@ typedef struct neu_manager {
neu_events_t * events;
neu_event_io_t *loop;
bool stop;
neu_plugin_manager_t *plugin_manager;
neu_node_manager_t * node_manager;
neu_subscribe_mgr_t * subscribe_manager;
nng_pipe persist_pipe;
neu_persister_t *persister;
} neu_manager_t;
int neu_manager_add_plugin(neu_manager_t *manager, const char *library);
@@ -52,42 +51,12 @@ int neu_manager_add_node(neu_manager_t *manager, const char *node_name,
int neu_manager_del_node(neu_manager_t *manager, const char *node_name);
UT_array *neu_manager_get_nodes(neu_manager_t *manager, neu_node_type_e type);
int neu_manager_add_group(neu_manager_t *manager, const char *driver,
const char *group, uint32_t interval);
int neu_manager_del_group(neu_manager_t *manager, const char *driver,
const char *group);
int neu_manager_update_group(neu_manager_t *manager, const char *driver,
const char *group, uint32_t interval);
int neu_manager_get_group(neu_manager_t *manager, const char *driver,
UT_array **groups);
int neu_manager_add_tag(neu_manager_t *manager, const char *driver,
const char *group, uint16_t n_tag, neu_datatag_t *tags,
uint16_t *index);
int neu_manager_del_tag(neu_manager_t *manager, const char *driver,
const char *group, uint16_t n_tag, char **tags);
int neu_manager_update_tag(neu_manager_t *manager, const char *driver,
const char *group, uint16_t n_tag,
neu_datatag_t *tags, uint16_t *index);
int neu_manager_get_tag(neu_manager_t *manager, const char *driver,
const char *group, UT_array **tags);
int neu_manager_subscribe(neu_manager_t *manager, const char *app,
const char *driver, const char *group);
int neu_manager_unsubscribe(neu_manager_t *manager, const char *app,
const char *driver, const char *group);
UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app);
int neu_manager_get_node_state(neu_manager_t *manager, const char *node,
neu_node_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);
int neu_manager_node_get_setting(neu_manager_t *manager, const char *node,
char **setting);
int neu_manager_get_adapter_info(neu_manager_t *manager, const char *name,
neu_persist_adapter_info_t *info);
src/core/node_manager.c
+25 -6
View File
@@ -52,8 +52,6 @@ void neu_node_manager_destroy(neu_node_manager_t *mgr)
HASH_ITER(hh, mgr->nodes, el, tmp)
{
HASH_DEL(mgr->nodes, el);
neu_adapter_uninit(el->adapter);
neu_adapter_destroy(el->adapter);
free(el->name);
free(el);
}
@@ -110,6 +108,11 @@ void neu_node_manager_del(neu_node_manager_t *mgr, const char *name)
}
}
uint16_t neu_node_manager_size(neu_node_manager_t *mgr)
{
return HASH_COUNT(mgr->nodes);
}
UT_array *neu_node_manager_get(neu_node_manager_t *mgr, neu_node_type_e type)
{
UT_array * array = NULL;
@@ -121,11 +124,10 @@ UT_array *neu_node_manager_get(neu_node_manager_t *mgr, neu_node_type_e type)
HASH_ITER(hh, mgr->nodes, el, tmp)
{
if (!el->is_static) {
if (el->adapter->plugin_info.module->type == type) {
if (el->adapter->module->type == type) {
neu_resp_node_info_t info = { 0 };
strcpy(info.node, el->adapter->name);
strcpy(info.plugin,
el->adapter->plugin_info.module->module_name);
strcpy(info.plugin, el->adapter->module->module_name);
utarray_push_back(array, &info);
}
}
@@ -159,7 +161,7 @@ UT_array *neu_node_manager_get_pipes(neu_node_manager_t *mgr,
HASH_ITER(hh, mgr->nodes, el, tmp)
{
if (!el->is_static) {
if (el->adapter->plugin_info.module->type == type) {
if (el->adapter->module->type == type) {
nng_pipe pipe = el->pipe;
utarray_push_back(pipes, &pipe);
}
@@ -169,6 +171,23 @@ UT_array *neu_node_manager_get_pipes(neu_node_manager_t *mgr,
return pipes;
}
UT_array *neu_node_manager_get_pipes_all(neu_node_manager_t *mgr)
{
UT_icd icd = { sizeof(nng_pipe), NULL, NULL, NULL };
UT_array * pipes = NULL;
node_entity_t *el = NULL, *tmp = NULL;
utarray_new(pipes, &icd);
HASH_ITER(hh, mgr->nodes, el, tmp)
{
nng_pipe pipe = el->pipe;
utarray_push_back(pipes, &pipe);
}
return pipes;
}
nng_pipe neu_node_manager_get_pipe(neu_node_manager_t *mgr, const char *name)
{
nng_pipe pipe = { 0 };
src/core/node_manager.h
+9 -7
View File
@@ -32,20 +32,22 @@ typedef struct neu_node_manager neu_node_manager_t;
neu_node_manager_t *neu_node_manager_create();
void neu_node_manager_destroy(neu_node_manager_t *mgr);
int neu_node_manager_add(neu_node_manager_t *mgr, neu_adapter_t *adapter);
int neu_node_manager_add_static(neu_node_manager_t *mgr,
neu_adapter_t * adapter);
int neu_node_manager_update(neu_node_manager_t *mgr, const char *name,
nng_pipe pipe);
void neu_node_manager_del(neu_node_manager_t *mgr, const char *name);
int neu_node_manager_add(neu_node_manager_t *mgr, neu_adapter_t *adapter);
int neu_node_manager_add_static(neu_node_manager_t *mgr,
neu_adapter_t * adapter);
int neu_node_manager_update(neu_node_manager_t *mgr, const char *name,
nng_pipe pipe);
void neu_node_manager_del(neu_node_manager_t *mgr, const char *name);
uint16_t neu_node_manager_size(neu_node_manager_t *mgr);
// neu_node_info array
// neu_resp_node_info array
UT_array *neu_node_manager_get(neu_node_manager_t *mgr, neu_node_type_e type);
neu_adapter_t *neu_node_manager_find(neu_node_manager_t *mgr, const char *name);
// nng_pipe array
UT_array *neu_node_manager_get_pipes(neu_node_manager_t *mgr,
neu_node_type_e type);
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);
#endif
src/core/plugin_manager.c
-4
View File
@@ -27,7 +27,6 @@
#include "adapter.h"
#include "plugin.h"
#include "dummy/dummy.h"
#include "restful/rest.h"
#include "plugin_manager.h"
@@ -226,9 +225,6 @@ void neu_plugin_manager_load_static(neu_plugin_manager_t * mgr,
instance->module =
(neu_plugin_module_t *) &default_dashboard_plugin_module;
}
if (strcmp(DEFAULT_DUMMY_PLUGIN_NAME, plugin_name) == 0) {
instance->module = (neu_plugin_module_t *) &default_dummy_plugin_module;
}
}
void neu_plugin_manager_destroy_instance(neu_plugin_manager_t * mgr,
src/core/plugin_manager.h
-1
View File
@@ -27,7 +27,6 @@
#include "plugin.h"
#define DEFAULT_DASHBOARD_PLUGIN_NAME "default-dashboard"
#define DEFAULT_DUMMY_PLUGIN_NAME "default-dummy"
typedef struct neu_plugin_instance {
void * handle;
src/core/storage.c
+155
View File
@@ -0,0 +1,155 @@
/**
* 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 "utils/log.h"
#include "storage.h"
void manager_strorage_plugin(neu_manager_t *manager)
{
UT_array *plugin_infos = NULL;
plugin_infos = neu_manager_get_plugins(manager);
int rv = neu_persister_store_plugins(manager->persister, plugin_infos);
if (0 != rv) {
nlog_error("failed to store plugins infos");
}
neu_persist_plugin_infos_free(plugin_infos);
return;
}
void manager_storage_add_node(neu_manager_t *manager, const char *node)
{
int rv = 0;
neu_persist_adapter_info_t adapter_info = {};
rv = neu_manager_get_adapter_info(manager, node, &adapter_info);
if (0 != rv) {
nlog_error("unable to get adapter:%s info", node);
return;
}
rv = neu_persister_store_adapter(manager->persister, &adapter_info);
if (0 != rv) {
nlog_error("failed to store adapter info");
}
neu_persist_adapter_info_fini(&adapter_info);
}
void manager_storage_del_node(neu_manager_t *manager, const char *node)
{
neu_persister_delete_adapter(manager->persister, node);
}
void manager_storage_subscribe(neu_manager_t *manager, const char *app)
{
UT_array *sub_infos = neu_manager_get_sub_group(manager, app);
int rv =
neu_persister_store_subscriptions(manager->persister, app, sub_infos);
if (0 != rv) {
nlog_error("fail store adapter:%s subscription infos", app);
}
utarray_free(sub_infos);
}
int manager_load_plugin(neu_manager_t *manager)
{
UT_array *plugin_infos = NULL;
int rv = neu_persister_load_plugins(manager->persister, &plugin_infos);
if (rv != 0) {
return rv;
}
utarray_foreach(plugin_infos, char **, name)
{
rv = neu_manager_add_plugin(manager, *name);
const char *ok_or_err = (0 == rv) ? "success" : "fail";
nlog_info("load plugin %s, lib:%s", ok_or_err, *name);
}
utarray_foreach(plugin_infos, char **, name) { free(*name); }
utarray_free(plugin_infos);
return rv;
}
int manager_load_node(neu_manager_t *manager)
{
UT_array *adapter_infos = NULL;
int rv = 0;
rv = neu_persister_load_adapters(manager->persister, &adapter_infos);
if (0 != rv) {
nlog_error("failed to load adapter infos");
return -1;
}
utarray_foreach(adapter_infos, neu_persist_adapter_info_t *, adapter_info)
{
rv = neu_manager_add_node(manager, adapter_info->name,
adapter_info->plugin_name);
const char *ok_or_err = (0 == rv) ? "success" : "fail";
nlog_info("load adapter %s type:%" PRId64 ", name:%s plugin:%s",
ok_or_err, adapter_info->type, adapter_info->name,
adapter_info->plugin_name);
}
neu_persist_adapter_infos_free(adapter_infos);
return rv;
}
int manager_load_subscribe(neu_manager_t *manager)
{
UT_array *nodes =
neu_node_manager_get(manager->node_manager, NEU_NA_TYPE_APP);
utarray_foreach(nodes, neu_resp_node_info_t *, node)
{
int rv = 0;
UT_array *sub_infos = NULL;
rv = neu_persister_load_subscriptions(manager->persister, node->node,
&sub_infos);
if (0 != rv) {
nlog_warn("load %s subscribetion infos error", node->node);
} else {
utarray_foreach(sub_infos, neu_persist_subscription_info_t *, info)
{
rv = neu_manager_subscribe(manager, info->sub_adapter_name,
info->src_adapter_name,
info->group_config_name);
const char *ok_or_err = (0 == rv) ? "success" : "fail";
nlog_info("%s load subscription app:%s driver:%s grp:%s",
ok_or_err, info->sub_adapter_name,
info->src_adapter_name, info->group_config_name);
}
neu_persist_subscription_infos_free(sub_infos);
}
}
utarray_free(nodes);
return 0;
}
plugins/dummy/dummy.h → src/core/storage.h
+13 -4
View File
@@ -17,11 +17,20 @@
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#ifndef _NEU_PLUGIN_DUMMY_H_
#define _NEU_PLUGIN_DUMMY_H_
#ifndef _NEU_MANAGER_STORAGE_H_
#define _NEU_MANAGER_STORAGE_H_
#include "plugin.h"
#include "persist/persist.h"
extern const neu_plugin_module_t default_dummy_plugin_module;
#include "manager_internal.h"
void manager_strorage_plugin(neu_manager_t *manager);
void manager_storage_del_node(neu_manager_t *manager, const char *node);
void manager_storage_add_node(neu_manager_t *manager, const char *node);
void manager_storage_subscribe(neu_manager_t *manager, const char *node);
int manager_load_plugin(neu_manager_t *manager);
int manager_load_node(neu_manager_t *manager);
int manager_load_subscribe(neu_manager_t *manager);
#endif