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

mqtt read group filter tags

Browse Source
This commit is contained in:
fengzero
2024-11-08 06:42:40 +00:00
parent 141b3f56b1
commit ebc91b0ce1
9 changed files with 105 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/json/json.h
+7
View File
@@ -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 {
include/neuron/json/neu_json_rw.h
+7 -5
View File
@@ -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);
include/neuron/msg.h
+13 -5
View File
@@ -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 {
plugins/mqtt/mqtt_handle.c
+8 -2
View File
@@ -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");
src/adapter/driver/driver.c
+3 -2
View File
@@ -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);
}
src/base/group.c
+21 -7
View File
@@ -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);
src/base/group.h
+2 -1
View File
@@ -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,
src/parser/neu_json_rw.c
+20 -3
View File
@@ -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);
}
src/utils/json.c
+24
View File
@@ -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;