Merge pull request #1803 from fengzeroz/v2.7

version: 2.7.1

.github/workflows/check.yml

@@ -41,7 +41,7 @@ jobs:
 
   ut:
     runs-on: ubuntu-20.04
-    container: ghcr.io/neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.7
 
 
     steps:
@@ -74,17 +74,9 @@ jobs:
             build/result/index.html
             build/neuron.log
 
-      - name: Upload coverage to Codecov
-        uses: codecov/codecov-action@v3
-        with:
-          gcov: true
-          fail_ci_if_error: true
-          gcov_include: include plugins src
-          gcov_ignore: tests simulator
-
   ft:
     runs-on: ubuntu-20.04
-    container: ghcr.io/neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.7
 
     steps:
       - uses: actions/checkout@v4
@@ -115,16 +107,9 @@ jobs:
             ft/reports/report.html
             build/neuron.log
 
-      - name: Upload coverage to Codecov
-        uses: codecov/codecov-action@v3
-        with:
-          gcov: true
-          gcov_include: include plugins src
-          gcov_ignore: tests simulator
-
   pft:
     runs-on: ubuntu-20.04
-    container: ghcr.io/neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.7
     strategy:
       matrix:
         plugin: [core, modbus, mqtt, metrics]
@@ -163,17 +148,3 @@ jobs:
           else
             pytest -s -v tests/ft/metrics/"test_metrics.py"          
           fi
-
-      - name: create cov report
-        run: |
-            sudo apt-get update
-            sudo apt-get -y install lcov
-            mkdir -p cov_report
-            ./create_cov_report.sh ${{ matrix.plugin }}
-
-      - name: Upload coverage to Codecov
-        uses: codecov/codecov-action@v3
-        with:
-          verbose: true
-          directory: ./cov_report/
-          files: cov-${{ matrix.plugin }}.info

.github/workflows/ci.yml

@@ -8,9 +8,9 @@ on:
       - published
 
 env:
-  VERSION: 2.7.0
-  UI_VERSION: 2.6.1
-  BRANCH: main
+  VERSION: 2.7.1
+  UI_VERSION: 2.6.3
+  BRANCH: v2.7
 
 jobs:
   cross-build:
@@ -21,7 +21,7 @@ jobs:
           - [aarch64-linux-gnu, aarch64, arm64]
           - [arm-linux-gnueabihf, armv4, armhf]
           - [x86_64-linux-gnu, x86_64, amd64]
-    container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-main
+    container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-v2.7
 
     steps:
       - uses: actions/checkout@v4
@@ -64,7 +64,7 @@ jobs:
       matrix:
         arch:
           - [x86_64-linux-gnu, x86_64, amd64]
-    container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-main
+    container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-v2.7
 
     steps:
       - uses: actions/checkout@v4

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

cmake/aarch64-linux-gnu.cmake

@@ -1,7 +1,7 @@
 set(CMAKE_SYSTEM_NAME Linux)
 set(COMPILER_PREFIX aarch64-linux-gnu)
 set(CMAKE_SYSTEM_PROCESSOR aarch64)
-set(LIBRARY_DIR /home/neuron/main/libs)
+set(LIBRARY_DIR /home/neuron/v2.7/libs)
 
 set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
 set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)

cmake/arm-linux-gnueabi.cmake

@@ -1,7 +1,7 @@
 set(CMAKE_SYSTEM_NAME Linux)
 set(COMPILER_PREFIX arm-linux-gnueabi)
 set(CMAKE_SYSTEM_PROCESSOR armv5tejl)
-set(LIBRARY_DIR /home/neuron/main/libs)
+set(LIBRARY_DIR /home/neuron/v2.7/libs)
 
 set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
 set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)

cmake/arm-linux-gnueabihf.cmake

@@ -1,7 +1,7 @@
 set(CMAKE_SYSTEM_NAME Linux)
 set(COMPILER_PREFIX arm-linux-gnueabihf)
 set(CMAKE_SYSTEM_PROCESSOR armv7l)
-set(LIBRARY_DIR /home/neuron/main/libs)
+set(LIBRARY_DIR /home/neuron/v2.7/libs)
 
 set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
 set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)

cmake/x86_64-linux-gnu.cmake

@@ -1,7 +1,7 @@
 set(CMAKE_SYSTEM_NAME Linux)
 set(COMPILER_PREFIX x86_64-linux-gnu)
 set(CMAKE_SYSTEM_PROCESSOR x86_64)
-set(LIBRARY_DIR /home/neuron/main/libs)
+set(LIBRARY_DIR /home/neuron/v2.7/libs)
 
 set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
 set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)

ft/tag_test.robot

@@ -59,7 +59,7 @@ Add different types of tags to the group, it should return success
 Add a list of tags with errors, it should return the number of successful additions.
   	${res}=	Add Tags	modbus-node	group	${tag1},${tag2},${tag3},${tag4}
 
-  	Check Response Status           ${res}        200
+  	Check Response Status           ${res}        409
   	Check Error Code                ${res}        ${NEU_ERR_TAG_NAME_CONFLICT}
 	Should Be Equal As Integers	${res}[index]	0
 
@@ -139,7 +139,7 @@ Delete tag from non-existent group, it should return success
 Update non-existent tag, it should return failure.
 	${res}= 	Update Tags  modbus-node  group  ${tag4}
 
