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

Merge pull request #715 from fengzeroz/main

Browse Source 
chore(adapter): add node_deleted and get_all_tags msg
This commit is contained in:
fengzeroz
2022-08-18 16:32:02 +08:00
committed by GitHub
parent 5f4a389622 3ae3fb221e
commit a300afd28a
9 changed files with 146 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+29 -3
View File
@@ -47,6 +47,8 @@ typedef enum neu_reqresp_type {
NEU_REQ_UNSUBSCRIBE_GROUP,
NEU_REQ_GET_SUBSCRIBE_GROUP,
NEU_RESP_GET_SUBSCRIBE_GROUP,
NEU_REQ_GET_SUB_DRIVER_TAGS,
NEU_RESP_GET_SUB_DRIVER_TAGS,
NEU_REQ_APP_SUBSCRIBE_GROUP,
NEU_RESP_APP_SUBSCRIBE_GROUP,
@@ -89,6 +91,7 @@ typedef enum neu_reqresp_type {
NEU_REQRESP_TRANS_DATA,
NEU_REQRESP_NODES_STATE,
NEU_REQRESP_NODE_DELETED,
} neu_reqresp_type_e;
typedef enum {
@@ -107,6 +110,8 @@ static const char *neu_reqresp_type_string_t[] = {
[NEU_REQ_UNSUBSCRIBE_GROUP] = "NEU_REQ_UNSUBSCRIBE_GROUP",
[NEU_REQ_GET_SUBSCRIBE_GROUP] = "NEU_REQ_GET_SUBSCRIBE_GROUP",
[NEU_RESP_GET_SUBSCRIBE_GROUP] = "NEU_RESP_GET_SUBSCRIBE_GROUP",
[NEU_REQ_GET_SUB_DRIVER_TAGS] = "NEU_REQ_GET_SUB_DRIVER_TAGS",
[NEU_RESP_GET_SUB_DRIVER_TAGS] = "NEU_RESP_GET_SUB_DRIVER_TAGS",
[NEU_REQ_APP_SUBSCRIBE_GROUP] = "NEU_REQ_APP_SUBSCRIBE_GROUP",
[NEU_RESP_APP_SUBSCRIBE_GROUP] = "NEU_RESP_APP_SUBSCRIBE_GROUP",
@@ -147,8 +152,9 @@ static const char *neu_reqresp_type_string_t[] = {
[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",
[NEU_REQRESP_TRANS_DATA] = "NEU_REQRESP_TRANS_DATA",
[NEU_REQRESP_NODES_STATE] = "NEU_REQRESP_NODES_STATE",
[NEU_REQRESP_NODE_DELETED] = "NEU_REQRESP_NODE_DELETED",
};
inline static const char *neu_reqresp_type_string(neu_reqresp_type_e type)
@@ -289,7 +295,7 @@ typedef struct {
typedef struct neu_req_get_subscribe_group {
char app[NEU_NODE_NAME_LEN];
} neu_req_get_subscribe_group_t;
} neu_req_get_subscribe_group_t, neu_req_get_sub_driver_tags_t;
typedef struct neu_resp_subscribe_info {
char app[NEU_NODE_NAME_LEN];
@@ -297,6 +303,22 @@ typedef struct neu_resp_subscribe_info {
char group[NEU_GROUP_NAME_LEN];
} neu_resp_subscribe_info_t;
typedef struct {
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
UT_array *tags;
} neu_resp_get_sub_driver_tags_info_t;
inline static UT_icd *neu_resp_get_sub_driver_tags_info_icd()
{
static UT_icd icd = { sizeof(neu_resp_get_sub_driver_tags_info_t), NULL,
NULL, NULL };
return &icd;
}
typedef struct {
UT_array *infos; // array neu_resp_get_sub_driver_tags_info_t
} neu_resp_get_sub_driver_tags_t;
typedef struct neu_resp_get_subscribe_group {
UT_array *groups; // array neu_resp_subscribe_info_t
} neu_resp_get_subscribe_group_t;
@@ -384,6 +406,10 @@ typedef struct {
neu_resp_tag_value_t tags[];
} neu_reqresp_trans_data_t;
typedef struct {
char node[NEU_NODE_NAME_LEN];
} neu_reqresp_node_deleted_t;
void *neu_msg_gen(neu_reqresp_head_t *header, void *data);
inline static void neu_msg_exchange(neu_reqresp_head_t *header)
plugins/mqtt/mqtt_plugin.c
+2
View File
@@ -712,6 +712,8 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
utarray_free(states->states);
break;
}
case NEU_REQRESP_NODE_DELETED:
break;
default:
error = NEU_ERR_MQTT_FAILURE;
break;
src/adapter/adapter.c
+28
View File
@@ -157,6 +157,11 @@ void neu_adapter_init(neu_adapter_t *adapter)
nng_sendmsg(adapter->sock, init_msg, 0);
}
neu_node_type_e neu_adapter_get_type(neu_adapter_t *adapter)
{
return adapter->module->type;
}
static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
void *data)
{
@@ -171,6 +176,11 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
strcpy(header.receiver, cmd->driver);
break;
}
case NEU_REQ_DEL_NODE: {
neu_req_del_node_t *cmd = (neu_req_del_node_t *) data;
strcpy(header.receiver, cmd->node);
break;
}
case NEU_REQ_UPDATE_GROUP:
case NEU_REQ_GET_GROUP:
case NEU_REQ_DEL_GROUP:
@@ -195,6 +205,15 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
strcpy(header.receiver, cmd->node);
break;
}
case NEU_REQ_GET_SUB_DRIVER_TAGS: {
strcpy(header.receiver, "manager");
break;
}
case NEU_REQRESP_NODE_DELETED: {
neu_reqresp_node_deleted_t *cmd = (neu_reqresp_node_deleted_t *) data;
strcpy(header.receiver, cmd->node);
break;
}
default:
break;
}
@@ -248,6 +267,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
header->sender, neu_reqresp_type_string(header->type));
switch (header->type) {
case NEU_REQRESP_NODE_DELETED:
case NEU_RESP_GET_SUB_DRIVER_TAGS:
case NEU_REQ_UPDATE_LICENSE:
case NEU_REQ_APP_UNSUBSCRIBE_GROUP:
case NEU_RESP_GET_NODE_STATE:
@@ -886,8 +907,12 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data)
data_size = sizeof(neu_req_unsubscribe_t);
break;
case NEU_REQ_GET_SUBSCRIBE_GROUP:
case NEU_REQ_GET_SUB_DRIVER_TAGS:
data_size = sizeof(neu_req_get_subscribe_group_t);
break;
case NEU_RESP_GET_SUB_DRIVER_TAGS:
data_size = sizeof(neu_resp_get_sub_driver_tags_t);
break;
case NEU_RESP_GET_SUBSCRIBE_GROUP:
data_size = sizeof(neu_resp_get_subscribe_group_t);
break;
@@ -941,6 +966,9 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data)
case NEU_REQ_APP_UNSUBSCRIBE_GROUP:
data_size = sizeof(neu_req_app_unsubscribe_group_t);
break;
case NEU_REQRESP_NODE_DELETED:
data_size = sizeof(neu_reqresp_node_deleted_t);
break;
default:
assert(false);
break;
src/adapter/adapter_internal.h
+2
View File
@@ -67,6 +67,8 @@ void neu_adapter_init(neu_adapter_t *adapter);
int neu_adapter_start(neu_adapter_t *adapter);
int neu_adapter_stop(neu_adapter_t *adapter);
neu_node_type_e neu_adapter_get_type(neu_adapter_t *adapter);
int neu_adapter_uninit(neu_adapter_t *adapter);
void neu_adapter_destroy(neu_adapter_t *adapter);
src/adapter/driver/driver.c
+25
View File
@@ -553,6 +553,31 @@ int neu_adapter_driver_get_tag(neu_adapter_driver_t *driver, const char *group,
return ret;
}
void neu_adapter_driver_get_value_tag(neu_adapter_driver_t *driver,
const char *group, UT_array **tags)
{
int ret = neu_adapter_driver_get_tag(driver, group, tags);
if (ret == NEU_ERR_SUCCESS) {
utarray_foreach(*tags, neu_datatag_t *, tag)
{
if (tag->decimal != 0) {
if (tag->type == NEU_TYPE_UINT8 || tag->type == NEU_TYPE_INT8 ||
tag->type == NEU_TYPE_INT16 ||
tag->type == NEU_TYPE_UINT16 ||
tag->type == NEU_TYPE_INT32 ||
tag->type == NEU_TYPE_UINT32 ||
tag->type == NEU_TYPE_INT64 ||
tag->type == NEU_TYPE_UINT64 ||
tag->type == NEU_TYPE_FLOAT) {
tag->type = NEU_TYPE_DOUBLE;
}
}
}
} else {
utarray_new(*tags, neu_tag_get_icd());
}
}
UT_array *neu_adapter_driver_get_read_tag(neu_adapter_driver_t *driver,
const char * group)
{
src/adapter/driver/driver_internal.h
+10 -8
View File
@@ -46,14 +46,16 @@ int neu_adapter_driver_group_exist(neu_adapter_driver_t *driver,
const char * name);
UT_array *neu_adapter_driver_get_group(neu_adapter_driver_t *driver);
int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
neu_datatag_t *tag);
int neu_adapter_driver_del_tag(neu_adapter_driver_t *driver, const char *group,
const char *tag);
int neu_adapter_driver_update_tag(neu_adapter_driver_t *driver,
const char *group, neu_datatag_t *tag);
int neu_adapter_driver_get_tag(neu_adapter_driver_t *driver, const char *group,
UT_array **tags);
int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
neu_datatag_t *tag);
int neu_adapter_driver_del_tag(neu_adapter_driver_t *driver, const char *group,
const char *tag);
int neu_adapter_driver_update_tag(neu_adapter_driver_t *driver,
const char *group, neu_datatag_t *tag);
int neu_adapter_driver_get_tag(neu_adapter_driver_t *driver, const char *group,
UT_array **tags);
void neu_adapter_driver_get_value_tag(neu_adapter_driver_t *driver,
const char *group, UT_array **tags);
UT_array *neu_adapter_driver_get_read_tag(neu_adapter_driver_t *driver,
const char * group);
#endif
src/core/manager.c
+48 -2
View File
@@ -273,13 +273,30 @@ 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];
neu_resp_error_t error = { 0 };
neu_adapter_t * adapter =
neu_node_manager_find(manager->node_manager, cmd->node);
strcpy(header->receiver, cmd->node);
if (neu_node_manager_find(manager->node_manager, header->receiver) !=
NULL) {
if (adapter != NULL) {
manager_storage_del_node(manager, cmd->node);
header->type = NEU_REQ_NODE_UNINIT;
forward_msg(manager, msg, header->receiver);
if (neu_adapter_get_type(adapter) == NEU_NA_TYPE_DRIVER) {
UT_array *apps = neu_subscribe_manager_find_by_driver(
manager->subscribe_manager, cmd->node);
utarray_foreach(apps, neu_app_subscribe_t *, app)
{
neu_reqresp_node_deleted_t resp = { 0 };
header->type = NEU_REQRESP_NODE_DELETED;
strcpy(resp.node, header->receiver);
strcpy(header->receiver, app->app_name);
strcpy(header->sender, "manager");
reply(manager, header, &resp);
}
utarray_free(apps);
}
} else {
error.error = NEU_ERR_NODE_NOT_EXIST;
header->type = NEU_RESP_ERROR;
@@ -359,6 +376,35 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
reply(manager, header, &resp);
break;
}
case NEU_REQ_GET_SUB_DRIVER_TAGS: {
neu_req_get_sub_driver_tags_t *cmd =
(neu_req_get_sub_driver_tags_t *) &header[1];
neu_resp_get_sub_driver_tags_t resp = { 0 };
UT_array *groups = neu_manager_get_sub_group(manager, cmd->app);
utarray_new(resp.infos, neu_resp_get_sub_driver_tags_info_icd());
utarray_foreach(groups, neu_resp_subscribe_info_t *, info)
{
neu_resp_get_sub_driver_tags_info_t in = { 0 };
neu_adapter_t * driver =
neu_node_manager_find(manager->node_manager, info->driver);
assert(driver != NULL);
strcpy(in.driver, info->driver);
strcpy(in.group, info->group);
neu_adapter_driver_get_value_tag((neu_adapter_driver_t *) driver,
info->group, &in.tags);
utarray_push_back(resp.infos, &in);
}
utarray_free(groups);
strcpy(header->receiver, header->sender);
header->type = NEU_RESP_GET_SUB_DRIVER_TAGS;
reply(manager, header, &resp);
break;
}
case NEU_REQ_GET_NODES_STATE: {
neu_resp_get_nodes_state_t resp = { 0 };
UT_array *states = neu_node_manager_get_state(manager->node_manager);
src/core/plugin_manager.c
+1
View File
@@ -77,6 +77,7 @@ int neu_plugin_manager_add(neu_plugin_manager_t *mgr,
neu_plugin_module_t *pm = NULL;
plugin_entity_t * plugin = NULL;
assert(strlen(plugin_lib_name) <= NEU_PLUGIN_LIBRARY_LEN);
snprintf(lib_path, sizeof(lib_path) - 1, "./plugins/%s", plugin_lib_name);
handle = dlopen(lib_path, RTLD_NOW | RTLD_NODELETE);
zlog.conf
+1 -1
View File
@@ -9,4 +9,4 @@ simple = "%d [%V] %f:%L %m%n"
[rules]
*.INFO "./logs/%c.log", 10MB * 3 ~ "./logs/%c-%d(%Y%m%d).#2s.log"; simple
*.INFO "./logs/%c.log", 10MB * 3 ~ "./logs/%c.#2s.log"; simple