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

feat(core): handle subscribe multiple groups

Browse Source
This commit is contained in:
eeff
2023-08-07 14:38:18 +08:00
parent 4763d1cdf9
commit d41a83bcfe
3 changed files with 59 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+26
View File
@@ -48,6 +48,7 @@ typedef enum neu_reqresp_type {
NEU_REQ_SUBSCRIBE_GROUP,
NEU_REQ_UNSUBSCRIBE_GROUP,
NEU_REQ_UPDATE_SUBSCRIBE_GROUP,
NEU_REQ_SUBSCRIBE_GROUPS,
NEU_REQ_GET_SUBSCRIBE_GROUP,
NEU_RESP_GET_SUBSCRIBE_GROUP,
NEU_REQ_GET_SUB_DRIVER_TAGS,
@@ -151,6 +152,7 @@ static const char *neu_reqresp_type_string_t[] = {
[NEU_REQ_SUBSCRIBE_GROUP] = "NEU_REQ_SUBSCRIBE_GROUP",
[NEU_REQ_UNSUBSCRIBE_GROUP] = "NEU_REQ_UNSUBSCRIBE_GROUP",
[NEU_REQ_UPDATE_SUBSCRIBE_GROUP] = "NEU_REQ_UPDATE_SUBSCRIBE_GROUP",
[NEU_REQ_SUBSCRIBE_GROUPS] = "NEU_REQ_SUBSCRIBE_GROUPS",
[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",
@@ -400,6 +402,30 @@ typedef struct {
char group[NEU_GROUP_NAME_LEN];
} neu_req_unsubscribe_t;
typedef struct {
char *driver;
char *group;
char *params;
} neu_req_subscribe_group_info_t;
typedef struct {
char * app;
uint16_t n_group;
neu_req_subscribe_group_info_t *groups;
} neu_req_subscribe_groups_t;
static inline void
neu_req_subscribe_groups_fini(neu_req_subscribe_groups_t *req)
{
for (uint16_t i = 0; i < req->n_group; ++i) {
free(req->groups[i].driver);
free(req->groups[i].group);
free(req->groups[i].params);
}
free(req->groups);
free(req->app);
}
typedef struct neu_req_get_subscribe_group {
char app[NEU_NODE_NAME_LEN];
} neu_req_get_subscribe_group_t, neu_req_get_sub_driver_tags_t;
src/adapter/adapter.c
+3
View File
@@ -1429,6 +1429,9 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data)
case NEU_REQ_UNSUBSCRIBE_GROUP:
data_size = sizeof(neu_req_unsubscribe_t);
break;
case NEU_REQ_SUBSCRIBE_GROUPS:
data_size = sizeof(neu_req_subscribe_groups_t);
break;
case NEU_REQ_GET_SUBSCRIBE_GROUP:
case NEU_REQ_GET_SUB_DRIVER_TAGS:
data_size = sizeof(neu_req_get_subscribe_group_t);
src/core/manager.c
+30
View File
@@ -701,6 +701,36 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
reply(manager, header, &error);
break;
}
case NEU_REQ_SUBSCRIBE_GROUPS: {
neu_req_subscribe_groups_t *cmd =
(neu_req_subscribe_groups_t *) &header[1];
neu_resp_error_t error = { 0 };
for (uint16_t i = 0; i < cmd->n_group; ++i) {
neu_req_subscribe_group_info_t *info = &cmd->groups[i];
error.error = neu_manager_subscribe(manager, cmd->app, info->driver,
info->group, info->params);
if (0 != error.error) {
break;
}
error.error = neu_manager_send_subscribe(
manager, cmd->app, info->driver, info->group, info->params);
if (0 != error.error) {
break;
}
manager_storage_subscribe(manager, cmd->app, info->driver,
info->group, info->params);
}
neu_req_subscribe_groups_fini(cmd);
header->type = NEU_RESP_ERROR;
strcpy(header->receiver, header->sender);
reply(manager, header, &error);
break;
}
case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: {
neu_req_subscribe_t *cmd = (neu_req_subscribe_t *) &header[1];
neu_resp_error_t error = { 0 };