-  	Check Response Status           ${res}        200
+  	Check Response Status           ${res}        404
   	Check Error Code                ${res}        ${NEU_ERR_TAG_NOT_EXIST}
 
 	${res}=		Get Tags  modbus-node  group

ft/template.robot

@@ -96,7 +96,7 @@ Create a template with invalid tag type, it should fail.
     ${groups} =                   Create List           ${group}
 
     ${res} =                      Add Template          ${g_template}                 ${g_plugin}           ${groups}
-    Check Response Status         ${res}                206
+    Check Response Status         ${res}                400
     Check Error Code              ${res}                ${NEU_ERR_TAG_TYPE_NOT_SUPPORT}
 
     [Teardown]                    Del Template          ${g_template}
@@ -108,7 +108,7 @@ Create a template with invalid tag address, it should fail.
     ${groups} =                   Create List           ${group}
 
     ${res} =                      Add Template          ${g_template}                 ${g_plugin}           ${groups}
-    Check Response Status         ${res}                206
+    Check Response Status         ${res}                400
     Check Error Code              ${res}                ${NEU_ERR_TAG_ADDRESS_FORMAT_INVALID}
 
     [Teardown]                    Del Template          ${g_template}
@@ -802,7 +802,7 @@ Add tag with too long name to existing template group, it should fail.
 
     &{tag} =                      New Tag From          ${g_tag1}                     name=${g_long_str}
     ${res} =                      Add Template Tags     ${g_template}                 ${g_group1}[name]     ${tag}
-    Check Response Status         ${res}                206
+    Check Response Status         ${res}                400
     Check Error Code              ${res}                ${NEU_ERR_TAG_NAME_TOO_LONG}
 
     [Teardown]                    Del Template          ${g_template}
@@ -1026,7 +1026,7 @@ Update tag name to invalid value, it should fail.
 
     &{tag} =                      New Tag From          ${g_tag1}                     name=${g_long_str}
     ${res} =                      Put Template Tags     ${g_template}                 ${g_group1}[name]     ${tag}
-    Check Response Status         ${res}                206
+    Check Response Status         ${res}                400
     Check Error Code              ${res}                ${NEU_ERR_TAG_NAME_TOO_LONG}
 
     [Teardown]                    Del Template          ${g_template}

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,
@@ -98,6 +99,7 @@ typedef enum {
     NEU_ERR_LIBRARY_MODULE_VERSION_NOT_MATCH  = 2314,
     NEU_ERR_LIBRARY_NAME_NOT_CONFORM          = 2315,
     NEU_ERR_LIBRARY_CLIB_NOT_MATCH            = 2316,
+    NEU_ERR_LIBRARY_UPDATE_FAIL               = 2317,
 
     NEU_ERR_LICENSE_NOT_FOUND       = 2400,
     NEU_ERR_LICENSE_INVALID         = 2401,

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/datatag_handle.c

@@ -92,7 +92,8 @@ void handle_add_tags_resp(nng_aio *aio, neu_resp_add_tag_t *resp)
 
     neu_json_encode_by_fn(&res, neu_json_encode_au_tags_resp, &result);
 
-    neu_http_ok(aio, result);
+    NEU_JSON_RESPONSE_ERROR(resp->error,
+                            { neu_http_response(aio, resp->error, result); });
     free(result);
 }
 
@@ -183,8 +184,8 @@ void handle_add_gtags_resp(nng_aio *aio, neu_resp_add_tag_t *resp)
     res.index = resp->index;
 
     neu_json_encode_by_fn(&res, neu_json_encode_au_gtags_resp, &result);
-
-    neu_http_ok(aio, result);
+    NEU_JSON_RESPONSE_ERROR(resp->error,
+                            { neu_http_response(aio, resp->error, result); });
     free(result);
 }
 

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,
@@ -58,6 +59,13 @@ static int adapter_update_metric(neu_adapter_t *adapter,
 inline static void reply(neu_adapter_t *adapter, neu_reqresp_head_t *header,
                          void *data);
 
+int validate_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
+                   neu_resp_add_tag_t *resp);
+int try_add_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
+                  neu_resp_add_tag_t *resp);
+int add_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
+              neu_resp_add_tag_t *resp);
+
 static const adapter_callbacks_t callback_funs = {
     .command         = adapter_command,
     .response        = adapter_response,
@@ -103,6 +111,36 @@ 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 };
+    // replace path separators
+    for (int i = 0; node[i]; ++i) {
+        name[i] = ('/' == node[i] || '\\' == node[i]) ? '_' : node[i];
+    }
+    return zlog_get_category(name);
+}
+
 neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load)
 {
     int                  rv      = 0;
@@ -166,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 = {
@@ -201,7 +242,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load)
     common->adapter             = adapter;
     common->adapter_callbacks   = &adapter->cb_funs;
     common->link_state          = NEU_NODE_LINK_STATE_DISCONNECTED;
-    common->log                 = zlog_get_category(adapter->name);
+    common->log                 = get_log_category(adapter->name);
     strcpy(common->name, adapter->name);
 
     zlog_level_switch(common->log, default_log_level);
@@ -272,7 +313,7 @@ int neu_adapter_rename(neu_adapter_t *adapter, const char *new_name)
         return NEU_ERR_EINTERNAL;
     }
 
