From ebc91b0ce1baf29575a83229f293c2ebd359b1f2 Mon Sep 17 00:00:00 2001 From: fengzero Date: Fri, 8 Nov 2024 06:42:40 +0000 Subject: [PATCH] mqtt read group filter tags --- include/neuron/json/json.h | 7 +++++++ include/neuron/json/neu_json_rw.h | 12 +++++++----- include/neuron/msg.h | 18 +++++++++++++----- plugins/mqtt/mqtt_handle.c | 10 ++++++++-- src/adapter/driver/driver.c | 5 +++-- src/base/group.c | 28 +++++++++++++++++++++------- src/base/group.h | 3 ++- src/parser/neu_json_rw.c | 23 ++++++++++++++++++++--- src/utils/json.c | 24 ++++++++++++++++++++++++ 9 files changed, 105 insertions(+), 25 deletions(-) diff --git a/include/neuron/json/json.h b/include/neuron/json/json.h index a9a1a617..0e55ce76 100644 --- a/include/neuron/json/json.h +++ b/include/neuron/json/json.h @@ -48,6 +48,7 @@ typedef enum neu_json_type { NEU_JSON_ARRAY_FLOAT, NEU_JSON_ARRAY_DOUBLE, NEU_JSON_ARRAY_BOOL, + NEU_JSON_ARRAY_STR, NEU_JSON_VALUE = NEU_JSON_UNDEFINE } neu_json_type_e; @@ -106,6 +107,11 @@ typedef struct { uint8_t length; } neu_json_value_array_bool_t; +typedef struct { + char ** p_strs; + uint16_t length; +} neu_json_value_array_str_t; + typedef union neu_json_value { int64_t val_int; uint8_t val_bit; @@ -125,6 +131,7 @@ typedef union neu_json_value { neu_json_value_array_float_t val_array_float; neu_json_value_array_double_t val_array_double; neu_json_value_array_bool_t val_array_bool; + neu_json_value_array_str_t val_array_str; } neu_json_value_u; typedef enum neu_json_attribute { diff --git a/include/neuron/json/neu_json_rw.h b/include/neuron/json/neu_json_rw.h index 0ecabf39..ca22a89a 100644 --- a/include/neuron/json/neu_json_rw.h +++ b/include/neuron/json/neu_json_rw.h @@ -149,11 +149,13 @@ int neu_json_decode_write(char *buf, neu_json_write_t **result); void neu_json_decode_write_free(neu_json_write_t *req); typedef struct { - char *group; - char *node; - char *name; - char *desc; - bool sync; + char * group; + char * node; + char * name; + char * desc; + bool sync; + int n_tags; + char **tags; } neu_json_read_req_t; int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result); diff --git a/include/neuron/msg.h b/include/neuron/msg.h index ae8c8389..b203c9dc 100644 --- a/include/neuron/msg.h +++ b/include/neuron/msg.h @@ -637,11 +637,13 @@ typedef struct { } neu_resp_get_nodes_state_t, neu_reqresp_nodes_state_t; typedef struct neu_req_read_group { - char *driver; - char *group; - char *name; - char *desc; - bool sync; + char * driver; + char * group; + char * name; + char * desc; + bool sync; + uint16_t n_tag; + char ** tags; } neu_req_read_group_t; static inline void neu_req_read_group_fini(neu_req_read_group_t *req) @@ -650,6 +652,12 @@ static inline void neu_req_read_group_fini(neu_req_read_group_t *req) free(req->group); free(req->name); free(req->desc); + if (req->n_tag > 0) { + for (uint16_t i = 0; i < req->n_tag; i++) { + free(req->tags[i]); + } + free(req->tags); + } } typedef struct neu_req_read_group_paginate { diff --git a/plugins/mqtt/mqtt_handle.c b/plugins/mqtt/mqtt_handle.c index b1de56b2..624f71ee 100644 --- a/plugins/mqtt/mqtt_handle.c +++ b/plugins/mqtt/mqtt_handle.c @@ -198,8 +198,14 @@ static inline int send_read_req(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt, cmd.driver = req->node; cmd.group = req->group; cmd.sync = req->sync; - req->node = NULL; // ownership moved - req->group = NULL; // ownership moved + cmd.n_tag = req->n_tags; + cmd.tags = req->tags; + + req->node = NULL; // ownership moved + req->group = NULL; // ownership moved + req->n_tags = 0; + req->tags = NULL; + if (0 != neu_plugin_op(plugin, header, &cmd)) { neu_req_read_group_fini(&cmd); plog_error(plugin, "neu_plugin_op(NEU_REQ_READ_GROUP) fail"); diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index d1374b15..5d464f8c 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -568,7 +568,8 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver, neu_resp_read_group_t resp = { 0 }; neu_group_t * group = g->group; - UT_array *tags = neu_group_query_read_tag(group, cmd->name, cmd->desc); + UT_array *tags = neu_group_query_read_tag(group, cmd->name, cmd->desc, + cmd->n_tag, cmd->tags); utarray_new(resp.tags, neu_resp_tag_value_meta_icd()); @@ -651,7 +652,7 @@ void neu_adapter_driver_read_group_paginate(neu_adapter_driver_t *driver, group, cmd->name, cmd->desc, cmd->current_page, cmd->page_size, &resp.total_count); } else { - tags = neu_group_query_read_tag(group, cmd->name, cmd->desc); + tags = neu_group_query_read_tag(group, cmd->name, cmd->desc, 0, NULL); resp.total_count = utarray_len(tags); } diff --git a/src/base/group.c b/src/base/group.c index ce1fbf65..31df2d88 100644 --- a/src/base/group.c +++ b/src/base/group.c @@ -256,15 +256,26 @@ static inline bool description_contains(const neu_datatag_t *tag, void *data) } struct query { - char *name; - char *desc; + char * name; + char * desc; + uint16_t n_tagname; + char ** tagnames; }; static inline bool match_query(const neu_datatag_t *tag, void *data) { - struct query *q = data; - return (!q->name || name_contains(tag, q->name)) && + struct query *q = data; + bool filter = (!q->name || name_contains(tag, q->name)) && (!q->desc || description_contains(tag, q->desc)); + if (filter && q->n_tagname > 0) { + for (uint16_t i = 0; i < q->n_tagname; i++) { + if (strcmp(tag->name, q->tagnames[i]) == 0) { + return true; + } + } + return false; + } + return filter; } static inline bool is_readable_and_match_query(const neu_datatag_t *tag, @@ -285,12 +296,15 @@ UT_array *neu_group_query_tag(neu_group_t *group, const char *name) } UT_array *neu_group_query_read_tag(neu_group_t *group, const char *name, - const char *desc) + const char *desc, uint16_t n_tagname, + char **tagnames) { UT_array * array = NULL; struct query q = { - .name = (char *) name, - .desc = (char *) desc, + .name = (char *) name, + .desc = (char *) desc, + .n_tagname = n_tagname, + .tagnames = tagnames, }; pthread_mutex_lock(&group->mtx); diff --git a/src/base/group.h b/src/base/group.h index 07388971..6b20772f 100644 --- a/src/base/group.h +++ b/src/base/group.h @@ -42,7 +42,8 @@ UT_array * neu_group_get_tag(neu_group_t *group); UT_array * neu_group_query_tag(neu_group_t *group, const char *name); UT_array * neu_group_get_read_tag(neu_group_t *group); UT_array * neu_group_query_read_tag(neu_group_t *group, const char *name, - const char *desc); + const char *desc, uint16_t n_tagname, + char **tagnames); UT_array * neu_group_query_read_tag_paginate(neu_group_t *group, const char *name, const char *desc, int current_page, int page_size, diff --git a/src/parser/neu_json_rw.c b/src/parser/neu_json_rw.c index f14e7e39..2e3939fb 100644 --- a/src/parser/neu_json_rw.c +++ b/src/parser/neu_json_rw.c @@ -670,6 +670,13 @@ int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result) .t = NEU_JSON_OBJECT, .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, }, + { + .name = "tags", + .t = NEU_JSON_ARRAY_STR, + .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL, + .v.val_array_str.length = 0, + .v.val_array_str.p_strs = NULL, + }, }; neu_json_elem_t query_elems[] = { @@ -691,9 +698,11 @@ int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result) goto error; } - req->node = req_elems[0].v.val_str; - req->group = req_elems[1].v.val_str; - req->sync = req_elems[2].v.val_bool; + req->node = req_elems[0].v.val_str; + req->group = req_elems[1].v.val_str; + req->sync = req_elems[2].v.val_bool; + req->n_tags = req_elems[4].v.val_array_str.length; + req->tags = req_elems[4].v.val_array_str.p_strs; if (req_elems[3].v.val_object) { ret = neu_json_decode_by_json(req_elems[3].v.val_object, @@ -726,6 +735,14 @@ void neu_json_decode_read_req_free(neu_json_read_req_t *req) { free(req->group); free(req->node); + if (req->n_tags > 0) { + for (int i = 0; i < req->n_tags; i++) { + if (req->tags[i] != NULL) { + free(req->tags[i]); + } + } + free(req->tags); + } free(req); } diff --git a/src/utils/json.c b/src/utils/json.c index ef4e2424..8cd6890f 100644 --- a/src/utils/json.c +++ b/src/utils/json.c @@ -296,6 +296,10 @@ static int decode_object(json_t *root, neu_json_elem_t *ele) ele->t = NEU_JSON_ARRAY_INT64; json_array_foreach(ob, index, value) { + if (json_is_string(value)) { + ele->t = NEU_JSON_ARRAY_STR; + break; + } if (json_is_real(value)) { ele->t = NEU_JSON_ARRAY_DOUBLE; break; @@ -344,6 +348,26 @@ static int decode_object(json_t *root, neu_json_elem_t *ele) case NEU_JSON_BOOL: ele->v.val_bool = json_boolean_value(ob); break; + case NEU_JSON_ARRAY_STR: { + json_t *value = NULL; + + ele->v.val_array_str.length = json_array_size(ob); + if (ele->v.val_array_str.length > 0) { + int index = 0; + + ele->v.val_array_str.p_strs = + calloc(ele->v.val_array_str.length, sizeof(char *)); + json_array_foreach(ob, index, value) + { + const char *str_val = json_string_value(value); + if (str_val != NULL) { + ele->v.val_array_str.p_strs[index] = strdup(str_val); + } + } + } + + break; + } case NEU_JSON_ARRAY_BOOL: { json_t *value = NULL;