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 #1283 from eeff/eeff/dev

Browse Source 
feat(core): fulfill NEU_REQ_UPDATE_NODE request
This commit is contained in:
eeff
2023-05-22 10:42:08 +08:00
committed by GitHub
parent 07a55c8281 d1cc2c23ef
commit d8e5b5967f
11 changed files with 197 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+22
View File
@@ -55,6 +55,7 @@ typedef enum neu_reqresp_type {
NEU_REQ_NODE_UNINIT,
NEU_RESP_NODE_UNINIT,
NEU_REQ_ADD_NODE,
NEU_REQ_UPDATE_NODE,
NEU_REQ_DEL_NODE,
NEU_REQ_GET_NODE,
NEU_RESP_GET_NODE,
@@ -66,6 +67,8 @@ typedef enum neu_reqresp_type {
NEU_REQ_GET_NODES_STATE,
NEU_RESP_GET_NODES_STATE,
NEU_REQ_NODE_CTL,
NEU_REQ_NODE_RENAME,
NEU_RESP_NODE_RENAME,
NEU_REQ_ADD_GROUP,
NEU_REQ_DEL_GROUP,
@@ -144,6 +147,7 @@ static const char *neu_reqresp_type_string_t[] = {
[NEU_REQ_NODE_UNINIT] = "NEU_REQ_NODE_UNINIT",
[NEU_RESP_NODE_UNINIT] = "NEU_RESP_NODE_UNINIT",
[NEU_REQ_ADD_NODE] = "NEU_REQ_ADD_NODE",
[NEU_REQ_UPDATE_NODE] = "NEU_REQ_UPDATE_NODE",
[NEU_REQ_DEL_NODE] = "NEU_REQ_DEL_NODE",
[NEU_REQ_GET_NODE] = "NEU_REQ_GET_NODE",
[NEU_RESP_GET_NODE] = "NEU_RESP_GET_NODE",
@@ -155,6 +159,8 @@ static const char *neu_reqresp_type_string_t[] = {
[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_NODE_RENAME] = "NEU_REQ_NODE_RENAME",
[NEU_RESP_NODE_RENAME] = "NEU_RESP_NODE_RENAME",
[NEU_REQ_ADD_GROUP] = "NEU_REQ_ADD_GROUP",
[NEU_REQ_DEL_GROUP] = "NEU_REQ_DEL_GROUP",
@@ -270,6 +276,11 @@ typedef struct neu_req_add_node {
char plugin[NEU_PLUGIN_NAME_LEN];
} neu_req_add_node_t;
typedef struct neu_req_update_node {
char node[NEU_NODE_NAME_LEN];
char new_name[NEU_NODE_NAME_LEN];
} neu_req_update_node_t;
typedef struct neu_req_del_node {
char node[NEU_NODE_NAME_LEN];
} neu_req_del_node_t;
@@ -427,6 +438,17 @@ typedef struct neu_req_node_ctl {
neu_adapter_ctl_e ctl;
} neu_req_node_ctl_t;
typedef struct {
char node[NEU_NODE_NAME_LEN];
char new_name[NEU_NODE_NAME_LEN];
} neu_req_node_rename_t;
typedef struct {
char node[NEU_NODE_NAME_LEN];
char new_name[NEU_NODE_NAME_LEN];
int error;
} neu_resp_node_rename_t;
typedef struct neu_req_get_node_state {
char node[NEU_NODE_NAME_LEN];
} neu_req_get_node_state_t;
src/adapter/adapter.c
+24 -2
View File
@@ -222,9 +222,9 @@ int neu_adapter_rename(neu_adapter_t *adapter, const char *new_name)
neu_metrics_del_node(adapter);
}
free(adapter->name);
adapter->name = name;
adapter->metrics->name = name;
adapter->name = name;
if (adapter->metrics) {
adapter->metrics->name = name;
neu_metrics_add_node(adapter);
}
@@ -354,6 +354,7 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
strcpy(header.receiver, cmd->driver);
break;
}
case NEU_REQ_UPDATE_NODE:
case NEU_REQ_NODE_CTL:
case NEU_REQ_GET_NODE_STATE:
case NEU_REQ_GET_NODE_SETTING:
@@ -688,6 +689,18 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
reply(adapter, header, &error);
break;
}
case NEU_REQ_NODE_RENAME: {
neu_req_node_rename_t *cmd = (neu_req_node_rename_t *) &header[1];
neu_resp_node_rename_t resp = { 0 };
resp.error = neu_adapter_rename(adapter, cmd->new_name);
strcpy(header->receiver, header->sender);
strcpy(header->sender, cmd->new_name);
strcpy(resp.node, cmd->node);
strcpy(resp.new_name, cmd->new_name);
header->type = NEU_RESP_NODE_RENAME;
reply(adapter, 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 };
@@ -1242,6 +1255,9 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data)
case NEU_REQ_ADD_NODE_EVENT:
data_size = sizeof(neu_req_add_node_t);
break;
case NEU_REQ_UPDATE_NODE:
data_size = sizeof(neu_req_update_node_t);
break;
case NEU_REQ_DEL_NODE:
case NEU_REQ_DEL_NODE_EVENT:
data_size = sizeof(neu_req_del_node_t);
@@ -1328,6 +1344,12 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data)
case NEU_REQ_NODE_CTL_EVENT:
data_size = sizeof(neu_req_node_ctl_t);
break;
case NEU_REQ_NODE_RENAME:
data_size = sizeof(neu_req_node_rename_t);
break;
case NEU_RESP_NODE_RENAME:
data_size = sizeof(neu_resp_node_rename_t);
break;
case NEU_REQ_GET_NODE_STATE:
data_size = sizeof(neu_req_get_node_state_t);
break;
src/core/manager.c
+39
View File
@@ -570,6 +570,32 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
reply(manager, header, &e);
break;
}
case NEU_REQ_UPDATE_NODE: {
neu_req_update_node_t *cmd = (neu_req_update_node_t *) &header[1];
neu_resp_error_t e = { 0 };
if (NULL == neu_node_manager_find(manager->node_manager, cmd->node)) {
e.error = NEU_ERR_NODE_NOT_EXIST;
}
if (NULL !=
neu_node_manager_find(manager->node_manager, cmd->new_name)) {
// this also makes renaming to the original name an error
e.error = NEU_ERR_NODE_EXIST;
}
if (0 == e.error) {
header->type = NEU_REQ_NODE_RENAME;
forward_msg(manager, msg, header->receiver);
} else {
header->type = NEU_RESP_ERROR;
neu_msg_exchange(header);
neu_msg_exchange(header);
reply(manager, header, &e);
}
break;
}
case NEU_REQ_DEL_NODE: {
neu_req_del_node_t *cmd = (neu_req_del_node_t *) &header[1];
neu_resp_error_t error = { 0 };
@@ -772,6 +798,19 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
break;
}
case NEU_RESP_NODE_RENAME: {
neu_resp_node_rename_t *resp = (neu_resp_node_rename_t *) &header[1];
if (0 == resp->error) {
neu_manager_update_node_name(manager, resp->node, resp->new_name);
manager_storage_update_node(manager, resp->node, resp->new_name);
}
neu_resp_error_t e = { .error = resp->error };
header->type = NEU_RESP_ERROR;
reply(manager, header, &e);
break;
}
case NEU_REQ_ADD_NODE_EVENT:
case NEU_REQ_DEL_NODE_EVENT:
case NEU_REQ_NODE_CTL_EVENT:
src/core/manager_internal.c
+18
View File
@@ -103,6 +103,24 @@ UT_array *neu_manager_get_nodes(neu_manager_t *manager, neu_node_type_e type,
return neu_node_manager_filter(manager->node_manager, type, plugin, node);
}
int neu_manager_update_node_name(neu_manager_t *manager, const char *node,
const char *new_name)
{
int ret = 0;
if (neu_node_manager_is_driver(manager->node_manager, node)) {
ret = neu_subscribe_manager_update_driver_name(
manager->subscribe_manager, node, new_name);
} else {
ret = neu_subscribe_manager_update_app_name(manager->subscribe_manager,
node, new_name);
}
if (0 == ret) {
ret =
neu_node_manager_update_name(manager->node_manager, node, new_name);
}
return ret;
}
int neu_manager_add_template(neu_manager_t *manager, const char *name,
const char *plugin, uint16_t n_group,
neu_reqresp_template_group_t *groups)
src/core/manager_internal.h
+2
View File
@@ -56,6 +56,8 @@ 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,
const char *plugin, const char *node);
int neu_manager_update_node_name(neu_manager_t *manager, const char *node,
const char *new_name);
int neu_manager_add_template(neu_manager_t *manager, const char *name,
const char *plugin, uint16_t n_group,
src/core/node_manager.c
+37 -1
View File
@@ -20,8 +20,9 @@
#include <stdlib.h>
#include <string.h>
#include "adapter/adapter_internal.h"
#include "errcodes.h"
#include "adapter/adapter_internal.h"
#include "node_manager.h"
typedef struct node_entity {
@@ -104,6 +105,29 @@ int neu_node_manager_add_single(neu_node_manager_t *mgr, neu_adapter_t *adapter,
return 0;
}
int neu_node_manager_update_name(neu_node_manager_t *mgr, const char *node_name,
const char *new_node_name)
{
node_entity_t *node = NULL;
HASH_FIND_STR(mgr->nodes, node_name, node);
if (NULL == node) {
return NEU_ERR_NODE_NOT_EXIST;
}
char *new_name = strdup(new_node_name);
if (NULL == new_name) {
return NEU_ERR_EINTERNAL;
}
HASH_DEL(mgr->nodes, node);
free(node->name);
node->name = new_name;
HASH_ADD_STR(mgr->nodes, name, node);
return 0;
}
int neu_node_manager_update(neu_node_manager_t *mgr, const char *name,
nng_pipe pipe)
{
@@ -266,6 +290,18 @@ bool neu_node_manager_is_single(neu_node_manager_t *mgr, const char *name)
return false;
}
bool neu_node_manager_is_driver(neu_node_manager_t *mgr, const char *name)
{
node_entity_t *node = NULL;
HASH_FIND_STR(mgr->nodes, name, node);
if (node != NULL) {
return NEU_NA_TYPE_DRIVER == node->adapter->module->type;
}
return false;
}
UT_array *neu_node_manager_get_pipes(neu_node_manager_t *mgr,
neu_node_type_e type)
{
src/core/node_manager.h
+3
View File
@@ -37,6 +37,8 @@ int neu_node_manager_add_static(neu_node_manager_t *mgr,
neu_adapter_t * adapter);
int neu_node_manager_add_single(neu_node_manager_t *mgr, neu_adapter_t *adapter,
bool display);
int neu_node_manager_update_name(neu_node_manager_t *mgr, const char *node_name,
const char *new_node_name);
int neu_node_manager_update(neu_node_manager_t *mgr, const char *name,
nng_pipe pipe);
bool neu_node_manager_exist_uninit(neu_node_manager_t *mgr);
@@ -54,6 +56,7 @@ UT_array * neu_node_manager_get_adapter(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);
bool neu_node_manager_is_single(neu_node_manager_t *mgr, const char *name);
bool neu_node_manager_is_driver(neu_node_manager_t *mgr, const char *name);
// nng_pipe array
UT_array *neu_node_manager_get_pipes(neu_node_manager_t *mgr,
src/core/storage.c
+7
View File
@@ -56,6 +56,13 @@ void manager_storage_add_node(neu_manager_t *manager, const char *node)
neu_persist_node_info_fini(&node_info);
}
void manager_storage_update_node(neu_manager_t *manager, const char *node,
const char *new_name)
{
(void) manager;
neu_persister_update_node(node, new_name);
}
void manager_storage_del_node(neu_manager_t *manager, const char *node)
{
(void) manager;
src/core/storage.h
+2
View File
@@ -27,6 +27,8 @@
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_update_node(neu_manager_t *manager, const char *node,
const char *new_name);
void manager_storage_subscribe(neu_manager_t *manager, const char *app,
const char *driver, const char *group,
const char *params);
src/core/subscribe.c
+36
View File
@@ -231,3 +231,39 @@ void neu_subscribe_manager_remove(neu_subscribe_mgr_t *mgr, const char *driver,
}
}
}
int neu_subscribe_manager_update_app_name(neu_subscribe_mgr_t *mgr,
const char *app, const char *new_name)
{
sub_elem_t *el = NULL, *tmp = NULL;
HASH_ITER(hh, mgr->ss, el, tmp)
{
utarray_foreach(el->apps, neu_app_subscribe_t *, sub_app)
{
if (strcmp(app, sub_app->app_name) == 0) {
strcpy(sub_app->app_name, new_name);
}
}
}
return 0;
}
int neu_subscribe_manager_update_driver_name(neu_subscribe_mgr_t *mgr,
const char * driver,
const char * new_name)
{
sub_elem_t *el = NULL, *tmp = NULL;
HASH_ITER(hh, mgr->ss, el, tmp)
{
if (strcmp(driver, el->key.driver) == 0) {
HASH_DEL(mgr->ss, el);
strcpy(el->key.driver, new_name);
HASH_ADD(hh, mgr->ss, key, sizeof(sub_elem_key_t), el);
}
}
return 0;
}
src/core/subscribe.h
+7
View File
@@ -56,4 +56,11 @@ int neu_subscribe_manager_unsub(neu_subscribe_mgr_t *mgr, const char *driver,
void neu_subscribe_manager_remove(neu_subscribe_mgr_t *mgr, const char *driver,
const char *group);
int neu_subscribe_manager_update_app_name(neu_subscribe_mgr_t *mgr,
const char * app,
const char * new_name);
int neu_subscribe_manager_update_driver_name(neu_subscribe_mgr_t *mgr,
const char * driver,
const char * new_name);
#endif