-    zlog_category_t *log = zlog_get_category(name);
+    zlog_category_t *log = get_log_category(name);
     if (NULL == log) {
         free(name);
         free(old_name);
@@ -598,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;
 }
@@ -1033,65 +1081,26 @@ 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) {
-            for (int i = 0; i < cmd->n_group; i++) {
-                for (int j = 0; j < cmd->groups[i].n_tag; j++) {
-                    int ret = neu_adapter_driver_validate_tag(
-                        (neu_adapter_driver_t *) adapter, cmd->groups[i].group,
-                        &cmd->groups[i].tags[j]);
-                    if (ret == 0) {
-                        resp.index += 1;
-                    } else {
-                        resp.error = ret;
-                        resp.index = 0;
-                        i          = cmd->n_group;
-                        break;
-                    }
-                }
-            }
-        } else {
+        if (adapter->module->type != NEU_NA_TYPE_DRIVER) {
             resp.error = NEU_ERR_GROUP_NOT_ALLOW;
-        }
-        if (resp.index > 0) {
-            for (int i = 0; i < cmd->n_group; i++) {
-                int ret = neu_adapter_driver_try_add_tag(
-                    (neu_adapter_driver_t *) adapter, cmd->groups[i].group,
-                    cmd->groups[i].tags, cmd->groups[i].n_tag);
-                if (ret != 0) {
-                    resp.index = 0;
-                    resp.error = ret;
-                    break;
+        } 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 &&
+                add_gtags(adapter, cmd, &resp) == 0) {
+                for (int i = 0; i < cmd->n_group; i++) {
+                    adapter_storage_add_tags(cmd->driver, cmd->groups[i].group,
+                                             cmd->groups[i].tags,
+                                             cmd->groups[i].n_tag);
                 }
             }
         }
 
-        if (resp.index > 0) {
-            for (int i = 0; i < cmd->n_group; i++) {
-                for (int j = 0; j < cmd->groups[i].n_tag; j++) {
-                    int ret = neu_adapter_driver_add_tag(
-                        (neu_adapter_driver_t *) adapter, cmd->groups[i].group,
-                        &cmd->groups[i].tags[j], cmd->groups[i].interval);
-                    if (ret != 0) {
-                        neu_adapter_driver_try_del_tag(
-                            (neu_adapter_driver_t *) adapter,
-                            cmd->groups[i].n_tag - j);
-                        resp.index = 0;
-                        i          = cmd->n_group;
-                        resp.error = ret;
-                        break;
-                    }
-                }
-            }
-        }
-
-        if (resp.index) {
-            for (int i = 0; i < cmd->n_group; i++) {
-                adapter_storage_add_tags(cmd->driver, cmd->groups[i].group,
-                                         cmd->groups[i].tags,
-                                         cmd->groups[i].n_tag);
-            }
-        }
-
         for (int i = 0; i < cmd->n_group; i++) {
             for (int j = 0; j < cmd->groups[i].n_tag; j++) {
                 neu_tag_fini(&cmd->groups[i].tags[j]);
@@ -1258,6 +1267,97 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
     return 0;
 }
 
+int validate_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
+                   neu_resp_add_tag_t *resp)
+{
+    for (int group_index = 0; group_index < cmd->n_group; group_index++) {
+        for (int tag_index = 0; tag_index < cmd->groups[group_index].n_tag;
+             tag_index++) {
+            int validation_result = neu_adapter_driver_validate_tag(
+                (neu_adapter_driver_t *) adapter,
+                cmd->groups[group_index].group,
+                &cmd->groups[group_index].tags[tag_index]);
+            if (validation_result == 0) {
+                resp->index += 1;
+            } else {
+                resp->error = validation_result;
+                resp->index = 0;
+                return validation_result;
+            }
+        }
+    }
+    return 0;
+}
+
+int try_add_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
+                  neu_resp_add_tag_t *resp)
+{
+    for (int group_index = 0; group_index < cmd->n_group; group_index++) {
+        int add_result = neu_adapter_driver_try_add_tag(
+            (neu_adapter_driver_t *) adapter, cmd->groups[group_index].group,
+            cmd->groups[group_index].tags, cmd->groups[group_index].n_tag);
+        if (add_result != 0) {
+            for (int added_groups_count = 0; added_groups_count < group_index;
+                 added_groups_count++) {
+                neu_adapter_driver_try_del_tag(
+                    (neu_adapter_driver_t *) adapter,
+                    cmd->groups[added_groups_count].n_tag);
+            }
+            resp->index = 0;
+            resp->error = add_result;
+            return add_result;
+        }
+    }
+    return 0;
+}
+
+int add_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
+              neu_resp_add_tag_t *resp)
+{
+    for (int group_index = 0; group_index < cmd->n_group; group_index++) {
+        for (int tag_index = 0; tag_index < cmd->groups[group_index].n_tag;
+             tag_index++) {
+            int add_tag_result = neu_adapter_driver_add_tag(
+                (neu_adapter_driver_t *) adapter,
+                cmd->groups[group_index].group,
+                &cmd->groups[group_index].tags[tag_index],
+                cmd->groups[group_index].interval);
+            if (add_tag_result != 0) {
+                for (int added_group_index = 0; added_group_index < group_index;
+                     added_group_index++) {
+                    for (int added_tag_index = 0; added_tag_index <=
+                         cmd->groups[added_group_index].n_tag;
+                         added_tag_index++) {
+                        neu_adapter_driver_del_tag(
+                            (neu_adapter_driver_t *) adapter,
+                            cmd->groups[added_group_index].group,
+                            cmd->groups[added_group_index]
+                                .tags[added_tag_index]
+                                .name);
+                    }
+                }
+                for (int added_tag_index = 0; added_tag_index < tag_index;
+                     added_tag_index++) {
+                    neu_adapter_driver_del_tag(
+                        (neu_adapter_driver_t *) adapter,
+                        cmd->groups[group_index].group,
+                        cmd->groups[group_index].tags[added_tag_index].name);
+                }
+                for (int groups_count = 0; groups_count < cmd->n_group;
+                     groups_count++) {
+                    neu_adapter_driver_try_del_tag(
+                        (neu_adapter_driver_t *) adapter,
+                        cmd->groups[groups_count].n_tag);
+                }
+                resp->index = 0;
+                resp->error = add_tag_result;
+                return add_tag_result;
+            }
+        }
+    }
+    return 0;
+}
+
 void neu_adapter_destroy(neu_adapter_t *adapter)
 {
     nlog_notice("adapter %s destroy", adapter->name);
@@ -1276,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());
 
