GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-28 14:07:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1078 from eeff/eeff/dev

Browse Source 
feat(restful): get global config subscriptions/settings
This commit is contained in:
eeff
2023-03-01 00:31:14 +08:00
committed by GitHub
parent c494411763 e4a1e2958f
commit 149cecea03
1 changed files with 195 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/restful/global_config_handle.c
+195
View File
@@ -42,6 +42,9 @@ typedef enum {
STATE_GET_DRIVER,
STATE_GET_GROUP,
STATE_GET_TAG,
STATE_GET_SUBSCRIPTION,
STATE_GET_APP_SETTING,
STATE_GET_DRIVER_SETTING,
STATE_END,
} context_state_e;
@@ -62,6 +65,12 @@ static int get_groups(context_t *ctx, int unused);
static int get_groups_resp(context_t *ctx, neu_resp_get_driver_group_t *groups);
static int get_tags(context_t *ctx, neu_resp_driver_group_info_t *info);
static int get_tags_resp(context_t *ctx, neu_resp_get_tag_t *tags);
static int get_subscriptions(context_t *ctx, neu_resp_node_info_t *info);
static int get_subscriptions_resp(context_t * ctx,
neu_resp_get_subscribe_group_t *groups);
static int get_setting(context_t *ctx, neu_resp_node_info_t *info);
static int get_setting_resp(context_t * ctx,
neu_resp_get_node_setting_t *setting);
static context_t *context_new(nng_aio *aio)
{
@@ -145,6 +154,45 @@ static void context_next(context_t *ctx, neu_reqresp_type_e type, void *data)
NEXT(ctx, get_tags, ctx->iter);
break;
}
ctx->state = STATE_GET_SUBSCRIPTION;
// fall through
case STATE_GET_SUBSCRIPTION:
// generate empty array on fall through
NEXT(ctx, get_subscriptions_resp, ctx->iter ? data : NULL);
ctx->iter = utarray_next(ctx->apps, ctx->iter);
if (ctx->iter) {
NEXT(ctx, get_subscriptions, ctx->iter);
break;
}
ctx->state = STATE_GET_APP_SETTING;
// fall through
case STATE_GET_APP_SETTING:
if (NULL == ctx->iter) {
// generate empty array on fall through
NEXT(ctx, get_setting_resp, NULL);
} else if (NEU_RESP_ERROR != type) {
NEXT(ctx, get_setting_resp, data);
}
// ignore error response message, keep going
ctx->iter = utarray_next(ctx->apps, ctx->iter);
if (ctx->iter) {
NEXT(ctx, get_setting, ctx->iter);
break;
}
ctx->state = STATE_GET_DRIVER_SETTING;
// fall through
case STATE_GET_DRIVER_SETTING:
if (NULL == ctx->iter) {
// generate empty array on fall through
NEXT(ctx, get_setting_resp, NULL);
} else if (NEU_RESP_ERROR != type) {
NEXT(ctx, get_setting_resp, data);
}
ctx->iter = utarray_next(ctx->drivers, ctx->iter);
if (ctx->iter) {
NEXT(ctx, get_setting, ctx->iter);
break;
}
ctx->state = STATE_END;
break;
default:
@@ -366,6 +414,153 @@ end:
return rv;
}
static int get_subscriptions(context_t *ctx, neu_resp_node_info_t *info)
{
neu_plugin_t *plugin = neu_rest_get_plugin();
neu_reqresp_head_t header = {
.ctx = ctx->aio,
.type = NEU_REQ_GET_SUBSCRIBE_GROUP,
};
neu_req_get_subscribe_group_t cmd = { 0 };
strcpy(cmd.app, info->node);
if (0 != neu_plugin_op(plugin, header, &cmd)) {
return NEU_ERR_IS_BUSY;
}
return 0;
}
static int get_subscriptions_resp(context_t * ctx,
neu_resp_get_subscribe_group_t *groups)
{
int rv = 0;
neu_json_get_subscribe_resp_t sub_grp_configs = { 0 };
json_t * sub_obj = NULL;
json_t * sub_arr = NULL;
// allocate `subscriptions` array if none
if (NULL == (sub_arr = json_object_get(ctx->json, "subscriptions")) &&
(NULL == (sub_arr = json_array()) ||
// tag array ownership moved
0 != json_object_set_new(ctx->json, "subscriptions", sub_arr))) {
rv = NEU_ERR_EINTERNAL;
goto end;
}
if (NULL == groups || 0 == utarray_len(groups->groups)) {
// empty subscriptions array, all done
goto end;
}
sub_grp_configs.n_group = utarray_len(groups->groups);
sub_grp_configs.groups = calloc(
sub_grp_configs.n_group, sizeof(neu_json_get_subscribe_resp_group_t));
if (NULL == sub_grp_configs.groups) {
rv = NEU_ERR_EINTERNAL;
goto end;
}
utarray_foreach(groups->groups, neu_resp_subscribe_info_t *, group)
{
int index = utarray_eltidx(groups->groups, group);
sub_grp_configs.groups[index].driver = group->driver;
sub_grp_configs.groups[index].group = group->group;
sub_grp_configs.groups[index].params = group->params;
}
// accumulate subscription object in `subscriptions` array
if (NULL == (sub_obj = json_object()) ||
// subscription object ownership moved
0 != json_array_append_new(sub_arr, sub_obj)) {
rv = NEU_ERR_EINTERNAL;
goto end;
}
if (0 != neu_json_encode_get_subscribe_resp(sub_obj, &sub_grp_configs)) {
rv = NEU_ERR_EINTERNAL;
goto end;
}
neu_resp_node_info_t *info = ctx->iter;
if (0 != json_object_set_new(sub_obj, "app", json_string(info->node))) {
rv = NEU_ERR_EINTERNAL;
}
end:
free(sub_grp_configs.groups);
if (groups && groups->groups) {
utarray_free(groups->groups);
}
return rv;
}
static int get_setting(context_t *ctx, neu_resp_node_info_t *info)
{
neu_plugin_t *plugin = neu_rest_get_plugin();
neu_reqresp_head_t header = {
.ctx = ctx->aio,
.type = NEU_REQ_GET_NODE_SETTING,
};
neu_req_get_node_setting_t cmd = { 0 };
strcpy(cmd.ndoe, info->node);
if (0 != neu_plugin_op(plugin, header, &cmd)) {
return NEU_ERR_IS_BUSY;
}
return 0;
}
static int get_setting_resp(context_t * ctx,
neu_resp_get_node_setting_t *setting)
{
int rv = 0;
json_t *setting_obj = NULL;
json_t *setting_arr = NULL;
// allocate `settings` array if none
if (NULL == (setting_arr = json_object_get(ctx->json, "settings")) &&
(NULL == (setting_arr = json_array()) ||
// tag array ownership moved
0 != json_object_set_new(ctx->json, "settings", setting_arr))) {
rv = NEU_ERR_EINTERNAL;
goto end;
}
if (NULL == setting) {
// empty subscriptions array, all done
goto end;
}
// accumulate setting object in `settings` array
if (NULL == (setting_obj = json_object()) ||
// setting object ownership moved
0 != json_array_append_new(setting_arr, setting_obj)) {
rv = NEU_ERR_EINTERNAL;
goto end;
}
neu_json_get_node_setting_resp_t resp = {
.node = setting->node,
.setting = setting->setting,
};
if (0 != neu_json_encode_get_node_setting_resp(setting_obj, &resp)) {
rv = NEU_ERR_EINTERNAL;
}
end:
if (setting) {
free(setting->setting);
}
return rv;
}
void handle_get_global_config(nng_aio *aio)
{
NEU_VALIDATE_JWT(aio);