Merge pull request #1803 from fengzeroz/v2.7

version: 2.7.1

.github/workflows/ci.yml

@@ -8,7 +8,7 @@ on:
       - published
 
 env:
-  VERSION: 2.7.0
+  VERSION: 2.7.1
   UI_VERSION: 2.6.3
   BRANCH: v2.7
 

CMakeLists.txt

@@ -128,6 +128,7 @@ set(NEURON_SOURCES
     src/core/node_manager.c
     src/core/template_manager.c
     src/core/storage.c
+    src/adapter/msg_q.c
     src/adapter/storage.c
     src/adapter/adapter.c
     src/adapter/driver/cache.c

include/neuron/define.h

@@ -24,7 +24,7 @@
 
 #define NEU_VERSION_MAJOR 2
 #define NEU_VERSION_MINOR 7
-#define NEU_VERSION_FIX 0
+#define NEU_VERSION_FIX 1
 
 #define NEU_GET_VERSION_MAJOR(V) ((0xffff0000 & V) >> 16)
 #define NEU_GET_VERSION_MINOR(V) ((0x0000ff00 & V) >> 8)
@@ -34,6 +34,7 @@
 #define NEU_TAG_NAME_LEN 64
 #define NEU_TAG_ADDRESS_LEN 128
 #define NEU_TAG_DESCRIPTION_LEN 128
+#define NEU_GROUP_MAX_PER_NODE 512
 #define NEU_GROUP_NAME_LEN 128
 #define NEU_GROUP_INTERVAL_LIMIT 100
 #define NEU_DEFAULT_GROUP_INTERVAL 100

include/neuron/errcodes.h

@@ -70,6 +70,7 @@ typedef enum {
     NEU_ERR_GROUP_PARAMETER_INVALID  = 2105,
     NEU_ERR_GROUP_NOT_EXIST          = 2106,
     NEU_ERR_GROUP_NAME_TOO_LONG      = 2107,
+    NEU_ERR_GROUP_MAX_GROUPS         = 2108,
 
     NEU_ERR_TAG_NOT_EXIST              = 2201,
     NEU_ERR_TAG_NAME_CONFLICT          = 2202,

include/neuron/json/neu_json_rw.h

@@ -126,6 +126,8 @@ void neu_json_decode_write_free(neu_json_write_t *req);
 typedef struct {
     char *group;
     char *node;
+    char *name;
+    char *desc;
     bool  sync;
 } neu_json_read_req_t;
 

include/neuron/msg.h

@@ -515,7 +515,13 @@ neu_req_subscribe_groups_fini(neu_req_subscribe_groups_t *req)
 
 typedef struct neu_req_get_subscribe_group {
     char app[NEU_NODE_NAME_LEN];
-} neu_req_get_subscribe_group_t, neu_req_get_sub_driver_tags_t;
+    char driver[NEU_NODE_NAME_LEN];
+    char group[NEU_GROUP_NAME_LEN];
+} neu_req_get_subscribe_group_t;
+
+typedef struct {
+    char app[NEU_NODE_NAME_LEN];
+} neu_req_get_sub_driver_tags_t;
 
 typedef struct neu_resp_subscribe_info {
     char  app[NEU_NODE_NAME_LEN];
@@ -779,6 +785,8 @@ static inline void neu_req_inst_templates_fini(neu_req_inst_templates_t *req)
 typedef struct neu_req_read_group {
     char *driver;
     char *group;
+    char *name;
+    char *desc;
     bool  sync;
 } neu_req_read_group_t;
 
@@ -786,6 +794,8 @@ static inline void neu_req_read_group_fini(neu_req_read_group_t *req)
 {
     free(req->driver);
     free(req->group);
+    free(req->name);
+    free(req->desc);
 }
 
 typedef struct neu_req_write_tag {

plugins/restful/group_config_handle.c

@@ -356,23 +356,41 @@ void handle_grp_get_subscribe(nng_aio *aio)
 {
     int                           ret    = 0;
     neu_plugin_t *                plugin = neu_rest_get_plugin();
-    char                          node_name[NEU_NODE_NAME_LEN] = { 0 };
-    neu_req_get_subscribe_group_t cmd                          = { 0 };
-    neu_reqresp_head_t            header                       = {
+    neu_req_get_subscribe_group_t cmd    = { 0 };
+    neu_reqresp_head_t            header = {
         .ctx  = aio,
         .type = NEU_REQ_GET_SUBSCRIBE_GROUP,
     };
 
     NEU_VALIDATE_JWT(aio);
 
-    if (neu_http_get_param_str(aio, "app", node_name, sizeof(node_name)) <= 0) {
+    // required parameter
+    ret = neu_http_get_param_str(aio, "app", cmd.app, sizeof(cmd.app));
+    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;
+    }
+
+    // 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;
     }
 
-    strcpy(cmd.app, node_name);
     ret = neu_plugin_op(plugin, header, &cmd);
     if (ret != 0) {
         NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {

plugins/restful/rw_handle.c

@@ -43,6 +43,8 @@ void handle_read(nng_aio *aio)
 
             cmd.driver = req->node;
             cmd.group  = req->group;
+            cmd.name   = req->name;
+            cmd.desc   = req->desc;
             cmd.sync   = req->sync;
             req->node  = NULL;
             req->group = NULL;

src/adapter/adapter.c

@@ -39,11 +39,12 @@
 #include "plugin.h"
 #include "storage.h"
 
-static int adapter_trans_data(enum neu_event_io_type type, int fd,
-                              void *usr_data);
-static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data);
-static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
-                           void *data);
+static void *adapter_consumer(void *arg);
+static int   adapter_trans_data(enum neu_event_io_type type, int fd,
+                                void *usr_data);
+static int   adapter_loop(enum neu_event_io_type type, int fd, void *usr_data);
+static int   adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
+                             void *data);
 static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header,
                             void *data);
 static int adapter_responseto(neu_adapter_t *     adapter,
@@ -110,6 +111,26 @@ void neu_adapter_set_error(int error)
     create_adapter_error = error;
 }
 
+static void *adapter_consumer(void *arg)
+{
+    neu_adapter_t *adapter = (neu_adapter_t *) arg;
+    while (1) {
+        neu_msg_t *         msg    = NULL;
+        uint32_t            n      = adapter_msg_q_pop(adapter->msg_q, &msg);
+        neu_reqresp_head_t *header = neu_msg_get_header(msg);
+
+        nlog_debug("adapter(%s) recv msg from: %s %p, type: %s, %u",
+                   adapter->name, header->sender, header->ctx,
+                   neu_reqresp_type_string(header->type), n);
+        adapter->module->intf_funs->request(
+            adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
+        neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
+        neu_msg_free(msg);
+    }
+
+    return NULL;
+}
+
 static inline zlog_category_t *get_log_category(const char *node)
 {
     char name[NEU_NODE_NAME_LEN] = { 0 };
@@ -183,6 +204,9 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load)
         break;
     case NEU_NA_TYPE_NDRIVER:
     case NEU_NA_TYPE_APP: {
+        adapter->msg_q = adapter_msg_q_new(adapter->name, 1024);
+        pthread_create(&adapter->consumer_tid, NULL, adapter_consumer,
+                       (void *) adapter);
         while (true) {
             uint16_t           port  = neu_manager_get_port();
             struct sockaddr_in local = {
@@ -615,11 +639,18 @@ static int adapter_trans_data(enum neu_event_io_type type, int fd,
         return 0;
     }
 
+    if (header->type == NEU_REQRESP_TRANS_DATA) {
+        if (adapter_msg_q_push(adapter->msg_q, msg) < 0) {
+            nlog_warn("adapter: %s trans data msg q is full, drop msg",
+                      adapter->name);
+            neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
+            neu_msg_free(msg);
+        }
+        return 0;
+    }
+
     adapter->module->intf_funs->request(
         adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
-    if (header->type == NEU_REQRESP_TRANS_DATA) {
-        neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
-    }
     neu_msg_free(msg);
     return 0;
 }
@@ -1050,9 +1081,14 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_ADD_GTAG: {
         neu_req_add_gtag_t *cmd  = (neu_req_add_gtag_t *) &header[1];
         neu_resp_add_tag_t  resp = { 0 };
-
         if (adapter->module->type != NEU_NA_TYPE_DRIVER) {
             resp.error = NEU_ERR_GROUP_NOT_ALLOW;
+        } else if (neu_adapter_driver_new_group_count(
+                       (neu_adapter_driver_t *) adapter, cmd) +
+                       neu_adapter_driver_group_count(
+                           (neu_adapter_driver_t *) adapter) >
+                   NEU_GROUP_MAX_PER_NODE) {
+            resp.error = NEU_ERR_GROUP_MAX_GROUPS;
         } else {
             if (validate_gtags(adapter, cmd, &resp) == 0 &&
                 try_add_gtags(adapter, cmd, &resp) == 0 &&
@@ -1340,6 +1376,13 @@ void neu_adapter_destroy(neu_adapter_t *adapter)
         neu_node_metrics_free(adapter->metrics);
     }
 
+    if (adapter->consumer_tid != 0) {
+        pthread_cancel(adapter->consumer_tid);
+    }
+    if (adapter->msg_q != NULL) {
+        adapter_msg_q_free(adapter->msg_q);
+    }
+
     char *setting = NULL;
     if (adapter_load_setting(adapter->name, &setting) != 0) {
         remove_logs(adapter->name);

src/adapter/adapter_internal.h

@@ -26,6 +26,7 @@
 
 #include "adapter_info.h"
 #include "core/manager.h"
+#include "msg_q.h"
 
 struct neu_adapter {
     char *name;
@@ -45,6 +46,9 @@ struct neu_adapter {
     int control_fd;
     int trans_data_fd;
 
+    adapter_msg_q_t *msg_q;
+    pthread_t        consumer_tid;
+
     uint16_t trans_data_port;
 
     neu_events_t *events;

src/adapter/driver/driver.c

@@ -418,7 +418,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());
 
@@ -1105,6 +1105,29 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
     return ret;
 }
 
+uint16_t neu_adapter_driver_group_count(neu_adapter_driver_t *driver)
+{
+    u_int16_t num_groups = 0;
+    if (driver && driver->groups) {
+        num_groups = HASH_COUNT(driver->groups);
+    }
+    return num_groups;
+}
+
+uint16_t neu_adapter_driver_new_group_count(neu_adapter_driver_t *driver,
+                                            neu_req_add_gtag_t *  cmd)
+{
+    uint16_t new_groups_count = 0;
+    for (int i = 0; i < cmd->n_group; i++) {
+        group_t *group_in_driver;
+        HASH_FIND_STR(driver->groups, cmd->groups[i].group, group_in_driver);
+        if (!group_in_driver) {
+            new_groups_count++;
+        }
+    }
+    return new_groups_count;
+}
+
 static group_t *find_group(neu_adapter_driver_t *driver, const char *name)
 {
     group_t *find = NULL;

src/adapter/driver/driver_internal.h

@@ -53,6 +53,9 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver,
 int neu_adapter_driver_group_exist(neu_adapter_driver_t *driver,
                                    const char *          name);
 UT_array *neu_adapter_driver_get_group(neu_adapter_driver_t *driver);
+uint16_t  neu_adapter_driver_group_count(neu_adapter_driver_t *driver);
+uint16_t  neu_adapter_driver_new_group_count(neu_adapter_driver_t *driver,
+                                             neu_req_add_gtag_t *  cmd);
 
 int neu_adapter_driver_try_del_tag(neu_adapter_driver_t *driver, int n_tag);
 int neu_adapter_driver_try_add_tag(neu_adapter_driver_t *driver,

src/adapter/msg_q.c

@@ -0,0 +1,116 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2024 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ **/
+#include <pthread.h>
+
+#include "utils/log.h"
+#include "utils/utextend.h"
+
+#include "msg_q.h"
+
+struct item {
+    neu_msg_t *  msg;
+    struct item *prev, *next;
+};
+
+struct adapter_msg_q {
+    struct item *list;
+    uint32_t     max;
+    uint32_t     current;
+    char *       name;
+
+    pthread_mutex_t mtx;
+    pthread_cond_t  cond;
+};
+
+adapter_msg_q_t *adapter_msg_q_new(const char *name, uint32_t size)
+{
+    struct adapter_msg_q *q = calloc(1, sizeof(struct adapter_msg_q));
+
+    pthread_mutex_init(&q->mtx, NULL);
+    pthread_cond_init(&q->cond, NULL);
+    q->max     = size;
+    q->name    = strdup(name);
+    q->current = 0;
+
+    return q;
+}
+
+void adapter_msg_q_free(adapter_msg_q_t *q)
+{
+    struct item *tmp = NULL, *elt = NULL;
+    nlog_warn("app: %s, drop %u msg", q->name, q->current);
+    pthread_mutex_destroy(&q->mtx);
+    pthread_cond_destroy(&q->cond);
+
+    DL_FOREACH_SAFE(q->list, elt, tmp)
+    {
+        DL_DELETE(q->list, elt);
+        neu_reqresp_head_t *header = neu_msg_get_header(elt->msg);
+        neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
+        neu_msg_free(elt->msg);
+        free(elt);
+    }
+    free(q->name);
+    free(q);
+}
+
+int adapter_msg_q_push(adapter_msg_q_t *q, neu_msg_t *msg)
+{
+    int ret = -1;
+    pthread_mutex_lock(&q->mtx);
+    if (q->current < q->max) {
+        q->current += 1;
+        struct item *elt = calloc(1, sizeof(struct item));
+        elt->msg         = msg;
+        DL_APPEND(q->list, elt);
+        ret = 0;
+    }
+    pthread_mutex_unlock(&q->mtx);
+
+    if (ret == -1) {
+        nlog_warn("app: %s, msg q is full, %u(%u)", q->name, q->current,
+                  q->max);
+    } else {
+        pthread_cond_signal(&q->cond);
+    }
+
+    return ret;
+}
+
+uint32_t adapter_msg_q_pop(adapter_msg_q_t *q, neu_msg_t **p_data)
+{
+    uint32_t ret = 0;
+
+    pthread_mutex_lock(&q->mtx);
+    while (q->current == 0) {
+        pthread_cond_wait(&q->cond, &q->mtx);
+    }
+    struct item *elt = DL_LAST(q->list);
+
+    if (elt != NULL) {
+        DL_DELETE(q->list, elt);
+        *p_data = elt->msg;
+        free(elt);
+        q->current -= 1;
+        ret = q->current;
+    }
+
+    pthread_mutex_unlock(&q->mtx);
+    return ret;
+}

src/adapter/msg_q.h

@@ -0,0 +1,36 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2024 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ **/
+
+#ifndef ADAPTER_MSG_Q_H
+#define ADAPTER_MSG_Q_H
+
+#include <stdint.h>
+
+#include "base/msg_internal.h"
+#include "msg.h"
+
+typedef struct adapter_msg_q adapter_msg_q_t;
+
+adapter_msg_q_t *adapter_msg_q_new(const char *name, uint32_t size);
+void             adapter_msg_q_free(adapter_msg_q_t *q);
+
+int      adapter_msg_q_push(adapter_msg_q_t *q, neu_msg_t *msg);
+uint32_t adapter_msg_q_pop(adapter_msg_q_t *q, neu_msg_t **p_data);
+
+#endif

src/base/group.c

@@ -16,6 +16,7 @@
  * along with this program; if not, write to the Free Software Foundation,
  * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  **/
+#include <pthread.h>
 #include <string.h>
 #include <sys/time.h>
 
@@ -38,11 +39,11 @@ struct neu_group {
     tag_elem_t *tags;
     uint32_t    interval;
 
-    int64_t timestamp;
+    int64_t         timestamp;
+    pthread_mutex_t mtx;
 };
 
 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);
@@ -53,6 +54,7 @@ neu_group_t *neu_group_new(const char *name, uint32_t interval)
 
     group->name     = strdup(name);
     group->interval = interval;
+    pthread_mutex_init(&group->mtx, NULL);
 
     return group;
 }
@@ -61,6 +63,7 @@ void neu_group_destroy(neu_group_t *group)
 {
     tag_elem_t *el = NULL, *tmp = NULL;
 
+    pthread_mutex_lock(&group->mtx);
     HASH_ITER(hh, group->tags, el, tmp)
     {
         HASH_DEL(group->tags, el);
@@ -68,7 +71,9 @@ void neu_group_destroy(neu_group_t *group)
         neu_tag_free(el->tag);
         free(el);
     }
+    pthread_mutex_unlock(&group->mtx);
 
+    pthread_mutex_destroy(&group->mtx);
     free(group->name);
     free(group);
 }
@@ -118,8 +123,10 @@ int neu_group_add_tag(neu_group_t *group, const neu_datatag_t *tag)
 {
     tag_elem_t *el = NULL;
 
+    pthread_mutex_lock(&group->mtx);
     HASH_FIND_STR(group->tags, tag->name, el);
     if (el != NULL) {
+        pthread_mutex_unlock(&group->mtx);
         return NEU_ERR_TAG_NAME_CONFLICT;
     }
 
@@ -129,6 +136,7 @@ int neu_group_add_tag(neu_group_t *group, const neu_datatag_t *tag)
 
     HASH_ADD_STR(group->tags, name, el);
     update_timestamp(group);
+    pthread_mutex_unlock(&group->mtx);
 
     return 0;
 }
@@ -138,6 +146,7 @@ int neu_group_update_tag(neu_group_t *group, const neu_datatag_t *tag)
     tag_elem_t *el  = NULL;
     int         ret = NEU_ERR_TAG_NOT_EXIST;
 
+    pthread_mutex_lock(&group->mtx);
     HASH_FIND_STR(group->tags, tag->name, el);
     if (el != NULL) {
         neu_tag_copy(el->tag, tag);
@@ -145,6 +154,7 @@ int neu_group_update_tag(neu_group_t *group, const neu_datatag_t *tag)
         update_timestamp(group);
         ret = NEU_ERR_SUCCESS;
     }
+    pthread_mutex_unlock(&group->mtx);
 
     return ret;
 }
@@ -154,6 +164,7 @@ int neu_group_del_tag(neu_group_t *group, const char *tag_name)
     tag_elem_t *el  = NULL;
     int         ret = NEU_ERR_TAG_NOT_EXIST;
 
+    pthread_mutex_lock(&group->mtx);
     HASH_FIND_STR(group->tags, tag_name, el);
     if (el != NULL) {
         HASH_DEL(group->tags, el);
@@ -164,28 +175,33 @@ int neu_group_del_tag(neu_group_t *group, const char *tag_name)
         update_timestamp(group);
         ret = NEU_ERR_SUCCESS;
     }
+    pthread_mutex_unlock(&group->mtx);
 
     return ret;
 }
 
-UT_array *neu_group_get_tag(const neu_group_t *group)
+UT_array *neu_group_get_tag(neu_group_t *group)
 {
     UT_array *array = NULL;
 
+    pthread_mutex_lock(&group->mtx);
     array = to_array(group->tags);
+    pthread_mutex_unlock(&group->mtx);
 
     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;
 
     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);
         }
     }
@@ -193,11 +209,79 @@ UT_array *neu_group_query_tag(neu_group_t *group, const char *name)
     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 ||
+        (tag->description != NULL && strstr(tag->description, 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;
+}
+
 UT_array *neu_group_get_read_tag(neu_group_t *group)
 {
     UT_array *array = NULL;
 
-    array = to_read_array(group->tags);
+    pthread_mutex_lock(&group->mtx);
+    array = filter_tags(group->tags, is_readable, NULL);
+    pthread_mutex_unlock(&group->mtx);
 
     return array;
 }
@@ -216,10 +300,12 @@ neu_datatag_t *neu_group_find_tag(neu_group_t *group, const char *tag)
     tag_elem_t *   find   = NULL;
     neu_datatag_t *result = NULL;
 
+    pthread_mutex_lock(&group->mtx);
     HASH_FIND_STR(group->tags, tag, find);
     if (find != NULL) {
         result = neu_tag_dup(find->tag);
     }
+    pthread_mutex_unlock(&group->mtx);
 
     return result;
 }
@@ -269,24 +355,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

@@ -38,9 +38,11 @@ int          neu_group_update(neu_group_t *group, uint32_t interval);
 int          neu_group_add_tag(neu_group_t *group, const neu_datatag_t *tag);
 int          neu_group_update_tag(neu_group_t *group, const neu_datatag_t *tag);
 int          neu_group_del_tag(neu_group_t *group, const char *tag_name);
-UT_array *   neu_group_get_tag(const neu_group_t *group);
+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

@@ -182,9 +182,6 @@ static inline neu_msg_t *neu_msg_new(neu_reqresp_type_e t, void *ctx,
     case NEU_REQ_GET_TEMPLATES:
         body_size = neu_reqresp_size(NEU_RESP_GET_TEMPLATES);
         break;
-    case NEU_REQ_READ_GROUP:
-        body_size = neu_reqresp_size(NEU_RESP_READ_GROUP);
-        break;
     default:
         body_size = data_size;
     }

src/core/manager.c

@@ -1000,7 +1000,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         manager_storage_del_node(manager, cmd->node);
         if (neu_adapter_get_type(adapter) == NEU_NA_TYPE_APP) {
             UT_array *subscriptions = neu_subscribe_manager_get(
-                manager->subscribe_manager, cmd->node);
+                manager->subscribe_manager, cmd->node, NULL, NULL);
             neu_subscribe_manager_unsub_all(manager->subscribe_manager,
                                             cmd->node);
 
@@ -1165,8 +1165,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
     case NEU_REQ_GET_SUBSCRIBE_GROUP: {
         neu_req_get_subscribe_group_t *cmd =
             (neu_req_get_subscribe_group_t *) &header[1];
-        UT_array *groups =
-            neu_manager_get_sub_group_deep_copy(manager, cmd->app);
+        UT_array *groups = neu_manager_get_sub_group_deep_copy(
+            manager, cmd->app, cmd->driver, cmd->group);
         neu_resp_get_subscribe_group_t resp = { .groups = groups };
 
         strcpy(header->receiver, header->sender);

src/core/manager_internal.c

@@ -791,13 +791,17 @@ int neu_manager_unsubscribe(neu_manager_t *manager, const char *app,
 
 UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app)
 {
-    return neu_subscribe_manager_get(manager->subscribe_manager, app);
+    return neu_subscribe_manager_get(manager->subscribe_manager, app, NULL,
+                                     NULL);
 }
 
 UT_array *neu_manager_get_sub_group_deep_copy(neu_manager_t *manager,
-                                              const char *   app)
+                                              const char *   app,
+                                              const char *   driver,
+                                              const char *   group)
 {
-    UT_array *subs = neu_subscribe_manager_get(manager->subscribe_manager, app);
+    UT_array *subs = neu_subscribe_manager_get(manager->subscribe_manager, app,
+                                               driver, group);
 
     utarray_foreach(subs, neu_resp_subscribe_info_t *, sub)
     {

src/core/manager_internal.h

@@ -112,7 +112,9 @@ int       neu_manager_unsubscribe(neu_manager_t *manager, const char *app,
                                   const char *driver, const char *group);
 UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app);
 UT_array *neu_manager_get_sub_group_deep_copy(neu_manager_t *manager,
-                                              const char *   app);
+                                              const char *   app,
+                                              const char *   driver,
+                                              const char *   group);
 int neu_manager_add_ndriver_map(neu_manager_t *manager, const char *ndriver,
                                 const char *driver, const char *group);
 int neu_manager_del_ndriver_map(neu_manager_t *manager, const char *ndriver,

src/core/node_manager.c

@@ -386,4 +386,4 @@ UT_array *neu_node_manager_get_state(neu_node_manager_t *mgr)
     }
 
     return states;
-}
+}

src/core/subscribe.c

@@ -105,7 +105,8 @@ UT_array *neu_subscribe_manager_find_by_driver(neu_subscribe_mgr_t *mgr,
     return apps;
 }
 
-UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app)
+UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app,
+                                    const char *driver, const char *group)
 {
     sub_elem_t *el = NULL, *tmp = NULL;
     UT_array *  groups = NULL;
@@ -116,7 +117,9 @@ UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app)
     {
         utarray_foreach(el->apps, neu_app_subscribe_t *, sub_app)
         {
-            if (strcmp(sub_app->app_name, app) == 0) {
+            if (strcmp(sub_app->app_name, app) == 0 &&
+                (!driver || strstr(el->key.driver, driver)) &&
+                (!group || strstr(el->key.group, group))) {
                 neu_resp_subscribe_info_t info = { 0 };
 
                 strncpy(info.driver, el->key.driver, sizeof(info.driver));

src/core/subscribe.h

@@ -47,7 +47,8 @@ UT_array *neu_subscribe_manager_find(neu_subscribe_mgr_t *mgr,
                                      const char *driver, const char *group);
 UT_array *neu_subscribe_manager_find_by_driver(neu_subscribe_mgr_t *mgr,
                                                const char *         driver);
-UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app);
+UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app,
+                                    const char *driver, const char *group);
 size_t    neu_subscribe_manager_group_count(const neu_subscribe_mgr_t *mgr,
                                             const char *               app);
 UT_array *neu_subscribe_manager_get_ndriver_maps(neu_subscribe_mgr_t *mgr,

src/event/event_linux.c

@@ -66,7 +66,7 @@ struct event_data {
     bool  use;
 };
 
-#define EVENT_SIZE 1024
+#define EVENT_SIZE 1400
 
 struct neu_events {
     int       epoll_fd;
@@ -239,6 +239,7 @@ neu_event_timer_t *neu_event_add_timer(neu_events_t *          events,
     timer_ctx->event_data->usr_data       = timer.usr_data;
     timer_ctx->event_data->callback.timer = timer.cb;
     timer_ctx->event_data->ctx.timer = events->event_datas[index].ctx.timer;
+    timer_ctx->event_data->index     = index;
 
     timer_ctx->value = value;
     timer_ctx->fd    = timer_fd;
@@ -251,17 +252,17 @@ neu_event_timer_t *neu_event_add_timer(neu_events_t *          events,
     zlog_notice(neuron,
                 "add timer, second: %" PRId64 ", millisecond: %" PRId64
                 ", timer: %d in epoll %d, "
-                "ret: %d",
+                "ret: %d, index: %d",
                 timer.second, timer.millisecond, timer_fd, events->epoll_fd,
-                ret);
+                ret, index);
 
     return timer_ctx;
 }
 
 int neu_event_del_timer(neu_events_t *events, neu_event_timer_t *timer)
 {
-    zlog_notice(neuron, "del timer: %d from epoll: %d", timer->fd,
-                events->epoll_fd);
+    zlog_notice(neuron, "del timer: %d from epoll: %d, index: %d", timer->fd,
+                events->epoll_fd, timer->event_data->index);
 
     timer->stop = true;
     epoll_ctl(events->epoll_fd, EPOLL_CTL_DEL, timer->fd, NULL);
@@ -296,13 +297,14 @@ neu_event_io_t *neu_event_add_io(neu_events_t *events, neu_event_io_param_t io)
     io_ctx->event_data->usr_data    = io.usr_data;
     io_ctx->event_data->callback.io = io.cb;
     io_ctx->event_data->ctx.io      = events->event_datas[index].ctx.io;
+    io_ctx->event_data->index       = index;
 
     io_ctx->fd = io.fd;
 
     ret = epoll_ctl(events->epoll_fd, EPOLL_CTL_ADD, io.fd, &event);
 
-    nlog_notice("add io, fd: %d, epoll: %d, ret: %d(%d)", io.fd,
-                events->epoll_fd, ret, errno);
+    nlog_notice("add io, fd: %d, epoll: %d, ret: %d(%d), index: %d", io.fd,
+                events->epoll_fd, ret, errno, index);
     assert(ret == 0);
 
     return io_ctx;
@@ -314,7 +316,8 @@ int neu_event_del_io(neu_events_t *events, neu_event_io_t *io)
         return 0;
     }
 
-    zlog_notice(neuron, "del io: %d from epoll: %d", io->fd, events->epoll_fd);
+    zlog_notice(neuron, "del io: %d from epoll: %d, index: %d", io->fd,
+                events->epoll_fd, io->event_data->index);
 
     epoll_ctl(events->epoll_fd, EPOLL_CTL_DEL, io->fd, NULL);
     free_event(events, io->event_data->index);

src/parser/neu_json_rw.c

@@ -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);

src/utils/http.c

@@ -354,6 +354,7 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content)
     case NEU_ERR_TAG_PRECISION_INVALID:
     case NEU_ERR_TAG_NAME_TOO_LONG:
     case NEU_ERR_TAG_ADDRESS_TOO_LONG:
+    case NEU_ERR_GROUP_MAX_GROUPS:
     case NEU_ERR_LICENSE_MAX_TAGS:
     case NEU_ERR_LICENSE_BAD_CLOCK:
     case NEU_ERR_LICENSE_MODULE_INVALID:

tests/ft/driver/test_modbus.py

@@ -68,15 +68,15 @@ coil_bit_5 = [{"name": "coil_bit_5", "address": "1!000010",
                "attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_BIT}]
 
 input_bit_1 = [{"name": "input_bit_1", "address": "1!100001",
-                "attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
+                "attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT, "description": "bit_1"}]
 input_bit_2 = [{"name": "input_bit_2", "address": "1!100002",
-                "attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
+                "attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT, "description": "bit_2"}]
 input_bit_3 = [{"name": "input_bit_3", "address": "1!100003",
-                "attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
+                "attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT, "description": "bit_3"}]
 input_bit_4 = [{"name": "input_bit_4", "address": "1!100005",
-                "attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
+                "attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT, "description": "bit_4"}]
 input_bit_5 = [{"name": "input_bit_5", "address": "1!100010",
-                "attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
+                "attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT, "description": "bit_5"}]
 
 input_register_bit = [{"name": "input_register_bit", "address": "1!30101.15",
                        "attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
@@ -494,6 +494,40 @@ class TestModbus:
         assert 0 == api.read_tag(
             node=param[0], group='group', tag=input_bit_5[0]['name'])
 
+    @description(given="created modbus node and tags", when="read tags fuzz", then="should return correct result")
+    def test_read_tags_fuzz(self, param):
+        response = api.read_tags(
+            node=param[0], group='group', query={"name": "input_bit"})
+        assert 200 == response.status_code
+        assert 5 == len(response.json()["tags"])
+
+        response = api.read_tags(
+            node=param[0], group='group', query={"name": "input_bit_1"})
+        assert 200 == response.status_code
+        assert 1 == len(response.json()["tags"])
+
+        response = api.read_tags(
+            node=param[0], group='group', query={"description": "bit"})
+        assert 200 == response.status_code
+        assert 5 == len(response.json()["tags"])
+
+        response = api.read_tags(
+            node=param[0], group='group', query={"description": "bit_1"})
+        assert 200 == response.status_code
+        assert 1 == len(response.json()["tags"])
+
+        response = api.read_tags(
+            node=param[0], group='group', query={"name": "input_bit", "description": "bit"})
+        assert 200 == response.status_code
+        assert 5 == len(response.json()["tags"])
+
+        response = api.read_tags(
+            node=param[0], group='group', query={"name": "input_bit_1", "description": "bit_1"})
+        assert 200 == response.status_code
+        assert 1 == len(response.json()["tags"])
+
+
+
     @description(given="created modbus node and tags", when="read tags synchronously", then="read fail")
     def test_read_tags_sync(self, param):
         assert error.NEU_ERR_PLUGIN_NOT_SUPPORT_READ_SYNC == api.read_tag_error(

tests/ft/http_api/test_http_api.py

@@ -188,6 +188,103 @@ class TestHttp:
         assert len(config_data['settings']) == 1
         assert config_data['settings'][0]['node'] == 'mqtt'
 
+    @description(given="running neuron", when="get subscribe info", then="should success")
+    def test_get_subscribe_info(self):
+        app = 'MQTT-1'
+        params = {'topic':'/neuron/test'}
+        driver_1 = 'modbus-01'
+        driver_2 = 'modbus-02'
+        group_1 = 'group-01'
+        group_2 = 'group-02'
+        try:
+            api.add_node_check(app, 'MQTT')
+            api.add_node_check(driver_1, 'Modbus TCP')
+            api.add_node_check(driver_2, 'Modbus TCP')
+            api.add_group_check(driver_1, group_1, 1000)
+            api.add_group_check(driver_1, group_2, 1000)
+            api.add_group_check(driver_2, group_1, 1000)
+            api.add_group_check(driver_2, group_2, 1000)
+            api.subscribe_group_check(app, driver_1, group_1, params=params)
+            api.subscribe_group_check(app, driver_1, group_2, params=params)
+            api.subscribe_group_check(app, driver_2, group_1, params=params)
+            api.subscribe_group_check(app, driver_2, group_2, params=params)
+
+            # match all drivers
+            response = api.get_subscribe_group(app, driver='modbus')
+            assert 200 == response.status_code
+            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            assert len(groups) == 4
+            assert groups[0]['driver'] == driver_1
+            assert groups[0]['group'] == group_1
+            assert groups[1]['driver'] == driver_1
+            assert groups[1]['group'] == group_2
+            assert groups[2]['driver'] == driver_2
+            assert groups[2]['group'] == group_1
+            assert groups[3]['driver'] == driver_2
+            assert groups[3]['group'] == group_2
+
+            # match one driver
+            response = api.get_subscribe_group(app, driver=driver_1)
+            assert 200 == response.status_code
+            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            assert len(groups) == 2
+            assert groups[0]['driver'] == driver_1
+            assert groups[0]['group'] == group_1
+            assert groups[1]['driver'] == driver_1
+            assert groups[1]['group'] == group_2
+
+            # match none driver
+            response = api.get_subscribe_group(app, driver='none')
+            assert 200 == response.status_code
+            assert len(response.json()['groups']) == 0
+
+            # match all groups
+            response = api.get_subscribe_group(app, group='group')
+            assert 200 == response.status_code
+            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            assert len(groups) == 4
+            assert groups[0]['driver'] == driver_1
+            assert groups[0]['group'] == group_1
+            assert groups[1]['driver'] == driver_1
+            assert groups[1]['group'] == group_2
+            assert groups[2]['driver'] == driver_2
+            assert groups[2]['group'] == group_1
+            assert groups[3]['driver'] == driver_2
+            assert groups[3]['group'] == group_2
+
+            # match one group
+            response = api.get_subscribe_group(app, group=group_1)
+            assert 200 == response.status_code
+            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            assert len(groups) == 2
+            assert groups[0]['driver'] == driver_1
+            assert groups[0]['group'] == group_1
+            assert groups[1]['driver'] == driver_2
+            assert groups[1]['group'] == group_1
+
+            # match none group
+            response = api.get_subscribe_group(app, group='none')
+            assert 200 == response.status_code
+            assert len(response.json()['groups']) == 0
+
+            # match driver and group
+            response = api.get_subscribe_group(app, driver=driver_1, group=group_1)
+            assert 200 == response.status_code
+            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            assert len(groups) == 1
+            assert groups[0]['driver'] == driver_1
+            assert groups[0]['group'] == group_1
+            response = api.get_subscribe_group(app, driver=driver_2, group=group_2)
+            assert 200 == response.status_code
+            groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
+            assert len(groups) == 1
+            assert groups[0]['driver'] == driver_2
+            assert groups[0]['group'] == group_2
+        finally:
+            api.del_node(app)
+            api.del_node(driver_1)
+            api.del_node(driver_2)
+
     @description(given="running neuron", when="get mqtt shcema", then="success")
     def test_get_plugin_schema(self):
         response = api.get_plugin_schema(plugin='mqtt')

tests/ft/neuron/api.py

@@ -145,12 +145,15 @@ def get_tags(node, group):
     return requests.get(url=config.BASE_URL + "/api/v2/tags", headers={"Authorization": config.default_jwt}, params={"node": node, "group": group})
 
 
-def read_tags(node, group, sync=False):
-    return requests.post(url=config.BASE_URL + "/api/v2/read", headers={"Authorization": config.default_jwt}, json={"node": node, "group": group, "sync": sync})
+def read_tags(node, group, sync=False, query=None):
+    body = {"node": node, "group": group, "sync": sync}
+    if query:
+        body["query"] = query
+    return requests.post(url=config.BASE_URL + "/api/v2/read", headers={"Authorization": config.default_jwt}, json=body)
 
 
 def read_tag(node, group, tag, sync=False):
-    response = read_tags(node, group, sync)
+    response = read_tags(node, group, sync, query={"name": tag})
     assert 200 == response.status_code
     x = list(filter(lambda x: x['name'] == tag,
                     response.json()['tags']))
@@ -219,6 +222,10 @@ def get_plugins():
     return requests.get(url=config.BASE_URL + "/api/v2/plugin", headers={"Authorization": config.default_jwt})
 
 
+def get_subscribe_group(app, driver=None, group=None):
+    return requests.get(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, params={"app": app, "driver": driver, "group": group})
+
+
 @gen_check
 def add_template(name, plugin, groups=[]):
     return requests.post(url=config.BASE_URL + "/api/v2/template", json={"name": name, "plugin": plugin, "groups": groups}, headers={"Authorization": config.default_jwt})

version

@@ -1 +1 @@
-2.7.0
+2.7.1