Merge pull request #1949 from fengzeroz/v2.8

version: 2.8.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.8
 
     steps:
       - uses: actions/checkout@v4
@@ -60,24 +60,9 @@ jobs:
           cd build
           ctest --output-on-failure
 
-      - name: create cov report
-        run: |
-            sudo apt-get update
-            sudo apt-get -y install lcov
-            mkdir -p cov_report
-            ./create_cov_report.sh ut
-
-      - name: Upload coverage to Codecov
-        uses: codecov/codecov-action@v3
-        with:
-          verbose: true
-          fail_ci_if_error: true
-          directory: ./cov_report/
-          files: cov-ut.info
-
   ft:
     runs-on: ubuntu-20.04
-    container: ghcr.io/neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.8
 
     steps:
       - uses: actions/checkout@v4
@@ -105,24 +90,9 @@ jobs:
           path: |
             build/neuron.log
 
-      - name: create cov report
-        run: |
-            sudo apt-get update
-            sudo apt-get -y install lcov
-            mkdir -p cov_report
-            ./create_cov_report.sh ft
-
-      - name: Upload coverage to Codecov
-        uses: codecov/codecov-action@v3
-        with:
-          verbose: true
-          fail_ci_if_error: true
-          directory: ./cov_report/
-          files: cov-ft.info
-
   pft:
     runs-on: ubuntu-20.04
-    container: ghcr.io/neugates/build:x86_64-main
+    container: ghcr.io/neugates/build:x86_64-v2.8
     strategy:
       matrix:
         plugin: [core, ekuiper, modbus, mqtt, metrics]
@@ -164,17 +134,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
+  VERSION: 2.8.1
   UI_VERSION: 2.6.1
-  BRANCH: main
+  BRANCH: v2.8
 
 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.8
 
     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.8
 
     steps:
       - uses: actions/checkout@v4

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.8/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.8/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.8/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.8/libs)
 
 set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
 set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)

include/neuron/define.h

@@ -23,8 +23,8 @@
 #include <stdbool.h>
 
 #define NEU_VERSION_MAJOR 2
-#define NEU_VERSION_MINOR 7
-#define NEU_VERSION_FIX 0
+#define NEU_VERSION_MINOR 8
+#define NEU_VERSION_FIX 1
 
 #define NEU_GET_VERSION_MAJOR(V) ((0xffff0000 & V) >> 16)
 #define NEU_GET_VERSION_MINOR(V) ((0x0000ff00 & V) >> 8)

include/neuron/errcodes.h

