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

Merge pull request #1792 from eeff/eeff/dev

Browse Source 
feat(core): read group support filter
This commit is contained in:
fengzeroz
2024-01-03 17:47:12 +08:00
committed by GitHub
parent 595d7d4456 b4041b0be3
commit aa3462efe5
10 changed files with 156 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/adapter/driver/driver.c
+1 -1
View File
@@ -406,7 +406,7 @@ 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_get_read_tag(group);
UT_array *tags = neu_group_query_read_tag(group, cmd->name, cmd->desc);
utarray_new(resp.tags, neu_resp_tag_value_meta_icd());
src/base/group.c
+70 -24
View File
@@ -44,7 +44,6 @@ struct neu_group {
};
static UT_array *to_array(tag_elem_t *tags);
static UT_array *to_read_array(tag_elem_t *tags);
static void split_static_array(tag_elem_t *tags, UT_array **static_tags,
UT_array **other_tags);
static void update_timestamp(neu_group_t *group);
@@ -192,19 +191,84 @@ UT_array *neu_group_get_tag(neu_group_t *group)
return array;
}
UT_array *neu_group_query_tag(neu_group_t *group, const char *name)
static inline UT_array *
filter_tags(tag_elem_t *tags,
bool (*predicate)(const neu_datatag_t *, void *data), void *data)
{
tag_elem_t *el = NULL, *tmp = NULL;
UT_array * array = NULL;
pthread_mutex_lock(&group->mtx);
utarray_new(array, neu_tag_get_icd());
HASH_ITER(hh, group->tags, el, tmp)
HASH_ITER(hh, tags, el, tmp)
{
if (strstr(el->tag->name, name) != NULL) {
if (predicate(el->tag, data)) {
utarray_push_back(array, el->tag);
}
}
return array;
}
static inline bool is_readable(const neu_datatag_t *tag, void *data)
{
(void) data;
return neu_tag_attribute_test(tag, NEU_ATTRIBUTE_READ) ||
neu_tag_attribute_test(tag, NEU_ATTRIBUTE_SUBSCRIBE) ||
neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC);
}
static inline bool name_contains(const neu_datatag_t *tag, void *data)
{
const char *name = data;
return strstr(tag->name, name) != NULL;
}
static inline bool description_contains(const neu_datatag_t *tag, void *data)
{
const char *str = data;
return tag->description && strstr(tag->description, str) != NULL;
}
struct query {
char *name;
char *desc;
};
static inline bool match_query(const neu_datatag_t *tag, void *data)
{
struct query *q = data;
return (!q->name || name_contains(tag, q->name)) &&
(!q->desc || description_contains(tag, q->desc));
}
static inline bool is_readable_and_match_query(const neu_datatag_t *tag,
void * data)
{
return is_readable(tag, NULL) && match_query(tag, data);
}
UT_array *neu_group_query_tag(neu_group_t *group, const char *name)
{
UT_array *array = NULL;
pthread_mutex_lock(&group->mtx);
array = filter_tags(group->tags, name_contains, (void *) name);
pthread_mutex_unlock(&group->mtx);
return array;
}
UT_array *neu_group_query_read_tag(neu_group_t *group, const char *name,
const char *desc)
{
UT_array * array = NULL;
struct query q = {
.name = (char *) name,
.desc = (char *) desc,
};
pthread_mutex_lock(&group->mtx);
array = filter_tags(group->tags, is_readable_and_match_query, &q);
pthread_mutex_unlock(&group->mtx);
return array;
@@ -215,7 +279,7 @@ UT_array *neu_group_get_read_tag(neu_group_t *group)
UT_array *array = NULL;
pthread_mutex_lock(&group->mtx);
array = to_read_array(group->tags);
array = filter_tags(group->tags, is_readable, NULL);
pthread_mutex_unlock(&group->mtx);
return array;
@@ -290,24 +354,6 @@ static UT_array *to_array(tag_elem_t *tags)
return array;
}
static UT_array *to_read_array(tag_elem_t *tags)
{
tag_elem_t *el = NULL, *tmp = NULL;
UT_array * array = NULL;
utarray_new(array, neu_tag_get_icd());
HASH_ITER(hh, tags, el, tmp)
{
if (neu_tag_attribute_test(el->tag, NEU_ATTRIBUTE_READ) ||
neu_tag_attribute_test(el->tag, NEU_ATTRIBUTE_SUBSCRIBE) ||
neu_tag_attribute_test(el->tag, NEU_ATTRIBUTE_STATIC)) {
utarray_push_back(array, el->tag);
}
}
return array;
}
static void split_static_array(tag_elem_t *tags, UT_array **static_tags,
UT_array **other_tags)
{
src/base/group.h
+2
View File
@@ -41,6 +41,8 @@ int neu_group_del_tag(neu_group_t *group, const char *tag_name);
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);
uint16_t neu_group_tag_size(const neu_group_t *group);
neu_datatag_t *neu_group_find_tag(neu_group_t *group, const char *tag);
void neu_group_split_static_tags(neu_group_t *group, UT_array **static_tags,
src/base/msg_internal.h
-3
View File
@@ -149,9 +149,6 @@ static inline neu_msg_t *neu_msg_new(neu_reqresp_type_e t, void *ctx,
case NEU_REQ_GET_NODES_STATE:
body_size = neu_reqresp_size(NEU_RESP_GET_NODES_STATE);
break;
case NEU_REQ_READ_GROUP:
body_size = neu_reqresp_size(NEU_RESP_READ_GROUP);
break;
default:
body_size = data_size;
}
src/parser/neu_json_rw.c
+30
View File
@@ -430,6 +430,11 @@ int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result)
.t = NEU_JSON_BOOL,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
},
{
.name = "query",
.t = NEU_JSON_OBJECT,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
},
};
ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
req_elems);
@@ -441,11 +446,36 @@ int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result)
req->group = req_elems[1].v.val_str;
req->sync = req_elems[2].v.val_bool;
neu_json_elem_t query_elems[] = {
{
.name = "name",
.t = NEU_JSON_STR,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
},
{
.name = "description",
.t = NEU_JSON_STR,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
},
};
if (req_elems[3].v.val_object) {
ret = neu_json_decode_by_json(req_elems[3].v.val_object,
NEU_JSON_ELEM_SIZE(query_elems),
query_elems);
if (ret != 0) {
goto error;
}
req->name = query_elems[0].v.val_str;
req->desc = query_elems[1].v.val_str;
}
*result = req;
neu_json_decode_free(json_obj);
return ret;
error:
free(query_elems[0].v.val_str);
free(query_elems[1].v.val_str);
free(req_elems[0].v.val_str);
free(req_elems[1].v.val_str);
free(req);