GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-17 04:40:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

update api:sub info of datalayers plugin

Browse Source
This commit is contained in:
hxy7yx
2025-05-26 02:34:05 -07:00
parent f49af0f879
commit 83d50d0b9b
10 changed files with 292 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/msg.h
+7
View File
@@ -68,6 +68,8 @@ typedef enum neu_reqresp_type {
NEU_RESP_GET_DATALAYERS_GROUPS,
NEU_REQ_GET_DATALAYERS_TAGS,
NEU_RESP_GET_DATALAYERS_TAGS,
NEU_REQ_GET_DATALAYERS_TAG,
NEU_RESP_GET_DATALAYERS_TAG,
NEU_REQ_NODE_INIT,
NEU_REQ_NODE_UNINIT,
@@ -199,6 +201,8 @@ static const char *neu_reqresp_type_string_t[] = {
[NEU_RESP_GET_DATALAYERS_GROUPS] = "NEU_RESP_GET_DATALAYERS_GROUPS",
[NEU_REQ_GET_DATALAYERS_TAGS] = "NEU_REQ_GET_DATALAYERS_TAGS",
[NEU_RESP_GET_DATALAYERS_TAGS] = "NEU_RESP_GET_DATALAYERS_TAGS",
[NEU_REQ_GET_DATALAYERS_TAG] = "NEU_REQ_GET_DATALAYERS_TAG",
[NEU_RESP_GET_DATALAYERS_TAG] = "NEU_RESP_GET_DATALAYERS_TAG",
[NEU_REQ_NODE_INIT] = "NEU_REQ_NODE_INIT",
[NEU_REQ_NODE_UNINIT] = "NEU_REQ_NODE_UNINIT",
@@ -697,6 +701,9 @@ typedef struct neu_req_get_subscribe_group {
typedef struct {
char app[NEU_NODE_NAME_LEN];
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
char tag[NEU_TAG_NAME_LEN];
} neu_req_get_sub_driver_tags_t;
typedef struct neu_resp_subscribe_info {
plugins/restful/datalayers_handle.c
+73 -20
View File
@@ -41,27 +41,8 @@ void handle_datalayers_get_groups(nng_aio *aio)
NEU_VALIDATE_JWT(aio);
// required parameter
snprintf(cmd.app, sizeof(cmd.app), "%s", "DataStorage");
// optional parameter
ret = neu_http_get_param_str(aio, "driver", cmd.driver, sizeof(cmd.driver));
if (-1 == ret || (size_t) ret == sizeof(cmd.driver)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, error_code.error, result_error);
})
return;
}
// optional parameter
ret = neu_http_get_param_str(aio, "group", cmd.group, sizeof(cmd.group));
if (-1 == ret || (size_t) ret == sizeof(cmd.driver)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, error_code.error, result_error);
})
return;
}
ret = neu_plugin_op(plugin, header, &cmd);
if (ret != 0) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
@@ -111,7 +92,22 @@ void handle_datalayers_get_tags(nng_aio *aio)
NEU_VALIDATE_JWT(aio);
// required parameter
ret = neu_http_get_param_str(aio, "driver", cmd.driver, sizeof(cmd.driver));
if (ret <= 0 || (size_t) ret == sizeof(cmd.driver)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, error_code.error, result_error);
})
return;
}
ret = neu_http_get_param_str(aio, "group", cmd.group, sizeof(cmd.group));
if (ret <= 0 || (size_t) ret == sizeof(cmd.group)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, error_code.error, result_error);
})
return;
}
snprintf(cmd.app, sizeof(cmd.app), "%s", "DataStorage");
ret = neu_plugin_op(plugin, header, &cmd);
@@ -137,6 +133,63 @@ void handle_datalayers_get_tags_resp(nng_aio * aio,
neu_http_ok(aio, result);
free(result);
if (tags) {
utarray_foreach(tags->infos, neu_resp_get_sub_driver_tags_info_t *,
info)
{
utarray_free(info->tags);
}
utarray_free(tags->infos);
}
}
void handle_datalayers_get_tag(nng_aio *aio)
{
int ret = 0;
neu_plugin_t * plugin = neu_rest_get_plugin();
neu_req_get_sub_driver_tags_t cmd = { 0 };
neu_reqresp_head_t header = {
.ctx = aio,
.type = NEU_REQ_GET_DATALAYERS_TAG,
.otel_trace_type = NEU_OTEL_TRACE_TYPE_REST_COMM,
};
NEU_VALIDATE_JWT(aio);
ret = neu_http_get_param_str(aio, "tag", cmd.tag, sizeof(cmd.tag));
if (ret <= 0 || (size_t) ret == sizeof(cmd.tag)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, error_code.error, result_error);
})
return;
}
snprintf(cmd.app, sizeof(cmd.app), "%s", "DataStorage");
ret = neu_plugin_op(plugin, header, &cmd);
if (ret != 0) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
});
}
}
void handle_datalayers_get_tag_resp(nng_aio * aio,
neu_resp_get_sub_driver_tags_t *tags)
{
char *result = NULL;
int ret = neu_json_encode_by_fn(
tags, neu_json_encode_datalayers_get_tag_resp, &result);
if (ret != 0 || result == NULL) {
neu_http_response(
aio, 500, "{\"error\":4200,\"reason\":\"failed to encode tags\"}");
return;
}
neu_http_ok(aio, result);
free(result);
if (tags) {
utarray_foreach(tags->infos, neu_resp_get_sub_driver_tags_info_t *,
plugins/restful/datalayers_handle.h
+4
View File
@@ -32,4 +32,8 @@ void handle_datalayers_get_tags(nng_aio *aio);
void handle_datalayers_get_tags_resp(nng_aio * aio,
neu_resp_get_sub_driver_tags_t *tags);
void handle_datalayers_get_tag(nng_aio *aio);
void handle_datalayers_get_tag_resp(nng_aio * aio,
neu_resp_get_sub_driver_tags_t *tags);
#endif
plugins/restful/handle.c
+9
View File
@@ -155,6 +155,9 @@ static struct neu_http_handler cors_handler[] = {
{
.url = "/api/v2/datalayers/tags",
},
{
.url = "/api/v2/datalayers/tag",
},
};
static struct neu_http_handler rest_handlers[] = {
@@ -513,6 +516,12 @@ static struct neu_http_handler rest_handlers[] = {
.url = "/api/v2/datalayers/tags",
.value.handler = handle_datalayers_get_tags,
},
{
.method = NEU_HTTP_METHOD_GET,
.type = NEU_HTTP_HANDLER_FUNCTION,
.url = "/api/v2/datalayers/tag",
.value.handler = handle_datalayers_get_tag,
},
};
void neu_rest_handler(const struct neu_http_handler **handlers, uint32_t *size)
plugins/restful/rest.c
+7
View File
@@ -298,6 +298,13 @@ static int dashb_plugin_request(neu_plugin_t * plugin,
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
case NEU_RESP_GET_DATALAYERS_TAG:
handle_datalayers_get_tag_resp(header->ctx,
(neu_resp_get_sub_driver_tags_t *) data);
if (neu_otel_control_is_started() && trace) {
neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0);
}
break;
case NEU_RESP_GET_NODE_SETTING:
handle_get_node_setting_resp(header->ctx,
(neu_resp_get_node_setting_t *) data);
src/adapter/adapter.c
+1
View File
@@ -757,6 +757,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_RESP_GET_SUBSCRIBE_GROUP:
case NEU_RESP_GET_DATALAYERS_GROUPS:
case NEU_RESP_GET_DATALAYERS_TAGS:
case NEU_RESP_GET_DATALAYERS_TAG:
case NEU_RESP_ADD_TAG:
case NEU_RESP_ADD_GTAG:
case NEU_RESP_UPDATE_TAG:
src/base/msg_internal.h
+3 -1
View File
@@ -135,7 +135,9 @@ extern "C" {
XX(NEU_REQ_GET_DATALAYERS_GROUPS, neu_req_get_subscribe_group_t) \
XX(NEU_RESP_GET_DATALAYERS_GROUPS, neu_resp_get_subscribe_group_t) \
XX(NEU_REQ_GET_DATALAYERS_TAGS, neu_req_get_sub_driver_tags_t) \
XX(NEU_RESP_GET_DATALAYERS_TAGS, neu_resp_get_sub_driver_tags_t)
XX(NEU_RESP_GET_DATALAYERS_TAGS, neu_resp_get_sub_driver_tags_t) \
XX(NEU_REQ_GET_DATALAYERS_TAG, neu_req_get_sub_driver_tags_t) \
XX(NEU_RESP_GET_DATALAYERS_TAG, neu_resp_get_sub_driver_tags_t)
static inline size_t neu_reqresp_size(neu_reqresp_type_e t)
{
src/core/manager.c
+78
View File
@@ -1056,10 +1056,18 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
(neu_req_get_sub_driver_tags_t *) &header[1];
neu_resp_get_sub_driver_tags_t resp = { 0 };
UT_array *groups = neu_manager_get_sub_group(manager, cmd->app);
bool found = false;
utarray_new(resp.infos, neu_resp_get_sub_driver_tags_info_icd());
utarray_foreach(groups, neu_resp_subscribe_info_t *, info)
{
if (strcmp(info->driver, cmd->driver) != 0 ||
strcmp(info->group, cmd->group) != 0) {
continue;
}
found = true;
neu_resp_get_sub_driver_tags_info_t in = { 0 };
neu_adapter_t * driver =
neu_node_manager_find(manager->node_manager, info->driver);
@@ -1074,12 +1082,82 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
}
utarray_free(groups);
if (!found) {
utarray_free(resp.infos);
neu_resp_error_t e = { .error = NEU_ERR_GROUP_NOT_EXIST };
header->type = NEU_RESP_ERROR;
neu_msg_exchange(header);
reply(manager, header, &e);
break;
}
strcpy(header->receiver, header->sender);
header->type = NEU_RESP_GET_DATALAYERS_TAGS;
reply(manager, header, &resp);
break;
}
case NEU_REQ_GET_DATALAYERS_TAG: {
neu_req_get_sub_driver_tags_t *cmd =
(neu_req_get_sub_driver_tags_t *) &header[1];
neu_resp_get_sub_driver_tags_t resp = { 0 };
UT_array *groups = neu_manager_get_sub_group(manager, cmd->app);
bool found = false;
utarray_new(resp.infos, neu_resp_get_sub_driver_tags_info_icd());
utarray_foreach(groups, neu_resp_subscribe_info_t *, info)
{
neu_resp_get_sub_driver_tags_info_t in = { 0 };
neu_adapter_t * driver =
neu_node_manager_find(manager->node_manager, info->driver);
assert(driver != NULL);
neu_adapter_driver_get_value_tag((neu_adapter_driver_t *) driver,
info->group, &in.tags);
if (in.tags) {
UT_array *filtered_tags = NULL;
utarray_new(filtered_tags, neu_tag_get_icd());
utarray_foreach(in.tags, neu_datatag_t *, tag)
{
if (strstr(tag->name, cmd->tag) != NULL) {
utarray_push_back(filtered_tags, tag);
}
}
utarray_free(in.tags);
if (utarray_len(filtered_tags) > 0) {
strcpy(in.driver, info->driver);
strcpy(in.group, info->group);
in.tags = filtered_tags;
utarray_push_back(resp.infos, &in);
found = true;
} else {
utarray_free(filtered_tags);
}
}
}
utarray_free(groups);
if (!found) {
utarray_free(resp.infos);
neu_resp_error_t e = { .error = NEU_ERR_TAG_NOT_EXIST };
header->type = NEU_RESP_ERROR;
neu_msg_exchange(header);
reply(manager, header, &e);
break;
}
strcpy(header->receiver, header->sender);
header->type = NEU_RESP_GET_DATALAYERS_TAG;
reply(manager, header, &resp);
break;
}
case NEU_REQ_GET_NODES_STATE: {
neu_resp_get_nodes_state_t resp = { 0 };
UT_array *states = neu_node_manager_get_state(manager->node_manager);
src/parser/neu_json_datalayers.c
+108
View File
@@ -163,6 +163,7 @@ int neu_json_encode_datalayers_get_tags_resp(void *json_object_param,
json_object_set_new(tag_obj, "name", json_string(tag->name));
json_object_set_new(tag_obj, "type", json_integer(tag->type));
json_object_set_new(tag_obj, "category", json_integer(3));
json_object_set_new(tag_obj, "leaf", json_true());
json_object_set_new(tag_obj, "children", json_array());
json_array_append_new(tag_array, tag_obj);
@@ -183,6 +184,113 @@ int neu_json_encode_datalayers_get_tags_resp(void *json_object_param,
.v.val_object = final_obj,
} };
int ret = neu_json_encode_field(json_object_param, elems,
NEU_JSON_ELEM_SIZE(elems));
json_decref(final_obj);
return ret;
}
int neu_json_encode_datalayers_get_tag_resp(void *json_object_param,
void *param)
{
neu_resp_get_sub_driver_tags_t *resp =
(neu_resp_get_sub_driver_tags_t *) param;
json_t *node_array = json_array();
for (unsigned int i = 0; i < utarray_len(resp->infos); i++) {
neu_resp_get_sub_driver_tags_info_t *info =
(neu_resp_get_sub_driver_tags_info_t *) utarray_eltptr(resp->infos,
i);
char node_id[37];
char node_uuid_input[128];
snprintf(node_uuid_input, sizeof(node_uuid_input), "%s", info->driver);
generate_fixed_uuid(node_uuid_input, node_id);
json_t *node_obj = NULL;
for (size_t j = 0; j < json_array_size(node_array); j++) {
node_obj = json_array_get(node_array, j);
if (strcmp(json_string_value(json_object_get(node_obj, "id")),
node_id) == 0) {
break;
}
node_obj = NULL;
}
if (!node_obj) {
node_obj = json_object();
json_object_set_new(node_obj, "id", json_string(node_id));
json_object_set_new(node_obj, "name", json_string(info->driver));
json_object_set_new(node_obj, "category", json_integer(1));
json_t *node_children = json_array();
json_object_set_new(node_obj, "children", node_children);
json_array_append_new(node_array, node_obj);
}
json_t *group_obj = NULL;
char group_id[37];
char group_uuid_input[256];
snprintf(group_uuid_input, sizeof(group_uuid_input), "%s-%s",
info->driver, info->group);
generate_fixed_uuid(group_uuid_input, group_id);
json_t *children_array = json_object_get(node_obj, "children");
for (size_t k = 0; k < json_array_size(children_array); k++) {
group_obj = json_array_get(children_array, k);
if (strcmp(json_string_value(json_object_get(group_obj, "id")),
group_id) == 0) {
break;
}
group_obj = NULL;
}
if (!group_obj) {
group_obj = json_object();
json_object_set_new(group_obj, "id", json_string(group_id));
json_object_set_new(group_obj, "name", json_string(info->group));
json_object_set_new(group_obj, "category", json_integer(2));
json_t *group_children = json_array();
json_object_set_new(group_obj, "children", group_children);
json_array_append_new(children_array, group_obj);
}
json_t *tag_array = json_object_get(group_obj, "children");
if (info->tags) {
for (unsigned int j = 0; j < utarray_len(info->tags); j++) {
neu_datatag_t *tag =
(neu_datatag_t *) utarray_eltptr(info->tags, j);
char tag_id[37];
char tag_uuid_input[512];
snprintf(tag_uuid_input, sizeof(tag_uuid_input), "%s-%s-%s",
info->driver, info->group, tag->name);
generate_fixed_uuid(tag_uuid_input, tag_id);
json_t *tag_obj = json_object();
json_object_set_new(tag_obj, "id", json_string(tag_id));
json_object_set_new(tag_obj, "name", json_string(tag->name));
json_object_set_new(tag_obj, "type", json_integer(tag->type));
json_object_set_new(tag_obj, "category", json_integer(3));
json_object_set_new(tag_obj, "leaf", json_true());
json_object_set_new(tag_obj, "children", json_array());
json_array_append_new(tag_array, tag_obj);
}
}
}
json_t *final_obj = json_object();
json_object_set_new(final_obj, "error", json_integer(0));
json_object_set_new(final_obj, "children", node_array);
neu_json_elem_t elems[] = { {
.name = NULL,
.t = NEU_JSON_OBJECT,
.v.val_object = final_obj,
} };
int ret = neu_json_encode_field(json_object_param, elems,
NEU_JSON_ELEM_SIZE(elems));
json_decref(final_obj);
src/parser/neu_json_datalayers.h
+2
View File
@@ -47,6 +47,8 @@ int neu_json_encode_datalayers_get_groups_resp(void *json_object, void *param);
int neu_json_encode_datalayers_get_tags_resp(void *json_object, void *param);
int neu_json_encode_datalayers_get_tag_resp(void *json_object, void *param);
#ifdef __cplusplus
}
#endif