@@ -82,7 +82,6 @@ typedef enum {
     NEU_ERR_TAG_DESCRIPTION_TOO_LONG   = 2208,
     NEU_ERR_TAG_PRECISION_INVALID      = 2209,
     NEU_ERR_TAG_EXIST                  = 2210,
-    NEU_ERR_TAG_VALUE_INVALID          = 2211,
 
     NEU_ERR_LIBRARY_NOT_FOUND                 = 2301,
     NEU_ERR_LIBRARY_INFO_INVALID              = 2302,

include/neuron/msg.h

@@ -859,6 +859,8 @@ static inline void neu_tag_value_to_json(neu_resp_tag_value_meta_t *tag_value,
         tag_json->value.val_bit = tag_value->value.value.u8;
         break;
     case NEU_TYPE_STRING:
+    case NEU_TYPE_TIME:
+    case NEU_TYPE_DATA_AND_TIME:
         tag_json->t             = NEU_JSON_STR;
         tag_json->value.val_str = tag_value->value.value.str;
         break;

include/neuron/tag.h

@@ -59,6 +59,7 @@ typedef union {
     } value16;
     struct {
         neu_datatag_endian_e endian;
+        bool                 is_default;
     } value32;
     struct {
         neu_datatag_endian_e endian;
@@ -66,6 +67,7 @@ typedef union {
     struct {
         uint16_t                  length;
         neu_datatag_string_type_e type;
+        bool                      is_default;
     } string;
 
     struct {

include/neuron/type.h

@@ -31,25 +31,27 @@ extern "C" {
 #include <stdio.h>
 
 typedef enum {
-    NEU_TYPE_INT8   = 1,
-    NEU_TYPE_UINT8  = 2,
-    NEU_TYPE_INT16  = 3,
-    NEU_TYPE_UINT16 = 4,
-    NEU_TYPE_INT32  = 5,
-    NEU_TYPE_UINT32 = 6,
-    NEU_TYPE_INT64  = 7,
-    NEU_TYPE_UINT64 = 8,
-    NEU_TYPE_FLOAT  = 9,
-    NEU_TYPE_DOUBLE = 10,
-    NEU_TYPE_BIT    = 11,
-    NEU_TYPE_BOOL   = 12,
-    NEU_TYPE_STRING = 13,
-    NEU_TYPE_BYTES  = 14,
-    NEU_TYPE_ERROR  = 15,
-    NEU_TYPE_WORD   = 16,
-    NEU_TYPE_DWORD  = 17,
-    NEU_TYPE_LWORD  = 18,
-    NEU_TYPE_PTR    = 19,
+    NEU_TYPE_INT8          = 1,
+    NEU_TYPE_UINT8         = 2,
+    NEU_TYPE_INT16         = 3,
+    NEU_TYPE_UINT16        = 4,
+    NEU_TYPE_INT32         = 5,
+    NEU_TYPE_UINT32        = 6,
+    NEU_TYPE_INT64         = 7,
+    NEU_TYPE_UINT64        = 8,
+    NEU_TYPE_FLOAT         = 9,
+    NEU_TYPE_DOUBLE        = 10,
+    NEU_TYPE_BIT           = 11,
+    NEU_TYPE_BOOL          = 12,
+    NEU_TYPE_STRING        = 13,
+    NEU_TYPE_BYTES         = 14,
+    NEU_TYPE_ERROR         = 15,
+    NEU_TYPE_WORD          = 16,
+    NEU_TYPE_DWORD         = 17,
+    NEU_TYPE_LWORD         = 18,
+    NEU_TYPE_PTR           = 19,
+    NEU_TYPE_TIME          = 20,
+    NEU_TYPE_DATA_AND_TIME = 21,
 } neu_type_e;
 
 inline static const char *neu_type_string(neu_type_e type)
@@ -93,6 +95,10 @@ inline static const char *neu_type_string(neu_type_e type)
         return "NEU_TYPE_LWORD";
     case NEU_TYPE_PTR:
         return "NEU_TYPE_PTR";
+    case NEU_TYPE_TIME:
+        return "NEU_TYPE_TIME";
+    case NEU_TYPE_DATA_AND_TIME:
+        return "NEU_TYPE_DATA_AND_TIME";
     }
 
     return "";
@@ -188,7 +194,9 @@ static inline char *neu_value_str(neu_type_e type, neu_value_u value)
         snprintf(str, sizeof(str), "type: %s, value: %u", neu_type_string(type),
                  value.boolean);
         break;
-    case NEU_TYPE_STRING: {
+    case NEU_TYPE_STRING:
+    case NEU_TYPE_TIME:
+    case NEU_TYPE_DATA_AND_TIME: {
         snprintf(str, sizeof(str), "type: %s, value: %c%c%c",
                  neu_type_string(type), value.str[0], value.str[1],
                  value.str[2]);

persistence/0070_2.7.5_tag_time_type.sql

@@ -0,0 +1,43 @@
+/**
+* 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.
+**/
+BEGIN TRANSACTION;
+
+ALTER TABLE tags RENAME TO temp_tags;
+
+CREATE TABLE
+  IF NOT EXISTS tags (
+    driver_name TEXT NOT NULL,
+    group_name TEXT NOT NULL,
+    name TEXT NULL check (length (name) <= 128),
+    address TEXT NULL check (length (address) <= 128),
+    attribute INTEGER NOT NULL check (attribute BETWEEN 0 AND 15),
+    precision INTEGER NOT NULL check (precision BETWEEN 0 AND 17),
+    decimal REAL NOT NULL,
+    type INTEGER NOT NULL check (type BETWEEN 0 AND 21),
+    description TEXT NULL check (length (description) <= 512),
+    value TEXT,
+    UNIQUE (driver_name, group_name, name),
+    FOREIGN KEY (driver_name, group_name) REFERENCES groups (driver_name, name) ON UPDATE CASCADE ON DELETE CASCADE
+  );
+
+INSERT INTO tags SELECT * FROM temp_tags;
+
+DROP TABLE temp_tags;
+
+COMMIT;

plugins/ekuiper/plugin_ekuiper.c

@@ -162,7 +162,16 @@ static inline int start(neu_plugin_t *plugin, const char *url)
         return rv;
     }
 
+    nng_mtx_lock(plugin->mtx);
+    while (plugin->receiving) {
+        nng_mtx_unlock(plugin->mtx);
+        nng_msleep(10);
+        nng_mtx_lock(plugin->mtx);
+    }
+    plugin->receiving = true;
     nng_recv_aio(plugin->sock, plugin->recv_aio);
+    nng_mtx_unlock(plugin->mtx);
+
     return NEU_ERR_SUCCESS;
 }
 
@@ -333,6 +342,8 @@ static int ekuiper_plugin_request(neu_plugin_t *      plugin,
     case NEU_REQ_UPDATE_GROUP: {
         break;
     }
+    case NEU_REQ_DEL_GROUP:
+        break;
     case NEU_REQ_SUBSCRIBE_GROUP:
     case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: {
         neu_req_subscribe_t *sub_info = data;

plugins/ekuiper/plugin_ekuiper.h

@@ -37,6 +37,7 @@ struct neu_plugin {
     nng_socket          sock;
     nng_mtx *           mtx;
     bool                started;
+    bool                receiving;
     nng_aio *           recv_aio;
     char *              host;
     uint16_t            port;

plugins/ekuiper/read_write.c

@@ -83,6 +83,9 @@ void recv_data_callback(void *arg)
     rv = nng_aio_result(plugin->recv_aio);
     if (0 != rv) {
         plog_error(plugin, "nng_recv error: %s", nng_strerror(rv));
+        nng_mtx_lock(plugin->mtx);
+        plugin->receiving = false;
+        nng_mtx_unlock(plugin->mtx);
         return;
     }
 

plugins/modbus/modbus_req.c

@@ -699,7 +699,7 @@ static int process_protocol_buf(neu_plugin_t *plugin, uint8_t slave_id,
                 plog_recv_protocol(plugin, recv_buf, ret);
             }
             neu_protocol_unpack_buf_init(&pbuf, recv_buf, ret);
-            int ret_s = modbus_stack_recv(plugin->stack, 0, &pbuf);
+            int ret_s = modbus_stack_recv(plugin->stack, slave_id, &pbuf);
             if (ret_s == MODBUS_DEVICE_ERR) {
                 ret = ret_s;
             } else {

plugins/modbus/modbus_stack.c

@@ -84,10 +84,8 @@ int modbus_stack_recv(modbus_stack_t *stack, uint8_t slave_id,
         return -1;
     }
 
-    if (stack->protocol == MODBUS_PROTOCOL_TCP) {
-        if (code.slave_id != slave_id) {
-            return -1;
-        }
+    if (code.slave_id != slave_id) {
+        return -1;
     }
 
     switch (code.function) {
@@ -102,18 +100,36 @@ int modbus_stack_recv(modbus_stack_t *stack, uint8_t slave_id,
             return -1;
         }
 
-        if (stack->protocol == MODBUS_PROTOCOL_TCP && data.n_byte == 0xff) {
-            bytes = neu_protocol_unpack_buf(buf,
-                                            header.len -
-                                                sizeof(struct modbus_code) -
-                                                sizeof(struct modbus_data));
-            stack->value_fn(stack->ctx, code.slave_id,
-                            header.len - sizeof(struct modbus_code) -
-                                sizeof(struct modbus_data),
-                            bytes, 0);
-        } else {
+        switch (stack->protocol) {
+        case MODBUS_PROTOCOL_TCP:
+            if (data.n_byte == 0xff) {
+                bytes = neu_protocol_unpack_buf(buf,
+                                                header.len -
+                                                    sizeof(struct modbus_code) -
+                                                    sizeof(struct modbus_data));
+                if (bytes == NULL) {
+                    return -1;
+                }
+                stack->value_fn(stack->ctx, code.slave_id,
+                                header.len - sizeof(struct modbus_code) -
+                                    sizeof(struct modbus_data),
+                                bytes, 0);
+            } else {
+                bytes = neu_protocol_unpack_buf(buf, data.n_byte);
+                if (bytes == NULL) {
+                    return -1;
+                }
+                stack->value_fn(stack->ctx, code.slave_id, data.n_byte, bytes,
+                                0);
+            }
+            break;
+        case MODBUS_PROTOCOL_RTU:
             bytes = neu_protocol_unpack_buf(buf, data.n_byte);
+            if (bytes == NULL) {
+                return -1;
+            }
             stack->value_fn(stack->ctx, code.slave_id, data.n_byte, bytes, 0);
+            break;
         }
 
         break;

plugins/mqtt/mqtt.json

@@ -6,7 +6,6 @@
     "description_zh": "MQTT 客户端 ID",
     "type": "string",
     "attribute": "required",
-    "default": "neuron_${random_str}",
     "valid": {
       "length": 255
     }
@@ -64,7 +63,6 @@
     "description_zh": "接收点位写入请求的 MQTT 主题。",
     "attribute": "required",
     "type": "string",
-    "default": "/neuron/${node-name}/write/req",
     "valid": {
       "length": 255
     }
@@ -76,7 +74,6 @@
     "description_zh": "发送点位写入响应的 MQTT 主题。",
     "attribute": "required",
     "type": "string",
-    "default": "/neuron/${node-name}/write/resp",
     "valid": {
       "length": 255
     }

plugins/mqtt/mqtt_handle.c

@@ -606,6 +606,14 @@ int handle_unsubscribe_group(neu_plugin_t *         plugin,
     return 0;
 }
 
+int handle_del_group(neu_plugin_t *plugin, neu_req_del_group_t *req)
+{
+    route_tbl_del(&plugin->route_tbl, req->driver, req->group);
+    plog_notice(plugin, "del route driver:%s group:%s", req->driver,
+                req->group);
+    return 0;
+}
+
 int handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req)
 {
     route_tbl_update_group(&plugin->route_tbl, req->driver, req->group,

plugins/mqtt/mqtt_handle.h

@@ -49,6 +49,7 @@ int handle_unsubscribe_group(neu_plugin_t *         plugin,
                              neu_req_unsubscribe_t *unsub_info);
 
 int handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req);
+int handle_del_group(neu_plugin_t *plugin, neu_req_del_group_t *cmd);
 
 int handle_update_driver(neu_plugin_t *plugin, neu_req_update_node_t *req);
 int handle_del_driver(neu_plugin_t *plugin, neu_reqresp_node_deleted_t *req);

plugins/mqtt/mqtt_plugin.c

@@ -384,6 +384,9 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
     case NEU_REQ_UPDATE_GROUP:
         error = handle_update_group(plugin, data);
         break;
+    case NEU_REQ_DEL_GROUP:
+        error = handle_del_group(plugin, data);
+        break;
     case NEU_REQ_UPDATE_NODE:
         error = handle_update_driver(plugin, data);
         break;

src/adapter/adapter.c

@@ -152,18 +152,38 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load)
         break;
     }
 
-    adapter->control_fd = socket(AF_UNIX, SOCK_DGRAM | SOCK_NONBLOCK, 0);
+    adapter->control_fd = socket(AF_UNIX, SOCK_DGRAM, 0);
     if (adapter->control_fd <= 0) {
         free(adapter);
         return NULL;
     }
-    adapter->trans_data_fd = socket(AF_UNIX, SOCK_DGRAM | SOCK_NONBLOCK, 0);
+
+    adapter->trans_data_fd = socket(AF_UNIX, SOCK_DGRAM, 0);
     if (adapter->trans_data_fd <= 0) {
         close(adapter->control_fd);
         free(adapter);
         return NULL;
     }
 
+    struct timeval sock_timeout = {
+        .tv_sec  = 1,
+        .tv_usec = 0,
+    };
+    if (setsockopt(adapter->control_fd, SOL_SOCKET, SO_SNDTIMEO, &sock_timeout,
+                   sizeof(sock_timeout)) < 0 ||
+        setsockopt(adapter->control_fd, SOL_SOCKET, SO_RCVTIMEO, &sock_timeout,
+                   sizeof(sock_timeout)) < 0 ||
+        setsockopt(adapter->trans_data_fd, SOL_SOCKET, SO_SNDTIMEO,
+                   &sock_timeout, sizeof(sock_timeout)) < 0 ||
+        setsockopt(adapter->trans_data_fd, SOL_SOCKET, SO_RCVTIMEO,
+                   &sock_timeout, sizeof(sock_timeout)) < 0) {
+        nlog_error("fail to set sock timeout for adapter:%s", info->name);
+        close(adapter->trans_data_fd);
+        close(adapter->control_fd);
+        free(adapter);
+        return NULL;
+    }
+
     adapter->name                    = strdup(info->name);
     adapter->events                  = neu_event_new();
     adapter->state                   = NEU_NODE_RUNNING_STATE_INIT;
@@ -911,7 +931,10 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
             error.error = neu_adapter_driver_del_group(
                 (neu_adapter_driver_t *) adapter, cmd->group);
         } else {
-            error.error = NEU_ERR_GROUP_NOT_ALLOW;
+            adapter->module->intf_funs->request(
+                adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
+            neu_msg_free(msg);
+            break;
         }
 
         if (error.error == NEU_ERR_SUCCESS) {
@@ -1048,12 +1071,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
 
         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 (neu_adapter_validate_gtags(adapter, cmd, &resp) == 0 &&
                 neu_adapter_try_add_gtags(adapter, cmd, &resp) == 0 &&
@@ -1184,22 +1201,30 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
 int neu_adapter_validate_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
                                neu_resp_add_tag_t *resp)
 {
+    neu_adapter_driver_t *driver_adapter = (neu_adapter_driver_t *) adapter;
+    if (neu_adapter_driver_new_group_count(driver_adapter, cmd) +
+            neu_adapter_driver_group_count(driver_adapter) >
+        NEU_GROUP_MAX_PER_NODE) {
+        resp->error = NEU_ERR_GROUP_MAX_GROUPS;
+        resp->index = 0;
+        return NEU_ERR_GROUP_MAX_GROUPS;
+    }
+
     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++) {
+        neu_gdatatag_t *current_group = &cmd->groups[group_index];
+        for (int tag_index = 0; tag_index < current_group->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 {
+                driver_adapter, current_group->group,
+                &current_group->tags[tag_index]);
+            if (validation_result != 0) {
                 resp->error = validation_result;
                 resp->index = 0;
                 return validation_result;
             }
+            resp->index += 1;
         }
     }
+
     return 0;
 }
 
@@ -1233,6 +1258,8 @@ int neu_adapter_add_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
         neu_adapter_driver_add_group((neu_adapter_driver_t *) adapter,
                                      cmd->groups[group_index].group,
                                      cmd->groups[group_index].interval);
+        adapter_storage_add_group(adapter->name, cmd->groups[group_index].group,
+                                  cmd->groups[group_index].interval);
         for (int tag_index = 0; tag_index < cmd->groups[group_index].n_tag;
              tag_index++) {
             int add_tag_result = neu_adapter_driver_add_tag(
@@ -1262,12 +1289,17 @@ int neu_adapter_add_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
                         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[group_index].n_tag -
+                                                   tag_index);
+
+                for (++group_index; group_index < cmd->n_group; group_index++) {
                     neu_adapter_driver_try_del_tag(
                         (neu_adapter_driver_t *) adapter,
-                        cmd->groups[groups_count].n_tag);
+                        cmd->groups[group_index].n_tag);
                 }
+
                 resp->index = 0;
                 resp->error = add_tag_result;
                 return add_tag_result;

src/adapter/driver/cache.c

@@ -157,6 +157,8 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache,
             case NEU_TYPE_BIT:
             case NEU_TYPE_BOOL:
             case NEU_TYPE_STRING:
+            case NEU_TYPE_TIME:
+            case NEU_TYPE_DATA_AND_TIME:
             case NEU_TYPE_WORD:
             case NEU_TYPE_DWORD:
             case NEU_TYPE_LWORD:
@@ -299,6 +301,8 @@ int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group,
             value->value.value.boolean = elem->value.value.boolean;
             break;
         case NEU_TYPE_STRING:
+        case NEU_TYPE_TIME:
+        case NEU_TYPE_DATA_AND_TIME:
             memcpy(value->value.value.str, elem->value.value.str,
                    sizeof(elem->value.value.str));
             break;
@@ -381,6 +385,8 @@ int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache,
             value->value.value.boolean = elem->value.value.boolean;
             break;
         case NEU_TYPE_STRING:
+        case NEU_TYPE_TIME:
+        case NEU_TYPE_DATA_AND_TIME:
             memcpy(value->value.value.str, elem->value.value.str,
                    sizeof(elem->value.value.str));
             break;

src/adapter/driver/driver.c

@@ -472,6 +472,8 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type,
     switch (tag->type) {
     case NEU_TYPE_BOOL:
     case NEU_TYPE_STRING:
+    case NEU_TYPE_TIME:
+    case NEU_TYPE_DATA_AND_TIME:
     case NEU_TYPE_BYTES:
         break;
     case NEU_TYPE_BIT:
@@ -637,26 +639,6 @@ static void cal_decimal(neu_type_e tag_type, neu_type_e value_type,
     }
 }
 
-bool is_value_in_range(neu_type_e type, int64_t value)
-{
-    switch (type) {
-    case NEU_TYPE_INT8:
-        return value >= INT8_MIN && value <= INT8_MAX;
-    case NEU_TYPE_UINT8:
-        return value >= 0 && value <= UINT8_MAX;
-    case NEU_TYPE_INT16:
-        return value >= INT16_MIN && value <= INT16_MAX;
-    case NEU_TYPE_UINT16:
-        return value >= 0 && value <= UINT16_MAX;
-    case NEU_TYPE_INT32:
-        return value >= INT32_MIN && value <= INT32_MAX;
-    case NEU_TYPE_UINT32:
-        return value >= 0 && value <= UINT32_MAX;
-    default:
-        return true;
-    }
-}
-
 void neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
                                    neu_reqresp_head_t *  req)
 {
@@ -688,8 +670,7 @@ void neu_adapter_driver_write_tags(neu_adapter_driver_t *driver,
     for (int i = 0; i < cmd->n_tag; i++) {
         neu_datatag_t *tag = neu_group_find_tag(g->group, cmd->tags[i].tag);
         if (tag != NULL && neu_tag_attribute_test(tag, NEU_ATTRIBUTE_WRITE) &&
-            neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) == false &&
-            is_value_in_range(tag->type, cmd->tags[i].value.value.i64)) {
+            neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) == false) {
             if (tag->type == NEU_TYPE_FLOAT || tag->type == NEU_TYPE_DOUBLE) {
                 if (cmd->tags[i].value.type == NEU_TYPE_INT64) {
                     cmd->tags[i].value.value.d64 =
@@ -776,9 +757,7 @@ void neu_adapter_driver_write_gtags(neu_adapter_driver_t *driver,
                 neu_group_find_tag(g->group, cmd->groups[i].tags[k].tag);
             if (tag != NULL &&
                 neu_tag_attribute_test(tag, NEU_ATTRIBUTE_WRITE) &&
-                neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) == false &&
-                is_value_in_range(tag->type,
-                                  cmd->groups[i].tags[k].value.value.i64)) {
+                neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) == false) {
 
                 if (tag->type == NEU_TYPE_FLOAT ||
                     tag->type == NEU_TYPE_DOUBLE) {
@@ -860,12 +839,6 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
             neu_tag_free(tag);
             return;
         }
-        if (!is_value_in_range(tag->type, cmd->value.value.i64)) {
-            driver->adapter.cb_funs.driver.write_response(
-                &driver->adapter, req, NEU_ERR_TAG_VALUE_INVALID);
-            neu_tag_free(tag);
-            return;
-        }
         if (tag->type == NEU_TYPE_FLOAT || tag->type == NEU_TYPE_DOUBLE) {
             if (cmd->value.type == NEU_TYPE_INT64) {
                 cmd->value.value.d64 = (double) cmd->value.value.i64;

src/base/tag.c

@@ -161,19 +161,24 @@ int neu_datatag_parse_addr_option(const neu_datatag_t *      datatag,
 
             switch (t) {
             case 'H':
-                option->string.type = NEU_DATATAG_STRING_TYPE_H;
+                option->string.type       = NEU_DATATAG_STRING_TYPE_H;
+                option->string.is_default = false;
                 break;
             case 'L':
-                option->string.type = NEU_DATATAG_STRING_TYPE_L;
+                option->string.type       = NEU_DATATAG_STRING_TYPE_L;
+                option->string.is_default = false;
                 break;
             case 'D':
-                option->string.type = NEU_DATATAG_STRING_TYPE_D;
+                option->string.type       = NEU_DATATAG_STRING_TYPE_D;
+                option->string.is_default = false;
                 break;
             case 'E':
-                option->string.type = NEU_DATATAG_STRING_TYPE_D;
+                option->string.type       = NEU_DATATAG_STRING_TYPE_E;
+                option->string.is_default = false;
                 break;
             default:
-                option->string.type = NEU_DATATAG_STRING_TYPE_H;
+                option->string.type       = NEU_DATATAG_STRING_TYPE_H;
+                option->string.is_default = true;
                 break;
             }
 
@@ -210,10 +215,13 @@ int neu_datatag_parse_addr_option(const neu_datatag_t *      datatag,
     }
     case NEU_TYPE_FLOAT:
     case NEU_TYPE_INT32:
-    case NEU_TYPE_UINT32: {
+    case NEU_TYPE_UINT32:
+    case NEU_TYPE_DATA_AND_TIME:
+    case NEU_TYPE_TIME: {
         char *op = find_last_character(datatag->address, '#');
 
-        option->value32.endian = NEU_DATATAG_ENDIAN_LL32;
+        option->value32.endian     = NEU_DATATAG_ENDIAN_LL32;
+        option->value32.is_default = true;
         if (op != NULL) {
             char e1 = 0;
             char e2 = 0;
@@ -221,16 +229,20 @@ int neu_datatag_parse_addr_option(const neu_datatag_t *      datatag,
 
             if (n == 2) {
                 if (e1 == 'B' && e2 == 'B') {
-                    option->value32.endian = NEU_DATATAG_ENDIAN_BB32;
+                    option->value32.endian     = NEU_DATATAG_ENDIAN_BB32;
+                    option->value32.is_default = false;
                 }
                 if (e1 == 'B' && e2 == 'L') {
-                    option->value32.endian = NEU_DATATAG_ENDIAN_BL32;
+                    option->value32.endian     = NEU_DATATAG_ENDIAN_BL32;
+                    option->value32.is_default = false;
                 }
                 if (e1 == 'L' && e2 == 'L') {
-                    option->value32.endian = NEU_DATATAG_ENDIAN_LL32;
+                    option->value32.endian     = NEU_DATATAG_ENDIAN_LL32;
+                    option->value32.is_default = false;
                 }
                 if (e1 == 'L' && e2 == 'B') {
-                    option->value32.endian = NEU_DATATAG_ENDIAN_LB32;
+                    option->value32.endian     = NEU_DATATAG_ENDIAN_LB32;
+                    option->value32.is_default = false;
                 }
             }
         }
@@ -523,6 +535,8 @@ int neu_tag_get_static_value_json(neu_datatag_t *tag, neu_json_type_e *t,
         v->val_double = value->d64;
         break;
     case NEU_TYPE_STRING:
+    case NEU_TYPE_TIME:
+    case NEU_TYPE_DATA_AND_TIME:
         *t         = NEU_JSON_STR;
         v->val_str = value->str;
         break;
@@ -625,7 +639,8 @@ int neu_tag_set_static_value_json(neu_datatag_t *tag, neu_json_type_e t,
         break;
     }
     case NEU_JSON_STR: {
-        if (NEU_TYPE_STRING == tag->type) {
+        if (NEU_TYPE_STRING == tag->type || NEU_TYPE_TIME == tag->type ||
+            NEU_TYPE_DATA_AND_TIME == tag->type) {
             snprintf(value.str, sizeof(value.str), "%s", v->val_str);
         } else {
             rv = -1;
@@ -695,6 +710,8 @@ char *neu_tag_dump_static_value(const neu_datatag_t *tag)
         jval = json_real(value.d64);
         break;
     case NEU_TYPE_STRING:
+    case NEU_TYPE_TIME:
+    case NEU_TYPE_DATA_AND_TIME:
         jval = json_string(value.str);
         break;
     default:

src/connection/connection.c

@@ -1290,6 +1290,12 @@ int neu_conn_stream_consume(neu_conn_t *conn, void *context,
                 break;
             } else {
                 pthread_mutex_lock(&conn->mtx);
+                if (conn->offset - used < 0) {
+                    pthread_mutex_unlock(&conn->mtx);
+                    zlog_warn(conn->param.log, "reset offset: %d, used: %d",
+                              conn->offset, used);
+                    return -1;
+                }
                 conn->offset -= used;
                 memmove(conn->buf, conn->buf + used, conn->offset);
                 neu_protocol_unpack_buf_init(&protocol_buf, conn->buf,
@@ -1320,10 +1326,18 @@ int neu_conn_stream_tcp_server_consume(neu_conn_t *conn, int fd, void *context,
                 neu_conn_tcp_server_close_client(conn, fd);
                 break;
             } else {
+                pthread_mutex_lock(&conn->mtx);
+                if (conn->offset - used < 0) {
+                    pthread_mutex_unlock(&conn->mtx);
+                    zlog_warn(conn->param.log, "reset offset: %d, used: %d",
+                              conn->offset, used);
+                    return -1;
+                }
                 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);
             }
         }
     }

src/core/manager.c

@@ -99,9 +99,20 @@ neu_manager_t *neu_manager_create()
     manager->subscribe_manager = neu_subscribe_manager_create();
     manager->log_level         = ZLOG_LEVEL_NOTICE;
 
-    manager->server_fd = socket(AF_UNIX, SOCK_DGRAM | SOCK_NONBLOCK, 0);
+    manager->server_fd = socket(AF_UNIX, SOCK_DGRAM, 0);
     assert(manager->server_fd > 0);
 
+    struct timeval sock_timeout = {
+        .tv_sec  = 1,
+        .tv_usec = 0,
+    };
+    if (setsockopt(manager->server_fd, SOL_SOCKET, SO_SNDTIMEO, &sock_timeout,
+                   sizeof(sock_timeout)) < 0 ||
+        setsockopt(manager->server_fd, SOL_SOCKET, SO_RCVTIMEO, &sock_timeout,
+                   sizeof(sock_timeout)) < 0) {
+        assert(!"fail to set manager sock timeout");
+    }
+
     // abstract domain socket is a Linux extension, thus not portable.
     // use abstract domain socket here to avoid polluting the file system
     struct sockaddr_un local = {
@@ -511,6 +522,35 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
                 break;
             }
 
+            if (!neu_plugin_manager_create_instance_by_lib_name(
+                    manager->plugin_manager, cmd->library, &ins)) {
+                nlog_warn("library %s not instance", cmd->library);
+                free(cmd->so_file);
+                free(cmd->schema_file);
+                free(so_tmp_path);
+                header->type = NEU_RESP_ERROR;
+                e.error      = NEU_ERR_LIBRARY_UPDATE_FAIL;
+                strcpy(header->receiver, header->sender);
+                reply(manager, header, &e);
+                break;
+            }
+
+            if (strcmp(module_name, ins.module->module_name) != 0) {
+                nlog_warn("library %s module name mismatch!", cmd->library);
+                neu_plugin_manager_destroy_instance(manager->plugin_manager,
+                                                    &ins);
+                free(cmd->so_file);
+                free(cmd->schema_file);
+                free(so_tmp_path);
+                header->type = NEU_RESP_ERROR;
+                e.error      = NEU_ERR_LIBRARY_UPDATE_FAIL;
+                strcpy(header->receiver, header->sender);
+                reply(manager, header, &e);
+                break;
+            }
+
+            neu_plugin_manager_destroy_instance(manager->plugin_manager, &ins);
+
             if (!neu_plugin_manager_exists(manager->plugin_manager,
                                            module_name)) {
 
@@ -1098,9 +1138,24 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
             neu_msg_exchange(header);
             reply(manager, header, &e);
         } else {
+            UT_array *apps = neu_subscribe_manager_find(
+                manager->subscribe_manager, cmd->driver, cmd->group);
+
             forward_msg(manager, header, header->receiver);
             neu_subscribe_manager_remove(manager->subscribe_manager,
                                          cmd->driver, cmd->group);
+
+            if (NULL == apps) {
+                break;
+            }
+
+            // notify app node about group deletion
+            utarray_foreach(apps, neu_app_subscribe_t *, app)
+            {
+                forward_msg_copy(manager, header, app->app_name);
+            }
+
+            utarray_free(apps);
         }
         break;
     }

src/core/manager_internal.c

@@ -460,6 +460,7 @@ static inline int add_driver(neu_manager_t *manager, neu_req_driver_t *driver)
     if (0 != neu_adapter_validate_gtags(adapter, &cmd, &resp) ||
         0 != neu_adapter_try_add_gtags(adapter, &cmd, &resp) ||
         0 != neu_adapter_add_gtags(adapter, &cmd, &resp)) {
+        neu_adapter_uninit(adapter);
         neu_manager_del_node(manager, driver->node);
     }
 
@@ -498,11 +499,21 @@ int neu_manager_add_drivers(neu_manager_t *manager, neu_req_driver_array_t *req)
     for (uint16_t i = 0; i < req->n_driver; ++i) {
         ret = add_driver(manager, &req->drivers[i]);
         if (0 != ret) {
+            nlog_notice("add i:%" PRIu16 " driver:%s fail", i,
+                        req->drivers[i].node);
             while (i-- > 0) {
+                nlog_notice("rollback i:%" PRIu16 " driver:%s", i,
+                            req->drivers[i].node);
+                neu_adapter_t *adapter = neu_node_manager_find(
+                    manager->node_manager, req->drivers[i].node);
+                neu_adapter_uninit(adapter);
                 neu_manager_del_node(manager, req->drivers[i].node);
             }
+            nlog_error("fail to add %" PRIu16 " drivers", req->n_driver);
             break;
         }
+        nlog_notice("add i:%" PRIu16 " driver:%s success", i,
+                    req->drivers[i].node);
     }
 
     return ret;

src/core/plugin_manager.c

@@ -489,6 +489,38 @@ bool neu_plugin_manager_create_instance_by_path(neu_plugin_manager_t *mgr,
     return true;
 }
 
+bool neu_plugin_manager_create_instance_by_lib_name(
+    neu_plugin_manager_t *mgr, const char *lib_name,
+    neu_plugin_instance_t *instance)
+{
+    (void) mgr;
+    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);
+
+    for (size_t i = 0; i < sizeof(lib_paths) / sizeof(lib_paths[0]); i++) {
+        char lib_path[256] = { 0 };
+
+        snprintf(lib_path, sizeof(lib_path) - 1, "%s/%s", lib_paths[i],
+                 lib_name);
+        instance->handle = dlopen(lib_path, RTLD_NOW);
+
+        if (instance->handle == NULL) {
+            continue;
+        }
+
+        instance->module = (neu_plugin_module_t *) dlsym(instance->handle,
+                                                         "neu_plugin_module");
+        assert(instance->module != NULL);
+        return true;
+    }
+
+    assert(instance->handle != NULL);
+
+    return false;
+}
+
 bool neu_plugin_manager_remove_library(neu_plugin_manager_t *mgr,
                                        const char *          library)
 {

src/core/plugin_manager.h

@@ -68,6 +68,10 @@ bool neu_plugin_manager_create_instance_by_path(neu_plugin_manager_t *mgr,
                                                 neu_plugin_instance_t *instance,
                                                 int *                  error);
 
+bool neu_plugin_manager_create_instance_by_lib_name(
+    neu_plugin_manager_t *mgr, const char *lib_name,
+    neu_plugin_instance_t *instance);
+
 bool neu_plugin_manager_remove_library(neu_plugin_manager_t *mgr,
                                        const char *          library);
 

src/parser/neu_json_rw.c

@@ -407,14 +407,19 @@ void neu_json_decode_write_free(neu_json_write_t *req)
 
 int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result)
 {
-    int                  ret      = 0;
-    void *               json_obj = NULL;
-    neu_json_read_req_t *req      = calloc(1, sizeof(neu_json_read_req_t));
-    if (req == NULL) {
+    int   ret      = 0;
+    void *json_obj = NULL;
+
+    json_obj = neu_json_decode_new(buf);
+    if (NULL == json_obj) {
         return -1;
     }
 
-    json_obj = neu_json_decode_new(buf);
+    neu_json_read_req_t *req = calloc(1, sizeof(neu_json_read_req_t));
+    if (req == NULL) {
+        neu_json_decode_free(json_obj);
+        return -1;
+    }
 
     neu_json_elem_t req_elems[] = {
         {
@@ -436,15 +441,6 @@ int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result)
             .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
         },
     };
-    ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
-                                  req_elems);
-    if (ret != 0) {
-        goto error;
-    }
-
-    req->node  = req_elems[0].v.val_str;
-    req->group = req_elems[1].v.val_str;
-    req->sync  = req_elems[2].v.val_bool;
 
     neu_json_elem_t query_elems[] = {
         {
@@ -458,6 +454,17 @@ int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result)
             .attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
         },
     };
+
+    ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
+                                  req_elems);
+    if (ret != 0) {
+        goto error;
+    }
+
+    req->node  = req_elems[0].v.val_str;
+    req->group = req_elems[1].v.val_str;
+    req->sync  = req_elems[2].v.val_bool;
+
     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),
@@ -579,6 +586,8 @@ void neu_json_metas_to_json(neu_tag_meta_t *metas, int n_meta,
                 json_tag->metas[k].value.val_bit = metas[k].value.value.u8;
                 break;
             case NEU_TYPE_STRING:
+            case NEU_TYPE_TIME:
+            case NEU_TYPE_DATA_AND_TIME:
                 json_tag->metas[k].t             = NEU_JSON_STR;
                 json_tag->metas[k].value.val_str = metas[k].value.value.str;
                 break;

src/parser/neu_json_tag.c

@@ -199,6 +199,8 @@ int neu_json_tag_check_type(neu_json_tag_t *tag)
             return NEU_JSON_BOOL == tag->t;
         case NEU_TYPE_BYTES:
         case NEU_TYPE_STRING:
+        case NEU_TYPE_TIME:
+        case NEU_TYPE_DATA_AND_TIME:
             return NEU_JSON_STR == tag->t &&
                 strlen(tag->value.val_str) <= NEU_VALUE_SIZE;
         }

src/utils/http.c

@@ -320,7 +320,6 @@ 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_TAG_VALUE_INVALID:
     case NEU_ERR_GROUP_MAX_GROUPS:
     case NEU_ERR_LICENSE_MAX_TAGS:
     case NEU_ERR_LICENSE_BAD_CLOCK:

tests/ft/app/test_mqtt.py

@@ -484,6 +484,35 @@ class TestMQTT:
         finally:
             api.update_group_check(DRIVER, new_name, GROUP)
 
+    @description(
+        given="MQTT node",
+        when="delete subscribed driver group and add back group",
+        then="broker should receive data on upload topic",
+    )
+    def test_mqtt_upload_when_group_deleted(self, mocker, conf_fmt_tags):
+        api.node_setting_check(NODE, conf_fmt_tags)
+        api.node_ctl(NODE, config.NEU_CTL_START)
+
+        # delete driver group
+        api.del_group_check(DRIVER, GROUP)
+        # add back driver group
+        api.add_group_check(DRIVER, GROUP, INTERVAL)
+        api.add_tags_check(DRIVER, GROUP, tags=[TAG0])
+        # add back subscription on new topic
+        topic = UPLOAD_TOPIC + "-new"
+        params = {"topic": topic}
+        api.subscribe_group_check(NODE, DRIVER, GROUP, params)
+
+        msg = mocker.get(topic, timeout=3)
+        try:
+            assert msg is not None
+            assert DRIVER == msg["node"]
+            assert GROUP == msg["group"]
+            assert TAG0["name"] in [tag["name"] for tag in msg["tags"]]
+        finally:
+            params["topic"] = UPLOAD_TOPIC
+            api.subscribe_group_update(NODE, DRIVER, GROUP, params)
+
     @description(
         given="MQTT node and conf_cache",
         when="broken network connection restored",

tests/ft/driver/test_modbus.py

@@ -452,80 +452,6 @@ class TestModbus:
         assert error.NEU_ERR_TAG_ADDRESS_FORMAT_INVALID == response.json()[
             'error']
 
-    @description(given="created modbus node and tags", when="write tags with invalid value", then="write failed")
-    def test_write_tags_invalid_value(self, param):
-        response = api.write_tag(node=param[0], group='group', tag=hold_int16[0]['name'], value=32768)
-        assert 400 == response.status_code
-        assert error.NEU_ERR_TAG_VALUE_INVALID == response.json()['error']
-
-        response = api.write_tag(node=param[0], group='group', tag=hold_uint16[0]['name'], value=-1)
-        assert 400 == response.status_code
-        assert error.NEU_ERR_TAG_VALUE_INVALID == response.json()['error']
-
-        response = api.write_tag(node=param[0], group='group', tag=hold_int32[0]['name'], value=2147483648)
-        assert 400 == response.status_code
-        assert error.NEU_ERR_TAG_VALUE_INVALID == response.json()['error']
-
-        response = api.write_tag(node=param[0], group='group', tag=hold_uint32[0]['name'], value=-1)
-        assert 400 == response.status_code
-        assert error.NEU_ERR_TAG_VALUE_INVALID == response.json()['error']
-
-        response = api.write_tags(node=param[0], group='group', tag_values=[
-            {"tag": hold_int16[0]['name'], "value": 32768},
-            {"tag": hold_uint16[0]['name'], "value": 2},
-            {"tag": hold_int32[0]['name'], "value": 3},
-            {"tag": hold_uint32[0]['name'], "value": 4}])
-        assert 404 == response.status_code
-        assert error.NEU_ERR_TAG_NOT_EXIST == response.json()['error']
-
-        response = api.write_tags(node=param[0], group='group', tag_values=[
-            {"tag": hold_int16[0]['name'], "value": 1},
-            {"tag": hold_uint16[0]['name'], "value": -1},
-            {"tag": hold_int32[0]['name'], "value": 3},
-            {"tag": hold_uint32[0]['name'], "value": 4}])
-        assert 404 == response.status_code
-        assert error.NEU_ERR_TAG_NOT_EXIST == response.json()['error']
-
-        response = api.write_tags(node=param[0], group='group', tag_values=[
-            {"tag": hold_int16[0]['name'], "value": 1},
-            {"tag": hold_uint16[0]['name'], "value": 2},
-            {"tag": hold_int32[0]['name'], "value": 2147483648},
-            {"tag": hold_uint32[0]['name'], "value": 4}])
-        assert 404 == response.status_code
-        assert error.NEU_ERR_TAG_NOT_EXIST == response.json()['error']
-
-        response = api.write_tags(node=param[0], group='group', tag_values=[
-            {"tag": hold_int16[0]['name'], "value": 1},
-            {"tag": hold_uint16[0]['name'], "value": 2},
-            {"tag": hold_int32[0]['name'], "value": 3},
-            {"tag": hold_uint32[0]['name'], "value": -1}])
-        assert 404 == response.status_code
-        assert error.NEU_ERR_TAG_NOT_EXIST == response.json()['error']
-
-        modbus_write_gtags = {"node": param[0], "groups": [{"group": "group", "tags" : [{"tag": "hold_int16", "value": 32768},
-            {"tag": "hold_uint16", "value": 2}, {"tag": "hold_int32", "value": 3}, {"tag": "hold_uint32", "value": 4}]}]}
-        response = api.write_gtags(json=modbus_write_gtags)
-        assert 404 == response.status_code
-        assert error.NEU_ERR_TAG_NOT_EXIST == response.json()['error']
-
-        modbus_write_gtags = {"node": param[0], "groups": [{"group": "group", "tags" : [{"tag": "hold_int16", "value": 1},
-            {"tag": "hold_uint16", "value": -1}, {"tag": "hold_int32", "value": 3}, {"tag": "hold_uint32", "value": 4}]}]}
-        response = api.write_gtags(json=modbus_write_gtags)
-        assert 404 == response.status_code
-        assert error.NEU_ERR_TAG_NOT_EXIST == response.json()['error']
-
-        modbus_write_gtags = {"node": param[0], "groups": [{"group": "group", "tags" : [{"tag": "hold_int16", "value": 1},
-            {"tag": "hold_uint16", "value": 2}, {"tag": "hold_int32", "value": 2147483648}, {"tag": "hold_uint32", "value": 4}]}]}
-        response = api.write_gtags(json=modbus_write_gtags)
-        assert 404 == response.status_code
-        assert error.NEU_ERR_TAG_NOT_EXIST == response.json()['error']
-
-        modbus_write_gtags = {"node": param[0], "groups": [{"group": "group", "tags" : [{"tag": "hold_int16", "value": 1},
-            {"tag": "hold_uint16", "value": 2}, {"tag": "hold_int32", "value": 3}, {"tag": "hold_uint32", "value": -1}]}]}
-        response = api.write_gtags(json=modbus_write_gtags)
-        assert 404 == response.status_code
-        assert error.NEU_ERR_TAG_NOT_EXIST == response.json()['error']
-
     @description(given="created modbus node and tags", when="write/read tags", then="write/read success")
     def test_write_read_tags(self, param):
         api.write_tag_check(

tests/ft/neuron/api.py

@@ -318,6 +318,18 @@ def modbus_rtu_node_setting(node, port=502, connection_mode=0, transport_mode=0,
                                     "parity": parity, "baud": baud, "data": data})
 
 
+def modbus_tcp_node_setting_endian(node, port, connection_mode=0, transport_mode=0, interval=0, host='127.0.0.1', timeout=3000, max_retries=2, endianness=1):
+    return node_setting(node, json={"connection_mode": connection_mode, "transport_mode": transport_mode, "interval": interval,
+                                    "host": host, "port": port, "timeout": timeout, "max_retries": max_retries, "endianess": endianness})
+
+
+def modbus_rtu_node_setting_endian(node, port=502, connection_mode=0, transport_mode=0, interval=0, host='127.0.0.1', timeout=3000, max_retries=2, link=1, device="", stop=0, parity=0, baud=4, data=3, endianness=1):
+    return node_setting(node, json={"connection_mode": connection_mode, "transport_mode": transport_mode,
+                                    "interval": interval, "host": host, "port": port, "timeout": timeout,
+                                    "max_retries": max_retries, "link": link, "device": device, "stop": stop,
+                                    "parity": parity, "baud": baud, "data": data, "endianess": endianness})
+
+
 def mqtt_node_setting(node):
     return node_setting(node, json={"client-id": "neuron_aBcDeF", "qos": 0, "format": 0,
                                     "write-req-topic": f"/neuron/{node}/write/req", "write-resp-topic": f"/neuron/{node}/write/resp",

tests/ft/neuron/config.py

@@ -68,6 +68,11 @@ NEU_DATA_6 = 1
 NEU_DATA_7 = 2
 NEU_DATA_8 = 3
 
+ENDIAN_ABCD = 1,
+ENDIAN_BADC = 2,
+ENDIAN_DCBA = 3,
+ENDIAN_CDAB = 4,
+
 PLUGIN_MODBUS_TCP = "Modbus TCP"
 PLUGIN_MODBUS_RTU = "Modbus RTU"
 PLUGIN_MQTT = "MQTT"

tests/ft/neuron/error.py

@@ -53,7 +53,6 @@ NEU_ERR_TAG_ADDRESS_TOO_LONG = 2207
 NEU_ERR_TAG_DESCRIPTION_TOO_LONG = 2208
 NEU_ERR_TAG_PRECISION_INVALID = 2209
 NEU_ERR_TAG_EXIST = 2210
-NEU_ERR_TAG_VALUE_INVALID = 2211
 
 NEU_ERR_LIBRARY_NOT_FOUND = 2301
 NEU_ERR_LIBRARY_INFO_INVALID = 2302
@@ -71,6 +70,7 @@ NEU_ERR_LIBRARY_MODULE_KIND_NOT_SUPPORT = 2313
 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

tests/ft/plugin/test_plugin.py

@@ -200,7 +200,7 @@ class TestPlugin:
         response = api.updata_plugin(
             library_name=plugin_data['library'], so_file=plugin_data['so_file'], schema_file=plugin_data['schema_file'])
         assert 400 == response.status_code
-        assert NEU_ERR_LIBRARY_MODULE_NOT_EXISTS == response.json().get("error")
+        assert NEU_ERR_LIBRARY_UPDATE_FAIL == response.json().get("error")
 
     @description(given="incorrect configuration", when="update plugin", then="fail")
     def test_08_update_plugin_fail1(self):
@@ -244,7 +244,7 @@ class TestPlugin:
         assert 200 == response.status_code
         for plugin in response.json().get("plugins"):
             if plugin['name'] == 'c1':
-                assert plugin['version'] == '2.7.99'
+                assert plugin['version'] == '2.8.99'
 
     @description(given="incorrect configuration", when="update plugin", then="fail")
     def test_10_update_plugin_fail3(self):

tests/ft/tag/test_tag.py

@@ -9,7 +9,7 @@ class TestLog:
     @description(given="multiple non-existing groups and multiple correct tags", 
                  when="adding", 
                  then="successfully added")
-    def test_adding_correct_gtags_nonexisting_groups(self):
+    def test_adding_correct_gtags_nonexisting_groups(self, class_setup_and_teardown):
         response = api.add_node(node='modbus-tcp-tag-test', plugin=PLUGIN_MODBUS_TCP)
         assert 200 == response.status_code
         assert 0 == response.json()['error']
@@ -63,6 +63,9 @@ class TestLog:
         assert 200 == response.status_code
         assert NEU_ERR_SUCCESS == response.json()['error']
 
+        # check data persisted
+        class_setup_and_teardown.restart()
+
         response = api.get_group()
         assert 200       == response.status_code
         assert 'group_1' == response.json()['groups'][0]['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         = NEU_VERSION(2, 7, 99),
+    .version         = NEU_VERSION(2, 8, 99),
     .schema          = "c1",
     .module_name     = "c1",
     .module_descr    = "c1 v2",

version

@@ -1 +1 @@
-2.7.0-dev
+2.8.1