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

Merge pull request #1024 from eeff/eeff/dev

Browse Source 
fix(core): send NEU_REQ_SUBSCRIBE_GROUP when loading subscriptions
This commit is contained in:
fengzeroz
2023-01-11 10:55:47 +08:00
committed by GitHub
parent 467eff9686 a3451b7765
commit 587ba2a9f8
3 changed files with 41 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/manager_internal.c
+33
View File
@@ -172,6 +172,39 @@ int neu_manager_subscribe(neu_manager_t *manager, const char *app,
group, params, pipe);
}
int neu_manager_send_subscribe(neu_manager_t *manager, const char *app,
const char *driver, const char *group,
const char *params)
{
neu_reqresp_head_t header = { 0 };
header.type = NEU_REQ_SUBSCRIBE_GROUP;
strcpy(header.sender, "manager");
strcpy(header.receiver, app);
neu_req_subscribe_t cmd = { 0 };
strcpy(cmd.app, app);
strcpy(cmd.driver, driver);
strcpy(cmd.group, group);
if (params && NULL == (cmd.params = strdup(params))) {
return NEU_ERR_EINTERNAL;
}
nng_msg *out_msg;
nng_pipe pipe = neu_node_manager_get_pipe(manager->node_manager, app);
out_msg = neu_msg_gen(&header, &cmd);
nng_msg_set_pipe(out_msg, pipe);
if (nng_sendmsg(manager->socket, out_msg, 0) == 0) {
nlog_info("send %s to %s", neu_reqresp_type_string(header.type), app);
} else {
nng_msg_free(out_msg);
}
return 0;
}
int neu_manager_unsubscribe(neu_manager_t *manager, const char *app,
const char *driver, const char *group)
{
src/core/manager_internal.h
+3
View File
@@ -60,6 +60,9 @@ UT_array *neu_manager_get_driver_group(neu_manager_t *manager);
int neu_manager_subscribe(neu_manager_t *manager, const char *app,
const char *driver, const char *group,
const char *params);
int neu_manager_send_subscribe(neu_manager_t *manager, const char *app,
const char *driver, const char *group,
const char *params);
int neu_manager_unsubscribe(neu_manager_t *manager, const char *app,
const char *driver, const char *group);
UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app);
src/core/storage.c
+5
View File
@@ -154,6 +154,11 @@ int manager_load_subscribe(neu_manager_t *manager)
nlog_info("%s load subscription app:%s driver:%s grp:%s",
ok_or_err, node->node, info->driver_name,
info->group_name);
if (0 == rv) {
neu_manager_send_subscribe(manager, node->node,
info->driver_name,
info->group_name, info->params);
}
}
utarray_free(sub_infos);