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

Merge pull request #561 from fengzeroz/main

Browse Source 
add NEU_RESP_APP_SUBSCRIBE_GROUP msg nofity app subscribe group
This commit is contained in:
fengzeroz
2022-06-20 18:02:24 +08:00
committed by GitHub
parent ed459a0398 566487d1f8
commit 45eabf84da
11 changed files with 99 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/check.yml
+1 -1
View File
@@ -72,7 +72,7 @@ jobs:
mosquitto -v &
python3 -m pip install -U pip
python3 -m pip install -r ft/requirements.txt
python3 -m robot -P ft/ -d ft/reports ft
python3 -m robot --maxerrorlines=600 -P ft/ -d ft/reports ft
- uses: actions/upload-artifact@v3
with:
include/neuron/adapter.h
+6 -4
View File
@@ -52,6 +52,7 @@ typedef enum neu_reqresp_type {
NEU_RESP_GET_SUBSCRIBE_GROUP,
NEU_REQ_APP_SUBSCRIBE_GROUP,
NEU_RESP_APP_SUBSCRIBE_GROUP,
NEU_REQ_APP_UNSUBSCRIBE_GROUP,
NEU_REQ_NODE_INIT,
@@ -206,12 +207,12 @@ typedef struct {
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
UT_array *tags; // array neu_datatag_t
} neu_req_app_subscribe_group_t;
} neu_resp_app_subscribe_group_t;
typedef struct {
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
} neu_req_app_unsubscribe_group_t;
} neu_req_app_subscribe_group_t, neu_req_app_unsubscribe_group_t;
typedef struct {
char app[NEU_NODE_NAME_LEN];
@@ -419,9 +420,10 @@ inline static nng_msg *neu_msg_gen(neu_reqresp_head_t *header, void *data)
case NEU_REQ_UPDATE_LICENSE:
data_size = sizeof(neu_req_update_license_t);
break;
case NEU_REQ_APP_SUBSCRIBE_GROUP:
data_size = sizeof(neu_req_app_subscribe_group_t);
case NEU_RESP_APP_SUBSCRIBE_GROUP:
data_size = sizeof(neu_resp_app_subscribe_group_t);
break;
case NEU_REQ_APP_SUBSCRIBE_GROUP:
case NEU_REQ_APP_UNSUBSCRIBE_GROUP:
data_size = sizeof(neu_req_app_unsubscribe_group_t);
break;
plugins/ekuiper/plugin_ekuiper.c
+3 -3
View File
@@ -170,9 +170,9 @@ static int ekuiper_plugin_request(neu_plugin_t * plugin,
send_data(plugin, trans_data);
break;
}
case NEU_REQ_APP_SUBSCRIBE_GROUP: {
neu_req_app_subscribe_group_t *sub =
(neu_req_app_subscribe_group_t *) data;
case NEU_RESP_APP_SUBSCRIBE_GROUP: {
neu_resp_app_subscribe_group_t *sub =
(neu_resp_app_subscribe_group_t *) data;
utarray_free(sub->tags);
break;
plugins/mqtt/mqtt.c
+4 -4
View File
@@ -518,11 +518,11 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
case NEU_REQRESP_TRANS_DATA:
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;
case NEU_RESP_APP_SUBSCRIBE_GROUP: {
neu_resp_app_subscribe_group_t *sub =
(neu_resp_app_subscribe_group_t *) data;
// utarray_free(sub->tags);
utarray_free(sub->tags);
break;
}
default:
src/adapter/adapter.c
+19 -2
View File
@@ -226,7 +226,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
header = (neu_reqresp_head_t *) nng_msg_body(msg);
switch (header->type) {
case NEU_REQ_APP_SUBSCRIBE_GROUP:
case NEU_REQ_APP_UNSUBSCRIBE_GROUP:
case NEU_RESP_GET_NODE_STATE:
case NEU_RESP_GET_NODE_SETTING:
@@ -240,6 +239,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_RESP_GET_GROUP:
case NEU_RESP_ERROR:
case NEU_REQRESP_TRANS_DATA:
case NEU_RESP_APP_SUBSCRIBE_GROUP:
adapter->module->intf_funs->request(
adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
break;
@@ -267,8 +267,10 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
neu_msg_exchange(header);
reply(adapter, header, &error);
} else {
void *msg_dump = calloc(nng_msg_len(msg), 1);
memcpy(msg_dump, header, nng_msg_len(msg));
neu_adapter_driver_write_tag((neu_adapter_driver_t *) adapter,
header);
msg_dump);
}
break;
@@ -332,6 +334,21 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
break;
}
case NEU_REQ_APP_SUBSCRIBE_GROUP: {
neu_req_app_subscribe_group_t *cmd =
(neu_req_app_subscribe_group_t *) &header[1];
neu_resp_app_subscribe_group_t resp = { 0 };
neu_adapter_driver_get_tag((neu_adapter_driver_t *) adapter, cmd->group,
&resp.tags);
neu_msg_exchange(header);
strcpy(resp.driver, cmd->driver);
strcpy(resp.group, cmd->group);
header->type = NEU_RESP_APP_SUBSCRIBE_GROUP;
reply(adapter, header, &resp);
break;
}
case NEU_REQ_GET_TAG: {
neu_req_get_tag_t *cmd = (neu_req_get_tag_t *) &header[1];
neu_resp_error_t error = { .error = 0 };
src/adapter/driver/driver.c
+1
View File
@@ -76,6 +76,7 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
req->type = NEU_RESP_ERROR;
adapter->cb_funs.response(adapter, req, &nerror);
free(req);
}
static void update(neu_adapter_t *adapter, const char *group, const char *tag,
src/core/manager.c
+5
View File
@@ -359,6 +359,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
reply(manager, header, &e);
} else {
forward_msg(manager, msg, header->receiver);
neu_manager_notify_app_unsub_update(manager, cmd->driver,
cmd->group);
neu_subscribe_manager_remove(manager->subscribe_manager,
cmd->driver, cmd->group);
}
@@ -379,6 +381,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
free(cmd->tags);
} else {
forward_msg(manager, msg, header->receiver);
neu_manager_notify_app_sub_update(manager, cmd->driver, cmd->group);
}
break;
@@ -401,6 +404,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
free(cmd->tags);
} else {
forward_msg(manager, msg, header->receiver);
neu_manager_notify_app_sub_update(manager, cmd->driver, cmd->group);
}
break;
@@ -429,6 +433,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_RESP_GET_NODE_STATE:
case NEU_RESP_ERROR:
case NEU_RESP_READ_GROUP:
case NEU_RESP_APP_SUBSCRIBE_GROUP:
forward_msg(manager, msg, header->receiver);
break;
default:
src/core/manager_internal.c
+35 -3
View File
@@ -155,13 +155,14 @@ void neu_manager_notify_app_sub(neu_manager_t *manager, const char *app,
nng_msg * msg;
neu_reqresp_head_t header = { 0 };
neu_req_app_subscribe_group_t sub = { 0 };
nng_pipe pipe = neu_node_manager_get_pipe(manager->node_manager, app);
nng_pipe pipe = neu_node_manager_get_pipe(manager->node_manager, driver);
header.type = NEU_REQ_APP_SUBSCRIBE_GROUP;
strcpy(header.sender, app);
strcpy(header.receiver, driver);
strcpy(sub.driver, driver);
strcpy(sub.group, group);
// 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);
@@ -177,6 +178,9 @@ void neu_manager_notify_app_unsub(neu_manager_t *manager, const char *app,
nng_pipe pipe = neu_node_manager_get_pipe(manager->node_manager, app);
header.type = NEU_REQ_APP_UNSUBSCRIBE_GROUP;
strcpy(header.sender, "manager");
strcpy(header.receiver, app);
strcpy(unsub.driver, driver);
strcpy(unsub.group, group);
@@ -185,6 +189,34 @@ void neu_manager_notify_app_unsub(neu_manager_t *manager, const char *app,
nng_sendmsg(manager->socket, msg, 0);
}
void neu_manager_notify_app_unsub_update(neu_manager_t *manager,
const char *driver, const char *group)
{
UT_array *apps =
neu_subscribe_manager_find(manager->subscribe_manager, driver, group);
if (apps != NULL) {
utarray_foreach(apps, neu_app_subscribe_t *, app)
{
neu_manager_notify_app_unsub(manager, app->app_name, driver, group);
}
utarray_free(apps);
}
}
void neu_manager_notify_app_unsub_driver_update(neu_manager_t *manager,
const char * driver)
{
UT_array *apps = neu_subscribe_manager_find_by_driver(
manager->subscribe_manager, driver);
if (apps != NULL) {
utarray_foreach(apps, neu_app_subscribe_t *, app)
{
neu_manager_notify_app_unsub(manager, app->app_name, driver, "");
}
utarray_free(apps);
}
}
void neu_manager_notify_app_sub_update(neu_manager_t *manager,
const char *driver, const char *group)
{
src/core/manager_internal.h
+4
View File
@@ -64,6 +64,10 @@ void neu_manager_notify_app_sub(neu_manager_t *manager, const char *app,
const char *driver, const char *group);
void neu_manager_notify_app_unsub(neu_manager_t *manager, const char *app,
const char *driver, const char *group);
void neu_manager_notify_app_unsub_update(neu_manager_t *manager,
const char *driver, const char *group);
void neu_manager_notify_app_unsub_driver_update(neu_manager_t *manager,
const char * driver);
void neu_manager_notify_app_sub_update(neu_manager_t *manager,
const char *driver, const char *group);
src/core/subscribe.c
+18 -1
View File
@@ -91,6 +91,24 @@ UT_array *neu_subscribe_manager_find(neu_subscribe_mgr_t *mgr,
}
}
UT_array *neu_subscribe_manager_find_by_driver(neu_subscribe_mgr_t *mgr,
const char * driver)
{
sub_elem_t *el = NULL, *tmp = NULL;
UT_array * apps = NULL;
utarray_new(apps, &app_sub_icd);
HASH_ITER(hh, mgr->ss, el, tmp)
{
if (strcmp(driver, el->key.driver) == 0) {
utarray_concat(apps, el->apps);
}
}
return apps;
}
UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app)
{
sub_elem_t *el = NULL, *tmp = NULL;
@@ -98,7 +116,6 @@ UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app)
UT_icd icd = { sizeof(neu_resp_subscribe_info_t), NULL, NULL, NULL };
utarray_new(groups, &icd);
HASH_ITER(hh, mgr->ss, el, tmp)
{
utarray_foreach(el->apps, neu_app_subscribe_t *, sub_app)
src/core/subscribe.h
+3 -1
View File
@@ -28,7 +28,7 @@
typedef struct neu_subscribe_mgr neu_subscribe_mgr_t;
typedef struct neu_app_subscribe {
char app_name[128];
char app_name[NEU_NODE_NAME_LEN];
nng_pipe pipe;
} neu_app_subscribe_t;
@@ -38,6 +38,8 @@ void neu_subscribe_manager_destroy(neu_subscribe_mgr_t *mgr);
// neu_app_subscribe_t array
UT_array *neu_subscribe_manager_find(neu_subscribe_mgr_t *mgr,
const char *driver, const char *group);
UT_array *neu_subscribe_manager_find_by_driver(neu_subscribe_mgr_t *mgr,
const char * driver);
UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app);
int neu_subscribe_manager_sub(neu_subscribe_mgr_t *mgr, const char *driver,
const char *app, const char *group,