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

subscribe/unsubscribe/get subscribe

Browse Source
This commit is contained in:
fengzero
2022-06-02 07:45:33 +00:00
parent c8234633e0
commit 8e8fa443c2
14 changed files with 277 additions and 303 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter_msg.h
+3 -3
View File
@@ -110,8 +110,9 @@ typedef struct neu_group_info {
} neu_group_info_t;
typedef struct neu_subscribe_info {
char node_name[NEU_NODE_NAME_LEN];
char group_name[NEU_GROUP_NAME_LEN];
char app[NEU_NODE_NAME_LEN];
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
} neu_subscribe_info_t;
/* NEU_REQRESP_READ_DATA */
@@ -306,7 +307,6 @@ typedef struct neu_cmd_get_sub_grp_configs {
/* NEU_REQRESP_SUB_GRP_CONFIGS_RESP */
typedef struct neu_reqresp_sub_grp_configs {
int32_t result;
UT_array *groups;
} neu_reqresp_sub_grp_configs_t;
include/neuron/plugin.h
+4 -8
View File
@@ -116,14 +116,10 @@ intptr_t neu_system_update_group_config(neu_plugin_t *plugin,
int neu_system_get_group_configs(neu_plugin_t *plugin, const char *node_name,
UT_array **groups);
uint32_t neu_plugin_send_unsubscribe_cmd(neu_plugin_t *plugin,
const char * app_name,
const char * driver_name,
const char * group);
uint32_t neu_plugin_send_subscribe_cmd(neu_plugin_t *plugin,
const char * app_name,
const char * driver_name,
const char * group);
int neu_plugin_send_unsubscribe_cmd(neu_plugin_t *plugin, const char *app_name,
const char *driver_name, const char *group);
int neu_plugin_send_subscribe_cmd(neu_plugin_t *plugin, const char *app_name,
const char *driver_name, const char *group);
void neu_plugin_send_read_cmd(neu_plugin_t *plugin, uint32_t event_id,
const char *node_name, const char *group_name);
plugins/restful/group_config_handle.c
+37 -41
View File
@@ -126,13 +126,11 @@ void handle_grp_subscribe(nng_aio *aio)
neu_plugin_t *plugin = neu_rest_get_plugin();
REST_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
aio, neu_json_subscribe_req_t, neu_json_decode_subscribe_req, {
aio, neu_json_subscribe_req_t, neu_json_decode_subscribe_req,
{ NEU_JSON_RESPONSE_ERROR(
neu_plugin_send_subscribe_cmd(plugin, req->app_name,
req->driver_name, req->name);
NEU_JSON_RESPONSE_ERROR(NEU_ERR_SUCCESS, {
http_response(aio, error_code.error, result_error);
})
})
req->driver_name, req->name),
{ http_response(aio, error_code.error, result_error); }) })
}
void handle_grp_unsubscribe(nng_aio *aio)
@@ -140,52 +138,50 @@ void handle_grp_unsubscribe(nng_aio *aio)
neu_plugin_t *plugin = neu_rest_get_plugin();
REST_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
aio, neu_json_unsubscribe_req_t, neu_json_decode_unsubscribe_req, {
aio, neu_json_unsubscribe_req_t, neu_json_decode_unsubscribe_req,
{ NEU_JSON_RESPONSE_ERROR(
neu_plugin_send_unsubscribe_cmd(plugin, req->app_name,
req->driver_name, req->name);
NEU_JSON_RESPONSE_ERROR(NEU_ERR_SUCCESS, {
http_response(aio, error_code.error, result_error);
})
})
req->driver_name, req->name),
{ http_response(aio, error_code.error, result_error); }) })
}
void handle_grp_get_subscribe(nng_aio *aio)
{
(void) aio;
// neu_plugin_t * plugin = neu_rest_get_plugin();
// char * result = NULL;
// neu_json_get_subscribe_resp_t sub_grp_configs = { 0 };
// int index = 0;
// char node_name[NEU_NODE_NAME_LEN] = { 0 };
neu_plugin_t * plugin = neu_rest_get_plugin();
char * result = NULL;
neu_json_get_subscribe_resp_t sub_grp_configs = { 0 };
int index = 0;
char node_name[NEU_NODE_NAME_LEN] = { 0 };
// VALIDATE_JWT(aio);
VALIDATE_JWT(aio);
// if (http_get_param_str(aio, "node_name", sizeof(node_name)) != 0) {
// NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
// http_response(aio, error_code.error, result_error);
//})
// return;
//}
if (http_get_param_str(aio, "node_name", node_name, sizeof(node_name)) <=
0) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
http_response(aio, error_code.error, result_error);
})
return;
}
// UT_array *groups = neu_system_get_sub_group_configs(plugin, node_name);
UT_array *groups = neu_system_get_sub_group_configs(plugin, node_name);
// sub_grp_configs.n_group = utarray_len(groups);
// sub_grp_configs.groups = calloc(
// sub_grp_configs.n_group, sizeof(neu_json_get_subscribe_resp_group_t));
sub_grp_configs.n_group = utarray_len(groups);
sub_grp_configs.groups = calloc(
sub_grp_configs.n_group, sizeof(neu_json_get_subscribe_resp_group_t));
// utarray_foreach(groups, neu_subscribe_info_t *, group)
//{
// sub_grp_configs.groups[index].node_name = group->node_name;
// sub_grp_configs.groups[index].group_name = group->group_name;
utarray_foreach(groups, neu_subscribe_info_t *, group)
{
sub_grp_configs.groups[index].node_name = group->driver;
sub_grp_configs.groups[index].group_name = group->group;
// index += 1;
//}
index += 1;
}
// neu_json_encode_by_fn(&sub_grp_configs,
// neu_json_encode_get_subscribe_resp, &result);
neu_json_encode_by_fn(&sub_grp_configs, neu_json_encode_get_subscribe_resp,
&result);
// http_ok(aio, result);
// free(result);
// free(sub_grp_configs.groups);
// utarray_free(groups);
http_ok(aio, result);
free(result);
free(sub_grp_configs.groups);
utarray_free(groups);
}
src/adapter/adapter.c
+84 -177
View File
@@ -326,18 +326,7 @@ static int persister_singleton_load_subscriptions(neu_adapter_t *adapter,
{
int rv = 0;
neu_persister_t *persister = persister_singleton_get();
vector_t * sub_infos = NULL;
neu_node_id_t dst_node_id = 0;
(void) dst_node_id;
// rv = neu_manager_get_node_id_by_name(adapter->manager, adapter_name,
//&dst_node_id);
if (0 != rv) {
nlog_error("%s fail get node id by name: %s", adapter->name,
adapter_name);
return rv;
}
UT_array * sub_infos = NULL;
rv = neu_persister_load_subscriptions(persister, adapter_name, &sub_infos);
if (0 != rv) {
@@ -352,42 +341,15 @@ static int persister_singleton_load_subscriptions(neu_adapter_t *adapter,
return rv;
}
VECTOR_FOR_EACH(sub_infos, iter)
utarray_foreach(sub_infos, neu_persist_subscription_info_t *, info)
{
neu_persist_subscription_info_t *p = iterator_get(&iter);
neu_node_id_t src_node_id = 0;
(void) src_node_id;
// rv = neu_manager_get_node_id_by_name(adapter->manager,
// p->src_adapter_name, &src_node_id);
if (0 != rv) {
nlog_error("%s fail get node id by name: %s", adapter->name,
p->src_adapter_name);
break;
}
neu_taggrp_config_t *grp_config = NULL;
(void) grp_config;
// rv = neu_manager_adapter_get_grp_config_ref_by_name(
// adapter->manager, src_node_id, p->group_config_name, &grp_config);
// if (0 != rv) {
// break;
//}
// subscribe_node_cmd_t cmd = {
//.grp_config = grp_config,
//.dst_node_id = dst_node_id,
//.src_node_id = src_node_id,
//.sender_id = adapter->id,
//// .req_id = adapter_get_req_id(adapter),
//};
// rv = neu_manager_subscribe_node(adapter->manager, &cmd);
rv = neu_manager_subscribe(adapter->manager, info->sub_adapter_name,
info->src_adapter_name,
info->group_config_name);
const char *ok_or_err = (0 == rv) ? "success" : "fail";
nlog_info("%s %s load subscription dst:%s src:%s grp:%s", adapter->name,
ok_or_err, adapter_name, p->src_adapter_name,
p->group_config_name);
neu_taggrp_cfg_free((neu_taggrp_config_t *) grp_config);
nlog_info("%s %s load subscription app:%s driver:%s grp:%s",
adapter->name, ok_or_err, adapter_name,
info->src_adapter_name, info->group_config_name);
}
neu_persist_subscription_infos_free(sub_infos);
@@ -442,17 +404,14 @@ static int persister_singleton_load_data(neu_adapter_t *adapter)
}
rv = persister_singleton_load_grp_and_tags(adapter, adapter_info->name);
// rv = persister_singleton_load_setting(adapter, adapter_info->name,
// node_id);
if (0 != rv) {
continue;
}
if (ADAPTER_STATE_RUNNING == adapter_info->state) {
// rv = neu_manager_start_adapter_with_id(adapter->manager,
// node_id); if (0 != rv) { nlog_error("%s fail start adapter %s",
// adapter->name, adapter_info->name);
// node_id); if (0 != rv) { nlog_error("%s fail start adapter
// %s", adapter->name, adapter_info->name);
//}
}
}
@@ -631,36 +590,18 @@ static int persister_singleton_handle_datatags(neu_adapter_t *adapter,
}
static int persister_singleton_handle_subscriptions(neu_adapter_t *adapter,
neu_node_id_t node_id)
const char * app)
{
neu_persister_t *persister = persister_singleton_get();
char * adapter_name = NULL;
int rv = 0;
(void) node_id;
(void) adapter;
(void) persister;
// int rv = neu_manager_get_node_name_by_id(adapter->manager, node_id,
//&adapter_name);
neu_persister_t *persister = persister_singleton_get();
UT_array *sub_infos = neu_manager_get_sub_group(adapter->manager, app);
int rv = neu_persister_store_subscriptions(persister, app, sub_infos);
if (0 != rv) {
return rv;
nlog_error("%s fail store adapter:%s subscription infos", app, app);
}
// vector_t *sub_infos = NULL;
// rv = neu_manager_get_persist_subscription_infos(adapter->manager,
// node_id, &sub_infos);
// if (0 != rv) {
// nlog_error("%s fail get persist subscription infos", adapter->name);
// free(adapter_name);
// return rv;
//}
// rv = neu_persister_store_subscriptions(persister, adapter_name,
// sub_infos); if (0 != rv) { nlog_error("%s fail store adapter:%s
// subscription infos", adapter_name, adapter->name);
//}
// neu_persist_subscription_infos_free(sub_infos);
free(adapter_name);
utarray_free(sub_infos);
return rv;
}
@@ -721,24 +662,6 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd,
nlog_debug("Get command from plugin %d, %s", cmd->req_type, adapter->name);
switch (cmd->req_type) {
case NEU_REQRESP_SUBSCRIBE_NODE: {
ADAPTER_RESP_CODE(adapter, cmd, intptr_t, subscribe_node_cmd_t, rv,
NEU_REQRESP_ERR_CODE, p_result, {
ret = neu_manager_subscribe(
adapter->manager, req_cmd->app,
req_cmd->driver, req_cmd->group);
});
break;
}
case NEU_REQRESP_UNSUBSCRIBE_NODE: {
ADAPTER_RESP_CODE(adapter, cmd, intptr_t, unsubscribe_node_cmd_t, rv,
NEU_REQRESP_ERR_CODE, p_result, {
ret = neu_manager_unsubscribe(
adapter->manager, req_cmd->app,
req_cmd->driver, req_cmd->group);
});
break;
}
// case NEU_REQRESP_READ_DATA: {
// ADAPTER_SEND_MSG(adapter, cmd, rv, MSG_CMD_READ_DATA,
// read_data_cmd_t, neu_reqresp_read_t, {});
@@ -747,7 +670,8 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd,
// case NEU_REQRESP_WRITE_DATA: {
// ADAPTER_SEND_MSG(adapter, cmd, rv, MSG_CMD_WRITE_DATA,
// write_data_cmd_t, neu_reqresp_write_t, { neu_trans_buf_t *trans_buf;
// write_data_cmd_t, neu_reqresp_write_t, { neu_trans_buf_t
// *trans_buf;
// trans_buf = &cmd_ptr->trans_buf;
// neu_trans_buf_init(trans_buf, adapter->trans_kind,
@@ -755,6 +679,40 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd,
//})
// break;
//}
case NEU_REQRESP_SUBSCRIBE_NODE: {
ADAPTER_RESP_CODE(
adapter, cmd, intptr_t, subscribe_node_cmd_t, rv,
NEU_REQRESP_ERR_CODE, p_result, {
ret = neu_manager_subscribe(adapter->manager, req_cmd->app,
req_cmd->driver, req_cmd->group);
if (ret == 0) {
subscribe_node_cmd_t event = { 0 };
event.app = strdup(req_cmd->app);
event.driver = strdup(req_cmd->driver);
event.group = strdup(req_cmd->group);
ADAPTER_SEND_BUF(adapter, rv, MSG_EVENT_SUBSCRIBE_NODE,
&event, sizeof(event));
}
});
break;
}
case NEU_REQRESP_UNSUBSCRIBE_NODE: {
ADAPTER_RESP_CODE(
adapter, cmd, intptr_t, unsubscribe_node_cmd_t, rv,
NEU_REQRESP_ERR_CODE, p_result, {
ret = neu_manager_unsubscribe(adapter->manager, req_cmd->app,
req_cmd->driver, req_cmd->group);
if (ret == 0) {
unsubscribe_node_cmd_t event = { 0 };
event.app = strdup(req_cmd->app);
event.driver = strdup(req_cmd->driver);
event.group = strdup(req_cmd->group);
ADAPTER_SEND_BUF(adapter, rv, MSG_EVENT_UNSUBSCRIBE_NODE,
&event, sizeof(event));
}
});
break;
}
case NEU_REQRESP_ADD_NODE: {
ADAPTER_RESP_CODE(adapter, cmd, intptr_t, neu_cmd_add_node_t, rv,
@@ -933,7 +891,8 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd,
ret =
calloc(1, sizeof(neu_reqresp_node_setting_t));
// ret->result = neu_manager_adapter_get_setting(
// ret->result =
// neu_manager_adapter_get_setting(
// adapter->manager, req_cmd->node_id,
// &ret->setting);
});
@@ -958,7 +917,8 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd,
ADAPTER_RESP_CODE(adapter, cmd, intptr_t, neu_cmd_node_ctl_t, rv,
NEU_REQRESP_ERR_CODE, p_result, {
ret = 0;
// ret = neu_manager_adapter_ctl(adapter->manager,
// ret =
// neu_manager_adapter_ctl(adapter->manager,
// req_cmd->node_id, req_cmd->ctl);
// if (0 == ret) {
// char *node_name = NULL;
@@ -976,17 +936,14 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd,
}
case NEU_REQRESP_GET_SUB_GRP_CONFIGS: {
ADAPTER_RESP_CMD(adapter, cmd, neu_reqresp_sub_grp_configs_t,
neu_cmd_get_sub_grp_configs_t, rv,
NEU_REQRESP_SUB_GRP_CONFIGS_RESP, p_result, {
ret = calloc(
1, sizeof(neu_reqresp_sub_grp_configs_t));
// ret->result =
// neu_manager_adapter_get_sub_grp_configs(
// adapter->manager, req_cmd->node_id,
// &ret->sub_grp_configs);
});
ADAPTER_RESP_CMD(
adapter, cmd, neu_reqresp_sub_grp_configs_t,
neu_cmd_get_sub_grp_configs_t, rv, NEU_REQRESP_SUB_GRP_CONFIGS_RESP,
p_result, {
ret = calloc(1, sizeof(neu_reqresp_sub_grp_configs_t));
ret->groups = neu_manager_get_sub_group(adapter->manager,
req_cmd->node_name);
});
break;
}
@@ -1312,6 +1269,16 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
persister_singleton_load_data(adapter);
break;
}
case MSG_EVENT_SUBSCRIBE_NODE:
case MSG_EVENT_UNSUBSCRIBE_NODE: {
subscribe_node_cmd_t *cmd = msg_get_buf_ptr(pay_msg);
persister_singleton_handle_subscriptions(adapter, cmd->app);
free(cmd->app);
free(cmd->driver);
free(cmd->group);
break;
}
case MSG_EVENT_ADD_NODE:
// fall through
@@ -1377,15 +1344,6 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
break;
}
case MSG_EVENT_SUBSCRIBE_NODE:
// fall through
case MSG_EVENT_UNSUBSCRIBE_NODE: {
const neu_node_id_t *node_id_p = msg_get_buf_ptr(pay_msg);
persister_singleton_handle_subscriptions(adapter, *node_id_p);
break;
}
case MSG_EVENT_ADD_TAGS:
// fall through
@@ -1426,60 +1384,6 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
break;
}
// case MSG_CMD_SUBSCRIBE_NODE: {
// subscribe_node_cmd_t *cmd_ptr;
// cmd_ptr = (subscribe_node_cmd_t *) msg_get_buf_ptr(pay_msg);
// const neu_plugin_intf_funs_t *intf_funs;
// neu_request_t req;
// neu_reqresp_subscribe_node_t sub_node_req;
// sub_node_req.grp_config = cmd_ptr->grp_config;
// sub_node_req.dst_node_id = cmd_ptr->dst_node_id;
// sub_node_req.src_node_id = cmd_ptr->src_node_id;
// intf_funs = adapter->plugin_info.module->intf_funs;
// req.req_id = adapter_get_req_id(adapter);
// req.req_type = NEU_REQRESP_SUBSCRIBE_NODE;
// req.sender_id = cmd_ptr->sender_id;
// req.buf_len = sizeof(neu_reqresp_subscribe_node_t);
// req.buf = (void *) &sub_node_req;
// if (adapter->plugin_info.module->type == NEU_NA_TYPE_DRIVER) {
// neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter,
//&req);
//} else {
// intf_funs->request(adapter->plugin_info.plugin, &req);
//}
// break;
//}
// case MSG_CMD_UNSUBSCRIBE_NODE: {
// unsubscribe_node_cmd_t *cmd_ptr;
// cmd_ptr = (unsubscribe_node_cmd_t *) msg_get_buf_ptr(pay_msg);
// const neu_plugin_intf_funs_t * intf_funs;
// neu_request_t req;
// neu_reqresp_unsubscribe_node_t unsub_node_req;
// unsub_node_req.grp_config = cmd_ptr->grp_config;
// unsub_node_req.dst_node_id = cmd_ptr->dst_node_id;
// unsub_node_req.src_node_id = cmd_ptr->src_node_id;
// intf_funs = adapter->plugin_info.module->intf_funs;
// req.req_id = adapter_get_req_id(adapter);
// req.req_type = NEU_REQRESP_UNSUBSCRIBE_NODE;
// req.sender_id = cmd_ptr->sender_id;
// req.buf_len = sizeof(neu_reqresp_unsubscribe_node_t);
// req.buf = (void *) &unsub_node_req;
// if (adapter->plugin_info.module->type == NEU_NA_TYPE_DRIVER) {
// neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter,
//&req);
//} else {
// intf_funs->request(adapter->plugin_info.plugin, &req);
//}
// break;
//}
case MSG_CMD_READ_DATA: {
// read_data_cmd_t *cmd_ptr;
// cmd_ptr = (read_data_cmd_t *) msg_get_buf_ptr(pay_msg);
@@ -1497,8 +1401,9 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
// req.sender_id = cmd_ptr->sender_id;
// req.buf_len = sizeof(neu_reqresp_read_t);
// req.buf = (void *) &read_req;
// if (adapter->plugin_info.module->type == NEU_NA_TYPE_DRIVER) {
// neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter,
// if (adapter->plugin_info.module->type == NEU_NA_TYPE_DRIVER)
// { neu_adapter_driver_process_msg((neu_adapter_driver_t *)
// adapter,
//&req);
//} else {
// intf_funs->request(adapter->plugin_info.plugin, &req);
@@ -1567,8 +1472,9 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
// req.sender_id = cmd_ptr->sender_id;
// req.buf_len = sizeof(neu_reqresp_write_t);
// req.buf = (void *) &write_req;
// if (adapter->plugin_info.module->type == NEU_NA_TYPE_DRIVER) {
// neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter,
// if (adapter->plugin_info.module->type == NEU_NA_TYPE_DRIVER)
// { neu_adapter_driver_process_msg((neu_adapter_driver_t *)
// adapter,
//&req);
//} else {
// intf_funs->request(adapter->plugin_info.plugin, &req);
@@ -1796,7 +1702,8 @@ neu_adapter_id_t neu_adapter_get_id(neu_adapter_t *adapter)
return adapter->id;
}
// NOTE: Do NOT expose this function in header files, use for persistence only!
// NOTE: Do NOT expose this function in header files, use for persistence
// only!
neu_adapter_id_t neu_adapter_set_id(neu_adapter_t *adapter, neu_adapter_id_t id)
{
if (adapter == NULL) {
src/adapter/driver/driver.c
+16
View File
@@ -451,6 +451,22 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
return ret;
}
int neu_adapter_driver_find_group(neu_adapter_driver_t *driver,
const char * name)
{
group_t *find = NULL;
int ret = NEU_ERR_GROUP_NOT_EXIST;
nng_mtx_lock(driver->mtx);
HASH_FIND_STR(driver->groups, name, find);
if (find != NULL) {
ret = NEU_ERR_SUCCESS;
}
nng_mtx_unlock(driver->mtx);
return ret;
}
UT_array *neu_adapter_driver_get_group(neu_adapter_driver_t *driver)
{
group_t * el = NULL, *tmp = NULL;
src/adapter/driver/driver_internal.h
+2
View File
@@ -38,6 +38,8 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
uint32_t interval);
int neu_adapter_driver_del_group(neu_adapter_driver_t *driver,
const char * name);
int neu_adapter_driver_find_group(neu_adapter_driver_t *driver,
const char * name);
UT_array *neu_adapter_driver_get_group(neu_adapter_driver_t *driver);
int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver,
src/base/neu_plugin_common.c
+14 -16
View File
@@ -218,38 +218,36 @@ int neu_system_get_group_configs(neu_plugin_t *plugin, const char *node_name,
return error;
}
uint32_t neu_plugin_send_subscribe_cmd(neu_plugin_t *plugin,
const char * app_name,
const char * driver_name,
const char * group)
int neu_plugin_send_subscribe_cmd(neu_plugin_t *plugin, const char *app_name,
const char *driver_name, const char *group)
{
uint32_t event_id = 0;
neu_reqresp_subscribe_node_t req = { 0 };
int error = 0;
neu_reqresp_subscribe_node_t req = { 0 };
req.app_name = (char *) app_name;
req.driver_name = (char *) driver_name;
req.group = (char *) group;
PLUGIN_SEND_CMD(plugin, NEU_REQRESP_SUBSCRIBE_NODE, req, event_id)
PLUGIN_CALL_CMD(plugin, NEU_REQRESP_SUBSCRIBE_NODE, req, intptr_t,
{ error = (intptr_t) resp; })
return event_id;
return error;
}
uint32_t neu_plugin_send_unsubscribe_cmd(neu_plugin_t *plugin,
const char * app_name,
const char * driver_name,
const char * group)
int neu_plugin_send_unsubscribe_cmd(neu_plugin_t *plugin, const char *app_name,
const char *driver_name, const char *group)
{
uint32_t event_id = 0;
neu_reqresp_unsubscribe_node_t req = { 0 };
int error = 0;
neu_reqresp_unsubscribe_node_t req = { 0 };
req.app_name = (char *) app_name;
req.driver_name = (char *) driver_name;
req.group = (char *) group;
PLUGIN_SEND_CMD(plugin, NEU_REQRESP_UNSUBSCRIBE_NODE, req, event_id)
PLUGIN_CALL_CMD(plugin, NEU_REQRESP_UNSUBSCRIBE_NODE, req, intptr_t,
{ error = (intptr_t) resp; })
return event_id;
return error;
}
void neu_plugin_send_read_cmd(neu_plugin_t *plugin, uint32_t event_id,
src/core/manager.c
+31 -11
View File
@@ -173,23 +173,36 @@ UT_array *neu_manager_get_nodes(neu_manager_t *manager, neu_node_type_e type)
int neu_manager_subscribe(neu_manager_t *manager, const char *app,
const char *driver, const char *group)
{
(void) manager;
(void) app;
(void) driver;
(void) group;
int ret = NEU_ERR_SUCCESS;
nng_pipe pipe = { 0 };
neu_adapter_t *adapter =
neu_node_manager_find(manager->node_manager, driver);
return 0;
if (adapter == NULL) {
return NEU_ERR_NODE_NOT_EXIST;
}
ret =
neu_adapter_driver_find_group((neu_adapter_driver_t *) adapter, group);
if (ret != NEU_ERR_SUCCESS) {
return ret;
}
adapter = neu_node_manager_find(manager->node_manager, app);
if (adapter == NULL) {
return NEU_ERR_NODE_NOT_EXIST;
}
pipe = neu_node_manager_get_pipe(manager->node_manager, app);
return neu_subscribe_manager_sub(manager->subscribe_manager, driver, app,
group, pipe);
}
int neu_manager_unsubscribe(neu_manager_t *manager, const char *app,
const char *driver, const char *group)
{
(void) manager;
(void) app;
(void) driver;
(void) group;
return 0;
return neu_subscribe_manager_unsub(manager->subscribe_manager, driver, app,
group);
}
int neu_manager_add_group(neu_manager_t *manager, const char *driver,
@@ -300,6 +313,11 @@ int neu_manager_stop_node(neu_manager_t *manager, const char *node)
return 0;
}
UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app)
{
return neu_subscribe_manager_get(manager->subscribe_manager, app);
}
int neu_manager_get_adapter_info(neu_manager_t *manager, const char *name,
neu_persist_adapter_info_t *info)
{
@@ -430,6 +448,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
case MSG_EVENT_UPDATE_TAGS:
case MSG_EVENT_ADD_PLUGIN:
case MSG_EVENT_UPDATE_PLUGIN:
case MSG_EVENT_SUBSCRIBE_NODE:
case MSG_EVENT_UNSUBSCRIBE_NODE:
case MSG_EVENT_DEL_PLUGIN: {
forward_msg(manager, msg, manager->persist_pipe);
nlog_info("forward event %d to %s pipe: %d", msg_get_type(pay_msg),
src/core/subscribe.c
+40 -2
View File
@@ -18,13 +18,15 @@
**/
#include <stdlib.h>
#include "adapter.h"
#include "errcodes.h"
#include "utils/log.h"
#include "subscribex.h"
typedef struct sub_elem_key {
char driver[128];
char group[128];
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
} sub_elem_key_t;
typedef struct sub_elem {
@@ -35,6 +37,14 @@ typedef struct sub_elem {
UT_hash_handle hh;
} sub_elem_t;
typedef struct app_elem {
char name[NEU_NODE_NAME_LEN];
UT_array *groups;
UT_hash_handle hh;
} app_elem_t;
static const UT_icd app_sub_icd = { sizeof(neu_app_subscribe_t), NULL, NULL,
NULL };
@@ -81,6 +91,33 @@ UT_array *neu_subscribe_manager_find(neu_subscribe_mgr_t *mgr,
}
}
UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app)
{
sub_elem_t *el = NULL, *tmp = NULL;
UT_array * groups = NULL;
UT_icd icd = { sizeof(neu_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)
{
if (strcmp(sub_app->app_name, app) == 0) {
neu_subscribe_info_t info = { 0 };
strncpy(info.driver, el->key.driver, sizeof(info.driver));
strncpy(info.app, app, sizeof(info.app));
strncpy(info.group, el->key.group, sizeof(info.group));
utarray_push_back(groups, &info);
}
}
}
return groups;
}
int neu_subscribe_manager_sub(neu_subscribe_mgr_t *mgr, const char *driver,
const char *app, const char *group, nng_pipe pipe)
{
@@ -106,6 +143,7 @@ int neu_subscribe_manager_sub(neu_subscribe_mgr_t *mgr, const char *driver,
find = calloc(1, sizeof(sub_elem_t));
utarray_new(find->apps, &app_sub_icd);
find->key = key;
HASH_ADD(hh, mgr->ss, key, sizeof(sub_elem_key_t), find);
}
utarray_push_back(find->apps, &app_sub);
src/core/subscribex.h
+2
View File
@@ -22,6 +22,7 @@
#include <nng/nng.h>
#include "define.h"
#include "utils/utextend.h"
typedef struct neu_subscribe_mgr neu_subscribe_mgr_t;
@@ -37,6 +38,7 @@ 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_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,
nng_pipe pipe);
src/persist/json/persist_json_subs.c
+1 -11
View File
@@ -61,10 +61,6 @@ int neu_json_decode_subscriptions_req(char * buf,
.name = "src_adapter_name",
.t = NEU_JSON_STR,
},
{
.name = "read_interval",
.t = NEU_JSON_INT,
},
{
.name =
"group_config_name",
@@ -79,8 +75,7 @@ int neu_json_decode_subscriptions_req(char * buf,
p_subscription->sub_adapter_name = subscription_elems[0].v.val_str;
p_subscription->src_adapter_name = subscription_elems[1].v.val_str;
p_subscription->read_interval = subscription_elems[2].v.val_int;
p_subscription->group_config_name = subscription_elems[3].v.val_str;
p_subscription->group_config_name = subscription_elems[2].v.val_str;
p_subscription++;
}
@@ -141,11 +136,6 @@ int neu_json_encode_subscriptions_resp(void *json_object, void *param)
.t = NEU_JSON_STR,
.v.val_str = p_subscription->src_adapter_name,
},
{
.name = "read_interval",
.t = NEU_JSON_INT,
.v.val_int = p_subscription->read_interval,
},
{
.name = "group_config_name",
.t = NEU_JSON_STR,
src/persist/json/persist_json_subs.h
+6 -8
View File
@@ -32,10 +32,9 @@ extern "C" {
#endif
typedef struct {
char * group_config_name;
char * sub_adapter_name;
int64_t read_interval;
char * src_adapter_name;
char *group_config_name;
char *sub_adapter_name;
char *src_adapter_name;
} neu_json_subscriptions_req_subscription_t;
typedef struct {
@@ -48,10 +47,9 @@ int neu_json_decode_subscriptions_req(char * buf,
void neu_json_decode_subscriptions_req_free(neu_json_subscriptions_req_t *req);
typedef struct {
char * group_config_name;
char * sub_adapter_name;
int64_t read_interval;
char * src_adapter_name;
char *group_config_name;
char *sub_adapter_name;
char *src_adapter_name;
} neu_json_subscriptions_resp_subscription_t;
typedef struct {
src/persist/persist.c
+29 -16
View File
@@ -390,7 +390,7 @@ static int write_file_string(const char *fn, const char *s)
fclose(f);
nlog_debug("pwersister write %s to %s", s, tmp);
nlog_debug("persister write %s to %s", s, tmp);
// rename the temporary file to the destination file
if (0 != rename(tmp, fn)) {
@@ -964,7 +964,7 @@ int neu_persister_load_datatags(neu_persister_t *persister,
int neu_persister_store_subscriptions(neu_persister_t *persister,
const char * adapter_name,
vector_t * subscription_infos)
UT_array * subscription_infos)
{
char path[PATH_MAX_SIZE] = { 0 };
@@ -986,10 +986,19 @@ int neu_persister_store_subscriptions(neu_persister_t *persister,
return NEU_ERR_FAILURE;
}
neu_json_subscriptions_req_t subs_resp = {
.n_subscription = subscription_infos->size,
.subscriptions = subscription_infos->data,
};
neu_json_subscriptions_req_t subs_resp = { 0 };
int index = 0;
subs_resp.n_subscription = utarray_len(subscription_infos);
subs_resp.subscriptions =
calloc(subs_resp.n_subscription,
sizeof(neu_json_subscriptions_req_subscription_t));
utarray_foreach(subscription_infos, neu_subscribe_info_t *, info)
{
subs_resp.subscriptions[index].group_config_name = info->group;
subs_resp.subscriptions[index].src_adapter_name = info->driver;
subs_resp.subscriptions[index].sub_adapter_name = info->app;
index += 1;
}
char *result = NULL;
rv = neu_json_encode_by_fn(&subs_resp, neu_json_encode_subscriptions_resp,
@@ -999,13 +1008,14 @@ int neu_persister_store_subscriptions(neu_persister_t *persister,
}
free(result);
free(subs_resp.subscriptions);
return rv;
}
int neu_persister_load_subscriptions(neu_persister_t *persister,
const char * adapter_name,
vector_t ** subscription_infos)
UT_array ** subscription_infos)
{
char path[PATH_MAX_SIZE] = { 0 };
@@ -1034,16 +1044,19 @@ int neu_persister_load_subscriptions(neu_persister_t *persister,
return rv;
}
vector_t *vec = vector_new_move_from_buf(
subs_req->subscriptions, subs_req->n_subscription,
subs_req->n_subscription, sizeof(neu_persist_subscription_info_t));
UT_icd icd = { sizeof(neu_persist_subscription_info_t), NULL, NULL, NULL };
utarray_new(*subscription_infos, &icd);
if (NULL != vec) {
subs_req->n_subscription = 0;
subs_req->subscriptions = NULL;
*subscription_infos = vec;
} else {
rv = NEU_ERR_ENOMEM;
for (int i = 0; i < subs_req->n_subscription; i++) {
neu_persist_subscription_info_t info = { 0 };
info.group_config_name =
strdup(subs_req->subscriptions[i].group_config_name);
info.src_adapter_name =
strdup(subs_req->subscriptions[i].src_adapter_name);
info.sub_adapter_name =
strdup(subs_req->subscriptions[i].sub_adapter_name);
utarray_push_back(*subscription_infos, &info);
}
neu_json_decode_subscriptions_req_free(subs_req);
src/persist/persist.h
+8 -10
View File
@@ -90,17 +90,15 @@ static inline void neu_persist_datatag_infos_free(vector_t *datatag_infos)
}
static inline void
neu_persist_subscription_infos_free(vector_t *subscription_infos)
neu_persist_subscription_infos_free(UT_array *subscription_infos)
{
VECTOR_FOR_EACH(subscription_infos, iter)
utarray_foreach(subscription_infos, neu_persist_subscription_info_t *, info)
{
neu_persist_subscription_info_t *p =
(neu_persist_subscription_info_t *) iterator_get(&iter);
free(p->src_adapter_name);
free(p->sub_adapter_name);
free(p->group_config_name);
free(info->group_config_name);
free(info->src_adapter_name);
free(info->sub_adapter_name);
}
vector_free(subscription_infos);
utarray_free(subscription_infos);
}
/**
@@ -213,7 +211,7 @@ int neu_persister_load_datatags(neu_persister_t *persister,
*/
int neu_persister_store_subscriptions(neu_persister_t *persister,
const char * adapter_name,
vector_t * subscription_infos);
UT_array * subscription_infos);
/**
* Load adapter subscription infos.
* @param persister persiter object.
@@ -225,7 +223,7 @@ int neu_persister_store_subscriptions(neu_persister_t *persister,
*/
int neu_persister_load_subscriptions(neu_persister_t *persister,
const char * adapter_name,
vector_t ** subscription_infos);
UT_array ** subscription_infos);
/**
* Persist group config.