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

Browse Source 
fix(mqtt): handle group update
This commit is contained in:
fengzeroz
2023-09-01 14:26:57 +08:00
committed by GitHub
parent bb1427581d f503248f33
commit 31db617802
8 changed files with 88 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+16 -10
View File
@@ -76,7 +76,8 @@ typedef enum neu_reqresp_type {
NEU_REQ_ADD_GROUP,
NEU_REQ_DEL_GROUP,
NEU_REQ_UPDATE_GROUP,
NEU_RESP_UPDATE_GROUP,
NEU_REQ_UPDATE_DRIVER_GROUP,
NEU_RESP_UPDATE_DRIVER_GROUP,
NEU_REQ_GET_GROUP,
NEU_RESP_GET_GROUP,
NEU_REQ_GET_DRIVER_GROUP,
@@ -179,14 +180,15 @@ static const char *neu_reqresp_type_string_t[] = {
[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",
[NEU_REQ_UPDATE_GROUP] = "NEU_REQ_UPDATE_GROUP",
[NEU_RESP_UPDATE_GROUP] = "NEU_RESP_UPDATE_GROUP",
[NEU_REQ_GET_GROUP] = "NEU_REQ_GET_GROUP",
[NEU_RESP_GET_GROUP] = "NEU_RESP_GET_GROUP",
[NEU_REQ_GET_DRIVER_GROUP] = "NEU_REQ_GET_DRIVER_GROUP",
[NEU_RESP_GET_DRIVER_GROUP] = "NEU_RESP_GET_DRIVER_GROUP",
[NEU_REQ_ADD_GROUP] = "NEU_REQ_ADD_GROUP",
[NEU_REQ_DEL_GROUP] = "NEU_REQ_DEL_GROUP",
[NEU_REQ_UPDATE_GROUP] = "NEU_REQ_UPDATE_GROUP",
[NEU_REQ_UPDATE_DRIVER_GROUP] = "NEU_REQ_UPDATE_DRIVER_GROUP",
[NEU_RESP_UPDATE_DRIVER_GROUP] = "NEU_RESP_UPDATE_DRIVER_GROUP",
[NEU_REQ_GET_GROUP] = "NEU_REQ_GET_GROUP",
[NEU_RESP_GET_GROUP] = "NEU_RESP_GET_GROUP",
[NEU_REQ_GET_DRIVER_GROUP] = "NEU_REQ_GET_DRIVER_GROUP",
[NEU_RESP_GET_DRIVER_GROUP] = "NEU_RESP_GET_DRIVER_GROUP",
[NEU_REQ_ADD_TAG] = "NEU_REQ_ADD_TAG",
[NEU_RESP_ADD_TAG] = "NEU_RESP_ADD_TAG",
@@ -345,7 +347,11 @@ typedef struct {
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
char new_name[NEU_GROUP_NAME_LEN];
int error;
struct {
int error;
uint32_t unused; // padding for memory layout compatible with
// neu_req_update_group_t
};
} neu_resp_update_group_t;
typedef struct neu_req_del_group {
plugins/ekuiper/plugin_ekuiper.c
+3
View File
@@ -318,6 +318,9 @@ static int ekuiper_plugin_request(neu_plugin_t * plugin,
case NEU_REQ_UPDATE_NODE: {
break;
}
case NEU_REQ_UPDATE_GROUP: {
break;
}
case NEU_REQ_SUBSCRIBE_GROUP:
case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: {
neu_req_subscribe_t *sub_info = data;
plugins/mqtt/mqtt_handle.c
+9
View File
@@ -700,6 +700,15 @@ int handle_unsubscribe_group(neu_plugin_t * plugin,
return 0;
}
int handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req)
{
route_tbl_update_group(&plugin->route_tbl, req->driver, req->group,
req->new_name);
plog_notice(plugin, "update route driver:%s group:%s to %s", req->driver,
req->group, req->new_name);
return 0;
}
int handle_update_driver(neu_plugin_t *plugin, neu_req_update_node_t *req)
{
route_tbl_update_driver(&plugin->route_tbl, req->node, req->new_name);
plugins/mqtt/mqtt_handle.h
+2
View File
@@ -48,6 +48,8 @@ int handle_update_subscribe(neu_plugin_t * plugin,
int handle_unsubscribe_group(neu_plugin_t * plugin,
neu_req_unsubscribe_t *unsub_info);
int handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req);
int handle_update_driver(neu_plugin_t *plugin, neu_req_update_node_t *req);
int handle_del_driver(neu_plugin_t *plugin, neu_reqresp_node_deleted_t *req);
plugins/mqtt/mqtt_plugin.c
+3
View File
@@ -373,6 +373,9 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
case NEU_REQ_UNSUBSCRIBE_GROUP:
error = handle_unsubscribe_group(plugin, data);
break;
case NEU_REQ_UPDATE_GROUP:
error = handle_update_group(plugin, data);
break;
case NEU_REQ_UPDATE_NODE:
error = handle_update_driver(plugin, data);
break;
plugins/mqtt/mqtt_plugin.h
+12
View File
@@ -140,6 +140,18 @@ static inline void route_tbl_update_driver(route_entry_t **tbl,
}
}
static inline void route_tbl_update_group(route_entry_t **tbl,
const char *driver, const char *group,
const char *new_name)
{
route_entry_t *e = route_tbl_get(tbl, driver, group);
if (e) {
HASH_DEL(*tbl, e);
strcpy(e->key.group, new_name);
HASH_ADD(hh, *tbl, key, sizeof(e->key), e);
}
}
static inline void route_tbl_del_driver(route_entry_t **tbl, const char *driver)
{
route_entry_t *e = NULL, *tmp = NULL;
src/adapter/adapter.c
+5 -3
View File
@@ -501,6 +501,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_RESP_GET_NODE_STATE:
case NEU_RESP_GET_NODES_STATE:
case NEU_RESP_GET_NODE_SETTING:
case NEU_REQ_UPDATE_GROUP:
case NEU_REQ_SUBSCRIBE_GROUP:
case NEU_REQ_UPDATE_SUBSCRIBE_GROUP:
case NEU_REQ_UNSUBSCRIBE_GROUP:
@@ -701,7 +702,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
reply(adapter, header, &error);
break;
}
case NEU_REQ_UPDATE_GROUP: {
case NEU_REQ_UPDATE_DRIVER_GROUP: {
neu_req_update_group_t *cmd = (neu_req_update_group_t *) &header[1];
neu_resp_update_group_t resp = { 0 };
@@ -722,7 +723,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
strcpy(resp.driver, cmd->driver);
strcpy(resp.group, cmd->group);
strcpy(resp.new_name, cmd->new_name);
header->type = NEU_RESP_UPDATE_GROUP;
header->type = NEU_RESP_UPDATE_DRIVER_GROUP;
neu_msg_exchange(header);
reply(adapter, header, &resp);
break;
@@ -1477,10 +1478,11 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data)
data_size = sizeof(neu_req_add_group_t);
break;
case NEU_REQ_UPDATE_GROUP:
case NEU_REQ_UPDATE_DRIVER_GROUP:
case NEU_REQ_UPDATE_GROUP_EVENT:
data_size = sizeof(neu_req_update_group_t);
break;
case NEU_RESP_UPDATE_GROUP:
case NEU_RESP_UPDATE_DRIVER_GROUP:
data_size = sizeof(neu_resp_update_group_t);
break;
case NEU_REQ_DEL_GROUP:
src/core/manager.c
+38 -2
View File
@@ -868,7 +868,6 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQ_GET_TAG:
case NEU_REQ_GET_NDRIVER_TAGS:
case NEU_REQ_NODE_CTL:
case NEU_REQ_UPDATE_GROUP:
case NEU_REQ_ADD_GROUP: {
if (neu_node_manager_find(manager->node_manager, header->receiver) ==
NULL) {
@@ -911,11 +910,25 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
break;
}
case NEU_RESP_UPDATE_GROUP: {
case NEU_RESP_UPDATE_DRIVER_GROUP: {
neu_resp_update_group_t *resp = (neu_resp_update_group_t *) &header[1];
if (0 == resp->error) {
neu_manager_update_group_name(manager, resp->driver, resp->group,
resp->new_name);
UT_array *apps = neu_subscribe_manager_find(
manager->subscribe_manager, resp->driver, resp->new_name);
if (NULL != apps) {
// notify app node about group renaming
utarray_foreach(apps, neu_app_subscribe_t *, app)
{
header->type = NEU_REQ_UPDATE_GROUP;
forward_msg(manager, msg, app->app_name);
}
utarray_free(apps);
}
}
neu_resp_error_t e = { .error = resp->error };
@@ -940,6 +953,29 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
break;
}
case NEU_REQ_UPDATE_GROUP: {
neu_resp_error_t e = { 0 };
if (neu_node_manager_find(manager->node_manager, header->receiver) ==
NULL) {
e.error = NEU_ERR_NODE_NOT_EXIST;
} else if (!neu_node_manager_is_driver(manager->node_manager,
header->receiver)) {
e.error = NEU_ERR_GROUP_NOT_ALLOW;
} else {
header->type = NEU_REQ_UPDATE_DRIVER_GROUP;
forward_msg(manager, msg, header->receiver);
}
if (e.error) {
header->type = NEU_RESP_ERROR;
neu_msg_exchange(header);
reply(manager, header, &e);
}
break;
}
case NEU_REQ_DEL_GROUP: {
neu_req_del_group_t *cmd = (neu_req_del_group_t *) &header[1];