@@ -574,20 +574,62 @@ static void cal_decimal(neu_type_e tag_type, neu_type_e value_type,
 {
     switch (tag_type) {
     case NEU_TYPE_INT8:
-    case NEU_TYPE_UINT8:
+        if (value_type == NEU_TYPE_DOUBLE) {
+            value->i8 = (int8_t) round(value->d64 / decimal);
+        } else {
+            value->i8 = (int8_t) round(value->i64 / decimal);
+        }
+        break;
     case NEU_TYPE_INT16:
-    case NEU_TYPE_UINT16:
+        if (value_type == NEU_TYPE_DOUBLE) {
+            value->i16 = (int16_t) round(value->d64 / decimal);
+        } else {
+            value->i16 = (int16_t) round(value->i64 / decimal);
+        }
+        break;
     case NEU_TYPE_INT32:
-    case NEU_TYPE_UINT32:
+        if (value_type == NEU_TYPE_DOUBLE) {
+            value->i32 = (int32_t) round(value->d64 / decimal);
+        } else {
+            value->i32 = (int32_t) round(value->i64 / decimal);
+        }
+        break;
     case NEU_TYPE_INT64:
-    case NEU_TYPE_UINT64:
+        if (value_type == NEU_TYPE_DOUBLE) {
+            value->i64 = (int64_t) round(value->d64 / decimal);
+        } else {
+            value->i64 = (int64_t) round(value->i64 / decimal);
+        }
+        break;
+    case NEU_TYPE_UINT8:
+        if (value_type == NEU_TYPE_DOUBLE) {
+            value->u8 = (uint8_t) round(value->d64 / decimal);
+        } else {
+            value->u8 = (uint8_t) round(value->u64 / decimal);
+        }
+        break;
+    case NEU_TYPE_UINT16:
     case NEU_TYPE_WORD:
+        if (value_type == NEU_TYPE_DOUBLE) {
+            value->u16 = (uint16_t) round(value->d64 / decimal);
+        } else {
+            value->u16 = (uint16_t) round(value->u64 / decimal);
+        }
+        break;
+    case NEU_TYPE_UINT32:
     case NEU_TYPE_LWORD:
+        if (value_type == NEU_TYPE_DOUBLE) {
+            value->u32 = (uint32_t) round(value->d64 / decimal);
+        } else {
+            value->u32 = (uint32_t) round(value->u64 / decimal);
+        }
+        break;
+    case NEU_TYPE_UINT64:
     case NEU_TYPE_DWORD:
         if (value_type == NEU_TYPE_INT64 || value_type == NEU_TYPE_LWORD) {
-            value->u64 = value->u64 / decimal;
-        } else if (value_type == NEU_TYPE_DOUBLE) {
-            value->u64 = (uint64_t) round(value->d64 / decimal);
+            value->d64 = (uint64_t) round(value->d64 / decimal);
+        } else {
+            value->u64 = (uint64_t) round(value->u64 / decimal);
         }
         break;
     case NEU_TYPE_FLOAT:
@@ -1063,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/connection/connection.c

@@ -1274,30 +1274,29 @@ int neu_conn_stream_consume(neu_conn_t *conn, void *context,
     ssize_t ret = neu_conn_recv(conn, conn->buf + conn->offset,
                                 conn->buf_size - conn->offset);
     if (ret > 0) {
-        pthread_mutex_lock(&conn->mtx);
         zlog_recv_protocol(conn->param.log, conn->buf + conn->offset, ret);
         conn->offset += ret;
         neu_protocol_unpack_buf_t protocol_buf = { 0 };
         neu_protocol_unpack_buf_init(&protocol_buf, conn->buf, conn->offset);
-        int used = 0;
         while (neu_protocol_unpack_buf_unused_size(&protocol_buf) > 0) {
-            used = fn(context, &protocol_buf);
+            int used = fn(context, &protocol_buf);
 
             zlog_debug(conn->param.log, "buf used: %d offset: %d", used,
                        conn->offset);
-            if (used == 0 || used == -1) {
+            if (used == 0) {
+                break;
+            } else if (used == -1) {
+                neu_conn_disconnect(conn);
                 break;
             } else {
+                pthread_mutex_lock(&conn->mtx);
                 conn->offset -= used;
                 memmove(conn->buf, conn->buf + used, conn->offset);
                 neu_protocol_unpack_buf_init(&protocol_buf, conn->buf,
                                              conn->offset);
+                pthread_mutex_unlock(&conn->mtx);
             }
         }
-        pthread_mutex_unlock(&conn->mtx);
-        if (used == -1) {
-            neu_conn_disconnect(conn);
-        }
     }
 
     return ret;

src/core/manager.c

@@ -615,7 +615,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
             free(so_tmp_path);
             free(schema_tmp_path);
             header->type = NEU_RESP_ERROR;
-            e.error      = NEU_ERR_LIBRARY_ADD_FAIL;
+            e.error      = NEU_ERR_LIBRARY_UPDATE_FAIL;
             strcpy(header->receiver, header->sender);
             reply(manager, header, &e);
             break;
@@ -628,7 +628,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
             free(so_tmp_path);
             free(schema_tmp_path);
             header->type = NEU_RESP_ERROR;
-            e.error      = NEU_ERR_LIBRARY_ADD_FAIL;
+            e.error      = NEU_ERR_LIBRARY_UPDATE_FAIL;
             strcpy(header->receiver, header->sender);
             reply(manager, header, &e);
             break;
@@ -641,7 +641,21 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
             free(so_tmp_path);
             free(schema_tmp_path);
             header->type = NEU_RESP_ERROR;
-            e.error      = NEU_ERR_LIBRARY_ADD_FAIL;
+            e.error      = NEU_ERR_LIBRARY_UPDATE_FAIL;
+            strcpy(header->receiver, header->sender);
+            reply(manager, header, &e);
+            break;
+        }
+
+        if (neu_plugin_manager_update(manager->plugin_manager, cmd->library) !=
+            NEU_ERR_SUCCESS) {
+            nlog_warn("update library %s fail", cmd->library);
+            free(cmd->so_file);
+            free(cmd->schema_file);
+            free(so_tmp_path);
+            free(schema_tmp_path);
+            header->type = NEU_RESP_ERROR;
+            e.error      = NEU_ERR_LIBRARY_UPDATE_FAIL;
             strcpy(header->receiver, header->sender);
             reply(manager, header, &e);
             break;
@@ -986,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);
 
@@ -1151,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/plugin_manager.c

@@ -98,7 +98,7 @@ int neu_plugin_manager_add(neu_plugin_manager_t *mgr,
         snprintf(lib_path, sizeof(lib_path) - 1, "%s/%s", lib_paths[i],
                  plugin_lib_name);
 
-        handle = dlopen(lib_path, RTLD_NOW | RTLD_NODELETE);
+        handle = dlopen(lib_path, RTLD_NOW);
 
         if (handle == NULL) {
             nlog_warn("failed to open library %s, error: %s", lib_path,
@@ -173,6 +173,78 @@ int neu_plugin_manager_add(neu_plugin_manager_t *mgr,
     return NEU_ERR_SUCCESS;
 }
 
+int neu_plugin_manager_update(neu_plugin_manager_t *mgr,
+                              const char *          plugin_lib_name)
+{
+    char                 lib_path[256]    = { 0 };
+    void *               handle           = NULL;
+    void *               module           = NULL;
+    neu_plugin_module_t *pm               = NULL;
+    plugin_entity_t *    plugin           = NULL;
+    char                 lib_paths[3][64] = { 0 };
+    snprintf(lib_paths[0], sizeof(lib_paths[0]), "%s", g_plugin_dir);
+    snprintf(lib_paths[1], sizeof(lib_paths[1]), "%s/system", g_plugin_dir);
+    snprintf(lib_paths[2], sizeof(lib_paths[2]), "%s/custom", g_plugin_dir);
+
+    assert(strlen(plugin_lib_name) <= NEU_PLUGIN_LIBRARY_LEN);
+
+    for (size_t i = 0; i < sizeof(lib_paths) / sizeof(lib_paths[0]); i++) {
+
+        snprintf(lib_path, sizeof(lib_path) - 1, "%s/%s", lib_paths[i],
+                 plugin_lib_name);
+
+        handle = dlopen(lib_path, RTLD_NOW);
+
+        if (handle == NULL) {
+            nlog_warn("failed to open library %s, error: %s", lib_path,
+                      dlerror());
+        } else {
+            break;
+        }
+    }
+
+    if (handle == NULL) {
+        return NEU_ERR_LIBRARY_FAILED_TO_OPEN;
+    }
+
+    module = dlsym(handle, "neu_plugin_module");
+    if (module == NULL) {
+        dlclose(handle);
+        nlog_warn("failed to find neu_plugin_module symbol in %s", lib_path);
+        return NEU_ERR_LIBRARY_MODULE_INVALID;
+    }
+
+    pm = (neu_plugin_module_t *) module;
+
+    HASH_FIND_STR(mgr->plugins, pm->module_name, plugin);
+    if (plugin == NULL) {
+        dlclose(handle);
+        return NEU_ERR_LIBRARY_MODULE_NOT_EXISTS;
+    }
+
+    plugin->version = pm->version;
+    plugin->display = pm->display;
+    plugin->type    = pm->type;
+    plugin->kind    = pm->kind;
+    free(plugin->schema);
+    free(plugin->description);
+    free(plugin->description_zh);
+    plugin->schema         = strdup(pm->schema);
+    plugin->description    = strdup(pm->module_descr);
+    plugin->description_zh = strdup(pm->module_descr_zh);
+    plugin->single         = pm->single;
+    if (plugin->single) {
+        free(plugin->single_name);
+        plugin->single_name = strdup(pm->single_name);
+    }
+
+    nlog_notice("update plugin, name: %s, library: %s, kind: %d, type: %d",
+                plugin->name, plugin->lib_name, plugin->kind, plugin->type);
+
+    dlclose(handle);
+    return NEU_ERR_SUCCESS;
+}
+
 int neu_plugin_manager_del(neu_plugin_manager_t *mgr, const char *plugin_name)
 {
     plugin_entity_t *plugin = NULL;
@@ -322,7 +394,7 @@ int neu_plugin_manager_create_instance(neu_plugin_manager_t * mgr,
 
             snprintf(lib_path, sizeof(lib_path) - 1, "%s/%s", lib_paths[i],
                      plugin->lib_name);
-            instance->handle = dlopen(lib_path, RTLD_NOW | RTLD_NODELETE);
+            instance->handle = dlopen(lib_path, RTLD_NOW);
 
             if (instance->handle == NULL) {
                 continue;
@@ -387,7 +459,7 @@ bool neu_plugin_manager_create_instance_by_path(neu_plugin_manager_t *mgr,
 
     (void) mgr;
 
-    instance->handle = dlopen(plugin_path, RTLD_NOW | RTLD_NODELETE);
+    instance->handle = dlopen(plugin_path, RTLD_NOW);
 
     if (instance->handle == NULL) {
         const char *dl_err = dlerror();

src/core/plugin_manager.h

@@ -71,4 +71,7 @@ bool neu_plugin_manager_create_instance_by_path(neu_plugin_manager_t *mgr,
 bool neu_plugin_manager_remove_library(neu_plugin_manager_t *mgr,
                                        const char *          library);
 
+int neu_plugin_manager_update(neu_plugin_manager_t *mgr,
+                              const char *          plugin_lib_name);
+
 #endif

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;
@@ -182,8 +182,12 @@ neu_events_t *neu_event_new(void)
     neu_events_t *events = calloc(1, sizeof(struct neu_events));
 
     events->epoll_fd = epoll_create(1);
-    events->stop     = false;
-    events->n_event  = 0;
+
+    nlog_notice("create epoll: %d(%d)", events->epoll_fd, errno);
+    assert(events->epoll_fd > 0);
+
+    events->stop    = false;
+    events->n_event = 0;
     pthread_mutex_init(&events->mtx, NULL);
 
     pthread_create(&events->thread, NULL, event_loop, events);
@@ -235,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;
@@ -247,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);
@@ -273,8 +278,13 @@ int neu_event_del_timer(neu_events_t *events, neu_event_timer_t *timer)
 
 neu_event_io_t *neu_event_add_io(neu_events_t *events, neu_event_io_param_t io)
 {
-    int             ret      = 0;
-    int             index    = get_free_event(events);
+    int ret   = 0;
+    int index = get_free_event(events);
+
+    nlog_notice("add io, fd: %d, epoll: %d, index: %d", io.fd, events->epoll_fd,
+                index);
+    assert(index >= 0);
+
     neu_event_io_t *io_ctx   = &events->event_datas[index].ctx.io;
     io_ctx->event_data       = &events->event_datas[index];
     struct epoll_event event = {
@@ -287,13 +297,15 @@ 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", io.fd, events->epoll_fd,
-                ret);
+    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;
 }
@@ -304,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/main.c

@@ -125,9 +125,10 @@ static int neuron_run(const neu_cli_args_t *args)
     int           rv = 0;
 
     signal(SIGINT, sig_handler);
-    signal(SIGTERM, sig_handler);
     signal(SIGABRT, sig_handler);
     signal(SIGSEGV, sig_handler);
+    signal(SIGTERM, sig_handler);
+    signal(SIGKILL, sig_handler);
 
     // try to enable core dump
     rl.rlim_cur = rl.rlim_max = RLIM_INFINITY;

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

@@ -287,7 +287,6 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content)
     case NEU_ERR_LICENSE_EXPIRED:
     case NEU_ERR_LICENSE_DISABLED:
     case NEU_ERR_LICENSE_MAX_NODES:
-    case NEU_ERR_LICENSE_MAX_TAGS:
     case NEU_ERR_LICENSE_TOKEN_NOT_MATCH:
     case NEU_ERR_GROUP_ALREADY_SUBSCRIBED:
     case NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH:
@@ -341,12 +340,24 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content)
     case NEU_ERR_PLUGIN_NAME_TOO_LONG:
     case NEU_ERR_PLUGIN_NOT_SUPPORT_TEMPLATE:
     case NEU_ERR_LIBRARY_ADD_FAIL:
+    case NEU_ERR_LIBRARY_UPDATE_FAIL:
     case NEU_ERR_LIBRARY_MODULE_NOT_EXISTS:
     case NEU_ERR_LIBRARY_MODULE_KIND_NOT_SUPPORT:
     case NEU_ERR_LIBRARY_MODULE_VERSION_NOT_MATCH:
     case NEU_ERR_LIBRARY_NAME_NOT_CONFORM:
     case NEU_ERR_LIBRARY_CLIB_NOT_MATCH:
     case NEU_ERR_LIBRARY_ARCH_NOT_SUPPORT:
+    case NEU_ERR_TAG_ATTRIBUTE_NOT_SUPPORT:
+    case NEU_ERR_TAG_TYPE_NOT_SUPPORT:
+    case NEU_ERR_TAG_ADDRESS_FORMAT_INVALID:
+    case NEU_ERR_TAG_DESCRIPTION_TOO_LONG:
+    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:
         status = NNG_HTTP_STATUS_BAD_REQUEST;
         break;
     case NEU_ERR_FILE_NOT_EXIST:
@@ -382,15 +393,6 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content)
     case NEU_ERR_LIBRARY_SYSTEM_NOT_ALLOW_DEL:
         status = NNG_HTTP_STATUS_CONFLICT;
         break;
-    case NEU_ERR_TAG_ATTRIBUTE_NOT_SUPPORT:
-    case NEU_ERR_TAG_TYPE_NOT_SUPPORT:
-    case NEU_ERR_TAG_ADDRESS_FORMAT_INVALID:
-    case NEU_ERR_TAG_DESCRIPTION_TOO_LONG:
-    case NEU_ERR_TAG_PRECISION_INVALID:
-    case NEU_ERR_TAG_NAME_TOO_LONG:
-    case NEU_ERR_TAG_ADDRESS_TOO_LONG:
-        status = NNG_HTTP_STATUS_PARTIAL_CONTENT;
-        break;
     default:
         if (code >= NEU_ERR_PLUGIN_ERROR_START &&
             code <= NEU_ERR_PLUGIN_ERROR_END) {

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}]
@@ -116,6 +116,8 @@ hold_string_s = [{"name": "hold_string_s", "address": "1!400870.2",
 
 hold_int16_decimal = [{"name": "hold_int16_decimal", "address": "1!40103",
                        "attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_INT16, "decimal": 0.1}]
+hold_int16_decimal_neg = [{"name": "int16_decimal_neg", "address": "1!40140",
+                       "attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_INT16, "decimal": -1}]
 hold_uint16_decimal = [{"name": "hold_uint16_decimal", "address": "1!400105",
                         "attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_UINT16, "decimal": 0.2}]
 hold_int32_decimal = [{"name": "hold_int32_decimal", "address": "1!400199",
@@ -124,6 +126,8 @@ hold_uint32_decimal = [{"name": "hold_uint32_decimal", "address": "1!400115",
                         "attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_UINT32, "decimal": 0.3}]
 hold_float_decimal = [{"name": "hold_float_decimal", "address": "1!400119",
                        "attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_FLOAT, "decimal": 0.5}]
+hold_float_decimal_neg = [{"name": "float_decimal_neg", "address": "1!400150",
+                       "attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_FLOAT, "decimal": -1}]
 hold_double_decimal = [{"name": "hold_double_decimal", "address": "1!4000127",
                         "attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_DOUBLE, "decimal": 0.6}]
 
@@ -378,27 +382,27 @@ class TestModbus:
                        "attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_BYTES}]
 
         response = api.add_tags(node=param[0], group='group', tags=wrong_tag1)
-        assert 200 == response.status_code
+        assert 400 == response.status_code
         assert error.NEU_ERR_TAG_ATTRIBUTE_NOT_SUPPORT == response.json()[
             'error']
 
         response = api.add_tags(node=param[0], group='group', tags=wrong_tag2)
-        assert 200 == response.status_code
+        assert 400 == response.status_code
         assert error.NEU_ERR_TAG_TYPE_NOT_SUPPORT == response.json()[
             'error']
 
         response = api.add_tags(node=param[0], group='group', tags=wrong_tag3)
-        assert 200 == response.status_code
+        assert 400 == response.status_code
         assert error.NEU_ERR_TAG_ADDRESS_FORMAT_INVALID == response.json()[
             'error']
 
         response = api.add_tags(node=param[0], group='group', tags=wrong_tag4)
-        assert 200 == response.status_code
+        assert 400 == response.status_code
         assert error.NEU_ERR_TAG_ADDRESS_FORMAT_INVALID == response.json()[
             'error']
 
         response = api.add_tags(node=param[0], group='group', tags=wrong_tag5)
-        assert 200 == response.status_code
+        assert 400 == response.status_code
         assert error.NEU_ERR_TAG_ADDRESS_FORMAT_INVALID == response.json()[
             'error']
 
@@ -490,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(
@@ -567,7 +605,7 @@ class TestModbus:
         up_tag[0]['type'] = config.NEU_TYPE_BIT
         response = api.update_tags(
             node=param[0], group='group', tags=up_tag)
-        assert 200 == response.status_code
+        assert 400 == response.status_code
         assert error.NEU_ERR_TAG_ATTRIBUTE_NOT_SUPPORT == response.json()[
             'error']
 
@@ -633,6 +671,8 @@ class TestModbus:
     def test_add_decimal_tag(self, param):
         api.add_tags_check(
             node=param[0], group='group', tags=hold_int16_decimal)
+        api.add_tags_check(
+            node=param[0], group='group', tags=hold_int16_decimal_neg)
         api.add_tags_check(
             node=param[0], group='group', tags=hold_uint16_decimal)
         api.add_tags_check(
@@ -641,6 +681,8 @@ class TestModbus:
             node=param[0], group='group', tags=hold_uint32_decimal)
         api.add_tags_check(
             node=param[0], group='group', tags=hold_float_decimal)
+        api.add_tags_check(
+            node=param[0], group='group', tags=hold_float_decimal_neg)
         api.add_tags_check(
             node=param[0], group='group', tags=hold_double_decimal)
 
@@ -648,6 +690,8 @@ class TestModbus:
     def test_read_write_decimal_tag(self, param):
         api.write_tag_check(
             node=param[0], group='group', tag=hold_int16_decimal[0]['name'], value=11)
+        api.write_tag_check(
+            node=param[0], group='group', tag=hold_int16_decimal_neg[0]['name'], value=345)
         api.write_tag_check(
             node=param[0], group='group', tag=hold_uint16_decimal[0]['name'], value=23)
         api.write_tag_check(
@@ -656,12 +700,16 @@ class TestModbus:
             node=param[0], group='group', tag=hold_uint32_decimal[0]['name'], value=123)
         api.write_tag_check(
             node=param[0], group='group', tag=hold_float_decimal[0]['name'], value=121.314)
+        api.write_tag_check(
+            node=param[0], group='group', tag=hold_float_decimal_neg[0]['name'], value=345.6)
         api.write_tag_check(
             node=param[0], group='group', tag=hold_double_decimal[0]['name'], value=513.11)
 
         time.sleep(0.3)
         assert 11 == api.read_tag(
             node=param[0], group='group', tag=hold_int16_decimal[0]['name'])
+        assert 345 == api.read_tag(
+            node=param[0], group='group', tag=hold_int16_decimal_neg[0]['name'])
         assert 23 == api.read_tag(
             node=param[0], group='group', tag=hold_uint16_decimal[0]['name'])
         assert 123 == api.read_tag(
@@ -670,6 +718,8 @@ class TestModbus:
             node=param[0], group='group', tag=hold_uint32_decimal[0]['name'])
         compare_float(121.314, api.read_tag(
             node=param[0], group='group', tag=hold_float_decimal[0]['name']))
+        compare_float(345.6, api.read_tag(
+            node=param[0], group='group', tag=hold_float_decimal[0]['name']))
         assert 513.11 == api.read_tag(
             node=param[0], group='group', tag=hold_double_decimal[0]['name'])
 

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})

tests/ft/plugin/test_plugin.py

@@ -228,6 +228,24 @@ class TestPlugin:
         assert 400 == response.status_code
         assert NEU_ERR_BODY_IS_WRONG == response.json().get("error")
 
+    @description(given="incorrect configuration", when="update plugin", then="success")
+    def test_09_update_plugin_success0(self):
+        test_data = TestPlugin.test_data
+        plugin_data = {}
+        plugin_data['library'] = test_data['library_c1']
+        plugin_data['so_file'] = test_data['so_file_c1_2']
+        plugin_data['schema_file'] = test_data['schema_file']
+
+        response = api.updata_plugin(
+            library_name=plugin_data['library'], so_file=plugin_data['so_file'], schema_file=plugin_data['schema_file'])
+        assert 200 == response.status_code
+        assert NEU_ERR_SUCCESS == response.json().get("error")
+        response = api.get_plugins()
+        assert 200 == response.status_code
+        for plugin in response.json().get("plugins"):
+            if plugin['name'] == 'c1':
+                assert plugin['version'] == '2.7.99'
+
     @description(given="incorrect configuration", when="update plugin", then="fail")
     def test_10_update_plugin_fail3(self):
         test_data = TestPlugin.test_data

tests/ft/tag/test_tag.py

@@ -279,7 +279,7 @@ class TestLog:
             ]
         )
 
-        assert 200 == response.status_code
+        assert 400 == response.status_code
         assert NEU_ERR_TAG_TYPE_NOT_SUPPORT == response.json()['error']
 
         response = api.get_group()
@@ -342,7 +342,7 @@ class TestLog:
             ]
         )
 
-        assert 200 == response.status_code
+        assert 409 == response.status_code
         assert NEU_ERR_TAG_NAME_CONFLICT == response.json()['error']
 
         response = api.get_group()

tests/plugins/c1/c1_2.c

@@ -63,7 +63,7 @@ static const neu_plugin_intf_funs_t plugin_intf_funs = {
 };
 
 const neu_plugin_module_t neu_plugin_module = {
-    .version         = NEURON_PLUGIN_VER_1_0,
+    .version         = NEU_VERSION(2, 7, 99),
     .schema          = "c1",
     .module_name     = "c1",
     .module_descr    = "c1 v2",

version

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