diff --git a/plugins/restful/global_config_handle.c b/plugins/restful/global_config_handle.c index 4243af04..523d7af2 100644 --- a/plugins/restful/global_config_handle.c +++ b/plugins/restful/global_config_handle.c @@ -52,6 +52,7 @@ typedef enum { STATE_ADD_NODE, STATE_ADD_GROUP, STATE_ADD_TAG, + STATE_ADD_SUBSCRIPTION, STATE_END, } context_state_e; @@ -96,6 +97,7 @@ static int add_node(context_t *ctx, neu_json_get_nodes_resp_node_t *req); static int add_group(context_t * ctx, neu_json_get_driver_group_resp_group_t *data); static int add_tag(context_t *ctx, neu_json_add_tags_req_t *data); +static int add_subscription(context_t *ctx, neu_json_subscribe_req_t *data); static inline bool is_static_node(const char *name, const char *plugin) { @@ -384,6 +386,29 @@ static void put_context_next(context_t *ctx, neu_reqresp_type_e type, break; } + ctx->idx = 0; + ctx->state = STATE_ADD_SUBSCRIPTION; + } + // fall through + + case STATE_ADD_SUBSCRIPTION: { + neu_resp_error_t * err = data; + neu_json_global_config_req_subscriptions_t *subs = + ctx->req->subscriptions; + + if (ctx->idx > 0 && 0 != err->error) { + // received error response + ctx->error = err->error; + ctx->state = STATE_END; + break; + } + + if (ctx->idx < (size_t) subs->n_subscription) { + NEXT(ctx, add_subscription, &subs->subscriptions[ctx->idx]); + ++ctx->idx; + break; + } + ctx->state = STATE_END; break; } @@ -880,6 +905,29 @@ error: return ret; } +static int add_subscription(context_t *ctx, neu_json_subscribe_req_t *data) +{ + neu_plugin_t *plugin = neu_rest_get_plugin(); + + neu_reqresp_head_t header = { + .ctx = ctx->aio, + .type = NEU_REQ_SUBSCRIBE_GROUP, + }; + + neu_req_subscribe_t cmd = { 0 }; + strcpy(cmd.app, data->app); + strcpy(cmd.driver, data->driver); + strcpy(cmd.group, data->group); + cmd.params = data->params; // ownership moved + data->params = NULL; + + if (0 != neu_plugin_op(plugin, header, &cmd)) { + return NEU_ERR_IS_BUSY; + } + + return 0; +} + void handle_get_global_config(nng_aio *aio) { NEU_VALIDATE_JWT(aio); diff --git a/src/parser/neu_json_global_config.c b/src/parser/neu_json_global_config.c index 8161bdda..d9463a43 100644 --- a/src/parser/neu_json_global_config.c +++ b/src/parser/neu_json_global_config.c @@ -220,6 +220,140 @@ void neu_json_decode_global_config_req_tags_free( free(req_tags); } +static int decode_subscribe_req(json_t *json_obj, neu_json_subscribe_req_t *req) +{ + neu_json_elem_t req_elems[] = { + { + .name = "driver", + .t = NEU_JSON_STR, + }, + { + .name = "group", + .t = NEU_JSON_STR, + }, + }; + + int ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems), + req_elems); + if (ret != 0) { + goto error; + } + + ret = neu_json_dump_key(json_obj, "params", &req->params, false); + if (0 != ret) { + goto error; + } + + req->driver = req_elems[0].v.val_str; + req->group = req_elems[1].v.val_str; + return 0; + +error: + free(req_elems[0].v.val_str); + free(req_elems[1].v.val_str); + return -1; +} + +static inline void fini_subscribe_req(neu_json_subscribe_req_t *req) +{ + free(req->app); + free(req->driver); + free(req->group); + free(req->params); +} + +int neu_json_decode_global_config_req_subscriptions( + void *json_obj, neu_json_global_config_req_subscriptions_t **result) +{ + int n_subscription = 0; + json_t * sub_array = NULL; + neu_json_subscribe_req_t * p = NULL; + neu_json_global_config_req_subscriptions_t *req = NULL; + + sub_array = json_object_get(json_obj, "subscriptions"); + if (!json_is_array(sub_array)) { + nlog_error("no `subscriptions` array field"); + return -1; + } + + for (size_t i = 0; i < json_array_size(sub_array); ++i) { + json_t *grp_array = + json_object_get(json_array_get(sub_array, i), "groups"); + if (!json_is_array(grp_array)) { + nlog_error("no subscriptions `group` field"); + return -1; + } + n_subscription += json_array_size(grp_array); + } + + req = calloc(1, sizeof(*req)); + if (NULL == req) { + nlog_error("calloc fail"); + return -1; + } + + req->n_subscription = n_subscription; + req->subscriptions = + calloc(req->n_subscription, sizeof(*req->subscriptions)); + if (NULL == req->subscriptions) { + free(req); + nlog_error("calloc fail"); + return -1; + } + + p = req->subscriptions; + for (int i = 0; i < req->n_subscription; ++i) { + json_t *sub_obj = json_array_get(sub_array, i); + json_t *app_obj = json_object_get(sub_obj, "app"); + + if (NULL == app_obj) { + nlog_error("no subscriptions `app` field"); + goto error; + } + + json_t *grp_array = json_object_get(sub_obj, "groups"); + for (size_t j = 0; j < json_array_size(grp_array); ++j) { + p->app = strdup(json_string_value(app_obj)); + if (NULL == p->app) { + nlog_error("strdup fail"); + goto error; + } + if (0 != decode_subscribe_req(json_array_get(grp_array, j), p)) { + free(p->app); + goto error; + } + } + + ++p; + } + + *result = req; + return 0; + +error: + while (--p >= req->subscriptions) { + fini_subscribe_req(p); + } + free(req->subscriptions); + free(req); + return -1; +} + +void neu_json_decode_global_config_req_subscriptions_free( + neu_json_global_config_req_subscriptions_t *req_sub) +{ + if (NULL == req_sub) { + return; + } + + for (int i = 0; i < req_sub->n_subscription; ++i) { + fini_subscribe_req(&req_sub->subscriptions[i]); + } + + free(req_sub->subscriptions); + free(req_sub); +} + int neu_json_decode_global_config_req(char * buf, neu_json_global_config_req_t **result) { @@ -239,7 +373,10 @@ int neu_json_decode_global_config_req(char * buf, 0 == neu_json_decode_get_driver_group_resp_json(json_obj, &req->groups) && - 0 == neu_json_decode_global_config_req_tags(json_obj, &req->tags)) { + 0 == neu_json_decode_global_config_req_tags(json_obj, &req->tags) && + 0 == + neu_json_decode_global_config_req_subscriptions( + json_obj, &req->subscriptions)) { *result = req; } else { ret = -1; @@ -256,6 +393,8 @@ void neu_json_decode_global_config_req_free(neu_json_global_config_req_t *req) neu_json_decode_get_nodes_resp_free(req->nodes); neu_json_decode_get_driver_group_resp_free(req->groups); neu_json_decode_global_config_req_tags_free(req->tags); + neu_json_decode_global_config_req_subscriptions_free( + req->subscriptions); free(req); } } diff --git a/src/parser/neu_json_global_config.h b/src/parser/neu_json_global_config.h index f8cee153..253432a6 100644 --- a/src/parser/neu_json_global_config.h +++ b/src/parser/neu_json_global_config.h @@ -41,9 +41,20 @@ void neu_json_decode_global_config_req_tags_free( neu_json_global_config_req_tags_t *tags); typedef struct { - neu_json_get_nodes_resp_t * nodes; - neu_json_get_driver_group_resp_t * groups; - neu_json_global_config_req_tags_t *tags; + int n_subscription; + neu_json_subscribe_req_t *subscriptions; +} neu_json_global_config_req_subscriptions_t; + +int neu_json_decode_global_config_req_subscriptions( + void *json_obj, neu_json_global_config_req_subscriptions_t **result); +void neu_json_decode_global_config_req_subscriptions_free( + neu_json_global_config_req_subscriptions_t *req_sub); + +typedef struct { + neu_json_get_nodes_resp_t * nodes; + neu_json_get_driver_group_resp_t * groups; + neu_json_global_config_req_tags_t * tags; + neu_json_global_config_req_subscriptions_t *subscriptions; } neu_json_global_config_req_t; int neu_json_decode_global_config_req(char * buf,