Merge pull request #2023 from fengzeroz/v2.8

version: 2.8.2

.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.2
   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

CMakeLists.txt

@@ -102,6 +102,11 @@ if (CLIB)
   add_definitions(-DNEU_CLIB=${CLIB})
 endif()
 
+if(CMAKE_BUILD_TYPE STREQUAL "Release")
+    message(STATUS "build release")
+    add_definitions(-DNEU_RELEASE)
+endif()
+
 
 add_library(neuron-base SHARED)
 target_sources(neuron-base PRIVATE ${NEURON_BASE_SOURCES} ${NEURON_SRC_PARSE}) 
@@ -109,7 +114,7 @@ target_include_directories(neuron-base
                            PRIVATE include/neuron src)
 target_link_libraries(neuron-base libssl.a libcrypto.a)
 target_link_libraries(neuron-base nng libzlog.so jansson jwt
-                      ${CMAKE_THREAD_LIBS_INIT})
+                      ${CMAKE_THREAD_LIBS_INIT} -lm)
 add_dependencies(neuron-base neuron-version)
 
 # dependency imposed by nng

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/connection/mqtt_client.h

@@ -53,6 +53,20 @@ typedef enum {
 
 typedef struct neu_mqtt_client_s neu_mqtt_client_t;
 
+/**
+ * Check that `topic_filter` is a valid MQTT topic filter.
+ */
+bool neu_mqtt_topic_filter_is_valid(const char *topic_filter);
+
+/**
+ * Check the given `topic_filter` matches the given `topic_name`.
+ *
+ * @Precondition  `topic_filter` must be a valid MQTT topic filter.
+ * @Precondition  `topic_name` must be a valid MQTT topic name.
+ */
+bool neu_mqtt_topic_filter_is_match(const char *topic_filter,
+                                    const char *topic_name);
+
 typedef void (*neu_mqtt_client_connection_cb_t)(void *data);
 typedef void (*neu_mqtt_client_publish_cb_t)(int errcode, neu_mqtt_qos_e qos,
                                              char *topic, uint8_t *payload,

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 2
 
 #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/metrics.h

@@ -41,16 +41,22 @@ typedef enum {
     NEU_METRIC_TYPE_GAUAGE,
     NEU_METRIC_TYPE_COUNTER_SET,
     NEU_METRIC_TYPE_ROLLING_COUNTER,
+
+    NEU_METRIC_TYPE_FLAG_NO_RESET = 0x80,
 } neu_metric_type_e;
 
+#define NEU_METRIC_TYPE_MASK 0x0F
+
 // node running state
 #define NEU_METRIC_RUNNING_STATE "running_state"
-#define NEU_METRIC_RUNNING_STATE_TYPE NEU_METRIC_TYPE_GAUAGE
+#define NEU_METRIC_RUNNING_STATE_TYPE \
+    (NEU_METRIC_TYPE_GAUAGE | NEU_METRIC_TYPE_FLAG_NO_RESET)
 #define NEU_METRIC_RUNNING_STATE_HELP "Node running state"
 
 // node link state
 #define NEU_METRIC_LINK_STATE "link_state"
-#define NEU_METRIC_LINK_STATE_TYPE NEU_METRIC_TYPE_GAUAGE
+#define NEU_METRIC_LINK_STATE_TYPE \
+    (NEU_METRIC_TYPE_GAUAGE | NEU_METRIC_TYPE_FLAG_NO_RESET)
 #define NEU_METRIC_LINK_STATE_HELP "Node link state"
 
 // last round trip time in millisends
@@ -86,13 +92,15 @@ typedef enum {
 // maintained by neuron core
 // number of tags in group
 #define NEU_METRIC_TAGS_TOTAL "tags_total"
-#define NEU_METRIC_TAGS_TOTAL_TYPE NEU_METRIC_TYPE_GAUAGE
+#define NEU_METRIC_TAGS_TOTAL_TYPE \
+    (NEU_METRIC_TYPE_GAUAGE | NEU_METRIC_TYPE_FLAG_NO_RESET)
 #define NEU_METRIC_TAGS_TOTAL_HELP "Total number of tags in the node"
 
 // maintained by neuron core
 // number of tags in group
 #define NEU_METRIC_GROUP_TAGS_TOTAL "group_tags_total"
-#define NEU_METRIC_GROUP_TAGS_TOTAL_TYPE NEU_METRIC_TYPE_GAUAGE
+#define NEU_METRIC_GROUP_TAGS_TOTAL_TYPE \
+    (NEU_METRIC_TYPE_GAUAGE | NEU_METRIC_TYPE_FLAG_NO_RESET)
 #define NEU_METRIC_GROUP_TAGS_TOTAL_HELP "Total number of tags in the group"
 
 // number of messages sent in last group timer
@@ -301,9 +309,24 @@ void neu_metrics_unregister_entry(const char *name);
 typedef void (*neu_metrics_cb_t)(const neu_metrics_t *metrics, void *data);
 void neu_metrics_visist(neu_metrics_cb_t cb, void *data);
 
+static inline bool neu_metric_type_is_counter(neu_metric_type_e type)
+{
+    return NEU_METRIC_TYPE_COUNTER == (type & NEU_METRIC_TYPE_MASK);
+}
+
+static inline bool neu_metric_type_is_rolling_counter(neu_metric_type_e type)
+{
+    return NEU_METRIC_TYPE_ROLLING_COUNTER == (type & NEU_METRIC_TYPE_MASK);
+}
+
+static inline bool neu_metric_type_no_reset(neu_metric_type_e type)
+{
+    return NEU_METRIC_TYPE_FLAG_NO_RESET & type;
+}
+
 static inline const char *neu_metric_type_str(neu_metric_type_e type)
 {
-    if (NEU_METRIC_TYPE_COUNTER == type) {
+    if (neu_metric_type_is_counter(type)) {
         return "counter";
     } else {
         return "gauge";
@@ -316,7 +339,7 @@ int neu_metric_entries_add(neu_metric_entry_t **entries, const char *name,
 
 static inline void neu_metric_entry_free(neu_metric_entry_t *entry)
 {
-    if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) {
+    if (neu_metric_type_is_rolling_counter(entry->type)) {
         neu_rolling_counter_free(entry->rcnt);
     }
     free(entry);
@@ -445,9 +468,9 @@ static inline int neu_node_metrics_update(neu_node_metrics_t *node_metrics,
         return -1;
     }
 
-    if (NEU_METRIC_TYPE_COUNTER == entry->type) {
+    if (neu_metric_type_is_counter(entry->type)) {
         entry->value += n;
-    } else if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) {
+    } else if (neu_metric_type_is_rolling_counter(entry->type)) {
         entry->value =
             neu_rolling_counter_inc(entry->rcnt, global_timestamp, n);
     } else {
@@ -465,9 +488,11 @@ static inline void neu_node_metrics_reset(neu_node_metrics_t *node_metrics)
     pthread_mutex_lock(&node_metrics->lock);
     HASH_LOOP(hh, node_metrics->entries, entry)
     {
-        entry->value = 0;
-        if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) {
-            neu_rolling_counter_reset(entry->rcnt);
+        if (!neu_metric_type_no_reset(entry->type)) {
+            entry->value = 0;
+            if (neu_metric_type_is_rolling_counter(entry->type)) {
+                neu_rolling_counter_reset(entry->rcnt);
+            }
         }
     }
 
@@ -476,9 +501,11 @@ static inline void neu_node_metrics_reset(neu_node_metrics_t *node_metrics)
     {
         HASH_LOOP(hh, g->entries, entry)
         {
-            entry->value = 0;
-            if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) {
-                neu_rolling_counter_reset(entry->rcnt);
+            if (!neu_metric_type_no_reset(entry->type)) {
+                entry->value = 0;
+                if (neu_metric_type_is_rolling_counter(entry->type)) {
+                    neu_rolling_counter_reset(entry->rcnt);
+                }
             }
         }
     }

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;
@@ -904,10 +906,11 @@ inline static void neu_msg_exchange(neu_reqresp_head_t *header)
 typedef struct {
     char driver[NEU_NODE_NAME_LEN];
     char path[128];
-    int  prg_int_param1;
-    int  prg_int_param2;
+    bool del_flag;
     char prg_str_param1[128];
     char prg_str_param2[128];
+    char prg_str_param3[128];
+    char prg_str_param4[128];
 } neu_req_prgfile_upload_t;
 
 typedef struct {

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 "";
@@ -130,7 +136,7 @@ typedef union {
 
 static inline char *neu_value_str(neu_type_e type, neu_value_u value)
 {
-    static __thread char str[32] = { 0 };
+    static __thread char str[128] = { 0 };
 
     switch (type) {
     case NEU_TYPE_INT8:
@@ -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;

plugins/restful/metric_handle.c

@@ -181,7 +181,7 @@ static inline void gen_single_node_metrics(neu_node_metrics_t *node_metrics,
     pthread_mutex_lock(&node_metrics->lock);
     HASH_LOOP(hh, node_metrics->entries, e)
     {
-        if (NEU_METRIC_TYPE_ROLLING_COUNTER == e->type) {
+        if (neu_metric_type_is_rolling_counter(e->type)) {
             // force clean stale value
             e->value = neu_rolling_counter_inc(e->rcnt, global_timestamp, 0);
         }
@@ -196,7 +196,7 @@ static inline void gen_single_node_metrics(neu_node_metrics_t *node_metrics,
     {
         HASH_LOOP(hh, g->entries, e)
         {
-            if (NEU_METRIC_TYPE_ROLLING_COUNTER == e->type) {
+            if (neu_metric_type_is_rolling_counter(e->type)) {
                 // force clean stale value
                 e->value =
                     neu_rolling_counter_inc(e->rcnt, global_timestamp, 0);
@@ -278,7 +278,7 @@ static void gen_all_node_metrics(neu_metrics_t *metrics, int type_filter,
             pthread_mutex_lock(&n->lock);
             HASH_FIND_STR(n->entries, r->name, e);
             if (e) {
-                if (NEU_METRIC_TYPE_ROLLING_COUNTER == e->type) {
+                if (neu_metric_type_is_rolling_counter(e->type)) {
                     // force clean stale value
                     e->value =
                         neu_rolling_counter_inc(e->rcnt, global_timestamp, 0);
@@ -294,7 +294,7 @@ static void gen_all_node_metrics(neu_metrics_t *metrics, int type_filter,
             {
                 HASH_FIND_STR(g->entries, r->name, e);
                 if (e) {
-                    if (NEU_METRIC_TYPE_ROLLING_COUNTER == e->type) {
+                    if (neu_metric_type_is_rolling_counter(e->type)) {
                         // force clean stale value
                         e->value = neu_rolling_counter_inc(e->rcnt,
                                                            global_timestamp, 0);

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;
@@ -406,13 +426,6 @@ static int adapter_update_metric(neu_adapter_t *adapter,
     return neu_node_metrics_update(adapter->metrics, group, metric_name, n);
 }
 
-static inline void adapter_reset_metrics(neu_adapter_t *adapter)
-{
-    if (NULL != adapter->metrics) {
-        neu_node_metrics_reset(adapter->metrics);
-    }
-}
-
 static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
                            void *data)
 {
@@ -911,7 +924,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 +1064,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 +1194,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 +1251,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 +1282,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;
@@ -1312,6 +1337,11 @@ void neu_adapter_destroy(neu_adapter_t *adapter)
     }
 
     neu_event_close(adapter->events);
+#ifdef NEU_RELEASE
+    if (adapter->handle != NULL) {
+        dlclose(adapter->handle);
+    }
+#endif
     free(adapter);
 }
 
@@ -1357,6 +1387,10 @@ int neu_adapter_start(neu_adapter_t *adapter)
     if (error == NEU_ERR_SUCCESS) {
         adapter->state = NEU_NODE_RUNNING_STATE_RUNNING;
         adapter_storage_state(adapter->name, adapter->state);
+        if (NEU_NA_TYPE_DRIVER == neu_adapter_get_type(adapter)) {
+            neu_adapter_driver_start_group_timer(
+                (neu_adapter_driver_t *) adapter);
+        }
     }
 
     return error;
@@ -1395,7 +1429,11 @@ int neu_adapter_stop(neu_adapter_t *adapter)
     if (error == NEU_ERR_SUCCESS) {
         adapter->state = NEU_NODE_RUNNING_STATE_STOPPED;
         adapter_storage_state(adapter->name, adapter->state);
-        adapter_reset_metrics(adapter);
+        if (NEU_NA_TYPE_DRIVER == neu_adapter_get_type(adapter)) {
+            neu_adapter_driver_stop_group_timer(
+                (neu_adapter_driver_t *) adapter);
+        }
+        neu_adapter_reset_metrics(adapter);
     }
 
     return error;

src/adapter/adapter_internal.h

@@ -96,6 +96,13 @@ int neu_adapter_set_setting(neu_adapter_t *adapter, const char *config);
 int neu_adapter_get_setting(neu_adapter_t *adapter, char **config);
 neu_node_state_t neu_adapter_get_state(neu_adapter_t *adapter);
 
+static inline void neu_adapter_reset_metrics(neu_adapter_t *adapter)
+{
+    if (NULL != adapter->metrics) {
+        neu_node_metrics_reset(adapter->metrics);
+    }
+}
+
 int  neu_adapter_register_group_metric(neu_adapter_t *adapter,
                                        const char *group_name, const char *name,
                                        const char *help, neu_metric_type_e type,

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

@@ -102,8 +102,49 @@ static void update_with_meta(neu_adapter_t *adapter, const char *group,
                              const char *tag, neu_dvalue_t value,
                              neu_tag_meta_t *metas, int n_meta);
 static void write_response(neu_adapter_t *adapter, void *r, neu_error error);
-static group_t *find_group(neu_adapter_driver_t *driver, const char *name);
-static void     store_write_tag(group_t *group, to_be_write_tag_t *tag);
+static group_t *   find_group(neu_adapter_driver_t *driver, const char *name);
+static void        store_write_tag(group_t *group, to_be_write_tag_t *tag);
+static inline void start_group_timer(neu_adapter_driver_t *driver,
+                                     group_t *             grp);
+static inline void stop_group_timer(neu_adapter_driver_t *driver, group_t *grp);
+
+static void format_tag_value(neu_resp_tag_value_meta_t *value)
+{
+    double scale = pow(10, 5);
+
+    int negative = 1;
+
+    if (value->value.value.d64 < 0) {
+        value->value.value.d64 *= -1;
+        negative = -1;
+    }
+
+    int64_t integer_part = (int64_t) value->value.value.d64;
+    double  decimal_part = value->value.value.d64 - integer_part;
+    decimal_part *= scale;
+    decimal_part = round(decimal_part);
+    char str[6]  = { 0 };
+    snprintf(str, sizeof(str), "%05" PRId64 "", (int64_t) decimal_part);
+    int i = 0, flag = 0;
+    for (; i < 4; i++) {
+        if (str[i] == '0' && str[i + 1] == '0') {
+            flag = 1;
+            break;
+        } else if (str[i] == '9' && str[i + 1] == '9') {
+            flag = 2;
+            break;
+        }
+    }
+    if (flag != 0 && i != 0) {
+        decimal_part = round(decimal_part / pow(10, 5 - i));
+        value->value.value.d64 =
+            (double) integer_part + decimal_part / pow(10, i);
+    } else {
+        value->value.value.d64 = (double) integer_part + decimal_part / scale;
+    }
+
+    value->value.value.d64 *= negative;
+}
 
 static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
 {
@@ -168,7 +209,7 @@ static void update_with_meta(neu_adapter_t *adapter, const char *group,
         update_metric(&driver->adapter, NEU_METRIC_TAG_READ_ERRORS_TOTAL,
                       NEU_TYPE_ERROR == value.type, NULL);
     }
-    nlog_info(
+    nlog_debug(
         "update driver: %s, group: %s, tag: %s, type: %s, timestamp: %" PRId64
         " n_meta: %d",
         driver->adapter.name, group, tag, neu_type_string(value.type),
@@ -201,19 +242,19 @@ static void update_im(neu_adapter_t *adapter, const char *group,
     if (first == NULL ||
         !neu_tag_attribute_test(first, NEU_ATTRIBUTE_SUBSCRIBE)) {
         utarray_free(tags);
-        nlog_info("update immediately, driver: %s, "
-                  "group: %s, tag: %s, type: %s, "
-                  "timestamp: %" PRId64,
-                  driver->adapter.name, group, tag, neu_type_string(value.type),
-                  global_timestamp);
+        nlog_debug("update immediately, driver: %s, "
+                   "group: %s, tag: %s, type: %s, "
+                   "timestamp: %" PRId64,
+                   driver->adapter.name, group, tag,
+                   neu_type_string(value.type), global_timestamp);
         return;
     }
 
-    nlog_info("update and report immediately, driver: %s, "
-              "group: %s, tag: %s, type: %s, "
-              "timestamp: %" PRId64,
-              driver->adapter.name, group, tag, neu_type_string(value.type),
-              global_timestamp);
+    nlog_debug("update and report immediately, driver: %s, "
+               "group: %s, tag: %s, type: %s, "
+               "timestamp: %" PRId64,
+               driver->adapter.name, group, tag, neu_type_string(value.type),
+               global_timestamp);
 
     neu_reqresp_head_t header = {
         .type = NEU_REQRESP_TRANS_DATA,
@@ -324,9 +365,7 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver)
         HASH_DEL(driver->groups, el);
 
         neu_adapter_driver_try_del_tag(driver, neu_group_tag_size(el->group));
-        neu_adapter_del_timer((neu_adapter_t *) driver, el->report);
-        neu_event_del_timer(driver->driver_events, el->read);
-        neu_event_del_timer(driver->driver_events, el->write);
+        stop_group_timer(driver, el);
         if (el->grp.group_free != NULL) {
             el->grp.group_free(&el->grp);
         }
@@ -349,34 +388,67 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver)
     return 0;
 }
 
+static inline void start_group_timer(neu_adapter_driver_t *driver, group_t *grp)
+{
+    uint32_t interval = neu_group_get_interval(grp->group);
+
+    neu_event_timer_param_t param = {
+        .second      = interval / 1000,
+        .millisecond = interval % 1000,
+        .usr_data    = (void *) grp,
+        .type        = NEU_EVENT_TIMER_NOBLOCK,
+    };
+
+    param.type = driver->adapter.module->timer_type;
+    param.cb   = read_callback;
+    grp->read  = neu_event_add_timer(driver->driver_events, param);
+
+    struct timespec t1 = {
+        .tv_sec  = 0,
+        .tv_nsec = 1000 * 1000 * 20,
+    };
+    struct timespec t2 = { 0 };
+    nanosleep(&t1, &t2);
+
+    param.type  = NEU_EVENT_TIMER_NOBLOCK;
+    param.cb    = report_callback;
+    grp->report = neu_adapter_add_timer((neu_adapter_t *) driver, param);
+
+    param.second      = 0;
+    param.millisecond = 3;
+    param.cb          = write_callback;
+    grp->write        = neu_event_add_timer(driver->driver_events, param);
+}
+
 void neu_adapter_driver_start_group_timer(neu_adapter_driver_t *driver)
 {
     group_t *el = NULL, *tmp = NULL;
 
     HASH_ITER(hh, driver->groups, el, tmp)
     {
-        uint32_t                interval = neu_group_get_interval(el->group);
-        neu_event_timer_param_t param    = {
-            .second      = interval / 1000,
-            .millisecond = interval % 1000,
-            .usr_data    = el,
-            .type        = NEU_EVENT_TIMER_NOBLOCK,
-        };
+        start_group_timer(driver, el);
+        neu_adapter_update_group_metric(
+            &driver->adapter, neu_group_get_name(el->group),
+            NEU_METRIC_GROUP_TAGS_TOTAL, neu_group_tag_size(el->group));
+    }
 
-        param.type = driver->adapter.module->timer_type;
-        param.cb   = read_callback;
-        el->read   = neu_event_add_timer(driver->driver_events, param);
+    driver->adapter.cb_funs.update_metric(
+        &driver->adapter, NEU_METRIC_TAGS_TOTAL, driver->tag_cnt, NULL);
+}
 
-        struct timespec t1 = {
-            .tv_sec  = 0,
-            .tv_nsec = 1000 * 1000 * 20,
-        };
-        struct timespec t2 = { 0 };
-        nanosleep(&t1, &t2);
-
-        param.type = NEU_EVENT_TIMER_NOBLOCK;
-        param.cb   = report_callback;
-        el->report = neu_adapter_add_timer((neu_adapter_t *) driver, param);
+static inline void stop_group_timer(neu_adapter_driver_t *driver, group_t *grp)
+{
+    if (grp->report) {
+        neu_adapter_del_timer((neu_adapter_t *) driver, grp->report);
+        grp->report = NULL;
+    }
+    if (grp->read) {
+        neu_event_del_timer(driver->driver_events, grp->read);
+        grp->read = NULL;
+    }
+    if (grp->write) {
+        neu_event_del_timer(driver->driver_events, grp->write);
+        grp->write = NULL;
     }
 }
 
@@ -384,13 +456,7 @@ void neu_adapter_driver_stop_group_timer(neu_adapter_driver_t *driver)
 {
     group_t *el = NULL, *tmp = NULL;
 
-    HASH_ITER(hh, driver->groups, el, tmp)
-    {
-        neu_adapter_del_timer((neu_adapter_t *) driver, el->report);
-        el->report = NULL;
-        neu_event_del_timer(driver->driver_events, el->read);
-        el->read = NULL;
-    }
+    HASH_ITER(hh, driver->groups, el, tmp) { stop_group_timer(driver, el); }
 }
 
 void neu_adapter_driver_read_group(neu_adapter_driver_t *driver,
@@ -472,6 +538,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 +705,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 +736,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 +823,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 +905,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;
@@ -917,13 +956,6 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
     if (find == NULL) {
         find = calloc(1, sizeof(group_t));
 
-        neu_event_timer_param_t param = {
-            .second      = interval / 1000,
-            .millisecond = interval % 1000,
-            .usr_data    = (void *) find,
-            .type        = NEU_EVENT_TIMER_NOBLOCK,
-        };
-
         pthread_mutex_init(&find->wt_mtx, NULL);
         pthread_mutex_init(&find->apps_mtx, NULL);
 
@@ -937,25 +969,9 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
         neu_group_split_static_tags(find->group, &find->static_tags,
                                     &find->grp.tags);
 
-        param.type = driver->adapter.module->timer_type;
-        param.cb   = read_callback;
-        find->read = neu_event_add_timer(driver->driver_events, param);
-
-        struct timespec t1 = {
-            .tv_sec  = 0,
-            .tv_nsec = 1000 * 1000 * 20,
-        };
-        struct timespec t2 = { 0 };
-        nanosleep(&t1, &t2);
-
-        param.type   = NEU_EVENT_TIMER_NOBLOCK;
-        param.cb     = report_callback;
-        find->report = neu_adapter_add_timer((neu_adapter_t *) driver, param);
-
-        param.second      = 0;
-        param.millisecond = 3;
-        param.cb          = write_callback;
-        find->write       = neu_event_add_timer(driver->driver_events, param);
+        if (NEU_NODE_RUNNING_STATE_RUNNING == driver->adapter.state) {
+            start_group_timer(driver, find);
+        }
 
         REGISTER_GROUP_METRIC(&driver->adapter, find->name,
                               NEU_METRIC_GROUP_TAGS_TOTAL,
@@ -1002,8 +1018,9 @@ int neu_adapter_driver_update_group(neu_adapter_driver_t *driver,
     }
 
     // stop the timer first to avoid race condition
-    neu_adapter_del_timer((neu_adapter_t *) driver, find->report);
-    neu_event_del_timer(driver->driver_events, find->read);
+    if (NEU_NODE_RUNNING_STATE_RUNNING == driver->adapter.state) {
+        stop_group_timer(driver, find);
+    }
 
     // a diminutive value should keep the interval untouched
     if (interval < NEU_GROUP_INTERVAL_LIMIT) {
@@ -1032,31 +1049,12 @@ int neu_adapter_driver_update_group(neu_adapter_driver_t *driver,
         }
     }
 
-    neu_event_timer_param_t param = {
-        .second      = interval / 1000,
-        .millisecond = interval % 1000,
-        .usr_data    = (void *) find,
-        .type        = NEU_EVENT_TIMER_NOBLOCK,
-    };
-
     neu_group_set_interval(find->group, interval);
 
     // restore the timers
-
-    param.type = driver->adapter.module->timer_type;
-    param.cb   = read_callback;
-    find->read = neu_event_add_timer(driver->driver_events, param);
-
-    struct timespec t1 = {
-        .tv_sec  = 0,
-        .tv_nsec = 1000 * 1000 * 20,
-    };
-    struct timespec t2 = { 0 };
-    nanosleep(&t1, &t2);
-
-    param.type   = NEU_EVENT_TIMER_NOBLOCK;
-    param.cb     = report_callback;
-    find->report = neu_adapter_add_timer((neu_adapter_t *) driver, param);
+    if (NEU_NODE_RUNNING_STATE_RUNNING == driver->adapter.state) {
+        start_group_timer(driver, find);
+    }
 
     return ret;
 }
@@ -1072,9 +1070,10 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
 
         neu_adapter_driver_try_del_tag(driver, neu_group_tag_size(find->group));
 
-        neu_adapter_del_timer((neu_adapter_t *) driver, find->report);
-        neu_event_del_timer(driver->driver_events, find->read);
-        neu_event_del_timer(driver->driver_events, find->write);
+        if (NEU_NODE_RUNNING_STATE_RUNNING == driver->adapter.state) {
+            stop_group_timer(driver, find);
+        }
+
         if (find->grp.group_free != NULL) {
             find->grp.group_free(&find->grp);
         }
@@ -1716,7 +1715,7 @@ static void read_report_group(int64_t timestamp, int64_t timeout,
             if (neu_driver_cache_meta_get_changed(cache, group, tag->name,
                                                   &value, tag_value.metas,
                                                   NEU_TAG_META_SIZE) != 0) {
-                nlog_info("tag: %s not changed", tag->name);
+                nlog_debug("tag: %s not changed", tag->name);
                 continue;
             }
         } else {
@@ -1861,6 +1860,9 @@ static void read_report_group(int64_t timestamp, int64_t timeout,
                     break;
                 }
             }
+            if (tag->precision == 0 && tag->type == NEU_TYPE_DOUBLE) {
+                format_tag_value(&tag_value);
+            }
         }
 
         utarray_push_back(tag_values, &tag_value);
@@ -2016,6 +2018,10 @@ static void read_group(int64_t timestamp, int64_t timeout,
                     break;
                 }
             }
+
+            if (tag->precision == 0 && tag->type == NEU_TYPE_DOUBLE) {
+                format_tag_value(&tag_value);
+            }
         }
         utarray_push_back(tag_values, &tag_value);
     }

src/base/metrics.c

@@ -212,16 +212,17 @@ static unsigned cpu_usage()
     return (double) work / total * 100.0 * sysconf(_SC_NPROCESSORS_CONF);
 }
 
-static bool has_core_dumps()
+static bool has_core_dump_in_dir(const char *dir, const char *prefix)
 {
-    DIR *dp = opendir("core");
-    if (NULL == dp) {
+    DIR *dp = opendir(dir);
+    if (dp == NULL) {
         return false;
     }
 
-    bool found = false;
-    for (struct dirent *de = NULL; NULL != (de = readdir(dp));) {
-        if (0 == strncmp("core", de->d_name, 4)) {
+    struct dirent *de;
+    bool           found = false;
+    while ((de = readdir(dp)) != NULL) {
+        if (strncmp(prefix, de->d_name, strlen(prefix)) == 0) {
             found = true;
             break;
         }
@@ -231,6 +232,46 @@ static bool has_core_dumps()
     return found;
 }
 
+static char *get_core_dir(const char *core_pattern)
+{
+    char *last_slash = strrchr(core_pattern, '/');
+    if (last_slash != NULL) {
+        static char core_dir[256];
+        ptrdiff_t   path_length = last_slash - core_pattern + 1;
+        strncpy(core_dir, core_pattern, path_length);
+        core_dir[path_length] = '\0';
+        return core_dir;
+    }
+    return NULL;
+}
+
+static bool has_core_dumps()
+{
+    if (has_core_dump_in_dir("core", "core-neuron")) {
+        return true;
+    }
+
+    FILE *fp = fopen("/proc/sys/kernel/core_pattern", "r");
+    if (fp == NULL) {
+        return false;
+    }
+
+    char core_pattern[256];
+    if (fgets(core_pattern, sizeof(core_pattern), fp) == NULL) {
+        fclose(fp);
+        return false;
+    }
+    fclose(fp);
+
+    char *core_dir =
+        (core_pattern[0] == '|') ? "/var/crash/" : get_core_dir(core_pattern);
+    if (core_dir == NULL) {
+        return false;
+    }
+
+    return has_core_dump_in_dir(core_dir, "core-neuron");
+}
+
 static inline void metrics_unregister_entry(const char *name)
 {
     neu_metric_entry_t *e = NULL;

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/connection/mqtt_client.c

@@ -33,6 +33,7 @@
 #include <nng/supplemental/util/platform.h>
 
 #include "connection/mqtt_client.h"
+#include "errcodes.h"
 #include "event/event.h"
 #include "utils/asprintf.h"
 #include "utils/time.h"
@@ -179,6 +180,93 @@ static inline uint8_t neu_mqtt_version_to_nng_mqtt_version(neu_mqtt_version_e v)
     }
 }
 
+bool neu_mqtt_topic_filter_is_valid(const char *topic_filter)
+{
+    if (NULL == topic_filter || '\0' == *topic_filter ||
+        strlen(topic_filter) > 65535) {
+        return false;
+    }
+
+    char c;
+    while ((c = *topic_filter++)) {
+        switch (c) {
+        case '#':
+            // '#' must be last character
+            return '\0' == *topic_filter;
+        case '+':
+            // '+' must occupy an entire level
+            if (*topic_filter && '/' != *topic_filter) {
+                return false;
+            }
+            break;
+        default:
+            for (; '/' != c; c = *topic_filter++) {
+                if ('#' == c || '+' == c) {
+                    return false;
+                } else if ('\0' == c) {
+                    return true;
+                }
+            }
+        }
+    }
+
+    return true;
+}
+
+static bool topic_filter_is_match(const char *topic_filter,
+                                  const char *topic_name,
+                                  uint32_t    topic_name_len)
+{
+    // The Server MUST NOT match Topic Filters starting with a wildcard
+    // character (# or +) with Topic Names beginning with a $ character
+    // [MQTT-4.7.2-1].
+    if ('$' == topic_name[0]) {
+        if ('$' != topic_filter[0]) {
+            return false;
+        }
+    } else if (0 == strcmp("#", topic_filter)) {
+        return true;
+    }
+
+    uint32_t i = 0;
+    while (true) {
+        switch (*topic_filter) {
+        case '#':
+            return true;
+        case '+':
+            while (i < topic_name_len && '/' != topic_name[i]) {
+                ++i;
+            };
+            ++topic_filter;
+            break;
+        case '/':
+            if (i < topic_name_len && '/' == topic_name[i]) {
+                ++i;
+            } else if ('#' != topic_filter[1]) { // '#' include parent level
+                return false;
+            }
+            ++topic_filter;
+            break;
+        case '\0':
+            return i == topic_name_len;
+        default:
+            for (; *topic_filter && '/' != *topic_filter; ++topic_filter, ++i) {
+                if (!(i < topic_name_len && *topic_filter == topic_name[i])) {
+                    return false;
+                }
+            }
+        }
+    }
+
+    return true;
+}
+
+bool neu_mqtt_topic_filter_is_match(const char *topic_filter,
+                                    const char *topic_name)
+{
+    return topic_filter_is_match(topic_filter, topic_name, strlen(topic_name));
+}
+
 static inline task_t *task_new(neu_mqtt_client_t *client)
 {
     task_t *task = calloc(1, sizeof(*task));
@@ -405,6 +493,26 @@ static inline void subscriptions_free(subscription_t *subscriptions)
     }
 }
 
+static inline subscription_t *
+subscription_find_match(subscription_t *subscriptions, const char *topic_name,
+                        uint32_t topic_name_len)
+{
+    subscription_t *sub = NULL, *tmp = NULL;
+    HASH_FIND(hh, subscriptions, topic_name, topic_name_len, sub);
+    if (NULL == sub) {
+        // subscription wildcard matching
+        // TODO: linear scanning is enough for current usages,
+        //       but may pose performance issues in the future
+        HASH_ITER(hh, subscriptions, sub, tmp)
+        {
+            if (topic_filter_is_match(sub->topic, topic_name, topic_name_len)) {
+                return sub;
+            }
+        }
+    }
+    return sub;
+}
+
 static int resub_cb(void *data)
 {
     neu_mqtt_client_t *client = data;
@@ -517,10 +625,9 @@ static void recv_cb(void *arg)
     log(debug, "recv [%.*s, QoS%d] %" PRIu32 " bytes", (unsigned) topic_len,
         topic, (int) qos, payload_len);
 
-    // find matching subscription
-    // NOTE: does not support wildcards
     nng_mtx_lock(client->mtx);
-    HASH_FIND(hh, client->subscriptions, topic, topic_len, subscription);
+    subscription =
+        subscription_find_match(client->subscriptions, topic, topic_len);
     if (NULL != subscription) {
         task_t *task = client_alloc_task(client);
         if (NULL != task) {
@@ -1375,6 +1482,10 @@ int neu_mqtt_client_subscribe(neu_mqtt_client_t *client, neu_mqtt_qos_e qos,
     int             rv           = 0;
     subscription_t *subscription = NULL;
 
+    if (!neu_mqtt_topic_filter_is_valid(topic)) {
+        return NEU_ERR_MQTT_SUBSCRIBE_FAILURE;
+    }
+
     nng_mtx_lock(client->mtx);
     if (NULL == client->recv_aio) {
         if ((rv = nng_aio_alloc(&client->recv_aio, recv_cb, client)) != 0) {

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

@@ -125,7 +125,6 @@ static int neuron_run(const neu_cli_args_t *args)
     int           rv = 0;
 
     signal(SIGINT, sig_handler);
-    signal(SIGABRT, sig_handler);
     signal(SIGTERM, sig_handler);
     signal(SIGKILL, sig_handler);
 

src/parser/neu_json_rw.c

@@ -43,6 +43,7 @@ int neu_json_encode_read_resp(void *json_object, void *param)
     neu_json_read_resp_tag_t *p_tag     = resp->tags;
     for (int i = 0; i < resp->n_tag; i++) {
         neu_json_elem_t tag_elems[2 + NEU_TAG_META_SIZE] = { 0 };
+        int             if_precision                     = 0;
 
         tag_elems[0].name      = "name";
         tag_elems[0].t         = NEU_JSON_STR;
@@ -57,16 +58,24 @@ int neu_json_encode_read_resp(void *json_object, void *param)
             tag_elems[1].t         = p_tag->t;
             tag_elems[1].v         = p_tag->value;
             tag_elems[1].precision = p_tag->precision;
+
+            if (p_tag->t == NEU_JSON_FLOAT || p_tag->t == NEU_JSON_DOUBLE) {
+                if_precision      = 1;
+                tag_elems[2].name = "precision";
+                tag_elems[2].t    = NEU_JSON_INT;
+                tag_elems[2].v.val_int =
+                    p_tag->precision > 0 ? p_tag->precision : 1;
+            }
         }
 
         for (int k = 0; k < p_tag->n_meta; k++) {
-            tag_elems[2 + k].name = p_tag->metas[k].name;
-            tag_elems[2 + k].t    = p_tag->metas[k].t;
-            tag_elems[2 + k].v    = p_tag->metas[k].value;
+            tag_elems[if_precision + 2 + k].name = p_tag->metas[k].name;
+            tag_elems[if_precision + 2 + k].t    = p_tag->metas[k].t;
+            tag_elems[if_precision + 2 + k].v    = p_tag->metas[k].value;
         }
 
-        tag_array =
-            neu_json_encode_array(tag_array, tag_elems, 2 + p_tag->n_meta);
+        tag_array = neu_json_encode_array(tag_array, tag_elems,
+                                          2 + if_precision + p_tag->n_meta);
         p_tag++;
     }
 
@@ -407,14 +416,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 +450,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 +463,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 +595,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;
         }
@@ -292,7 +294,8 @@ int neu_json_decode_add_tags_req(char *buf, neu_json_add_tags_req_t **result)
 
     json_obj = neu_json_decode_new(buf);
     if (NULL == json_obj) {
-        goto decode_fail;
+        free(req);
+        return -1;
     }
 
     neu_json_elem_t req_elems[] = {
@@ -563,7 +566,8 @@ int neu_json_decode_add_gtags_req(char *buf, neu_json_add_gtags_req_t **result)
 
     json_obj = neu_json_decode_new(buf);
     if (NULL == json_obj) {
-        goto decode_fail;
+        free(req);
+        return -1;
     }
 
     neu_json_elem_t req_elems[] = {
@@ -657,6 +661,10 @@ int neu_json_decode_del_tags_req(char *buf, neu_json_del_tags_req_t **result)
     }
 
     json_obj = neu_json_decode_new(buf);
+    if (NULL == json_obj) {
+        free(req);
+        return -1;
+    }
 
     neu_json_elem_t req_elems[] = { {
                                         .name = "node",
@@ -773,7 +781,8 @@ int neu_json_decode_update_tags_req(char *                       buf,
 
     json_obj = neu_json_decode_new(buf);
     if (NULL == json_obj) {
-        goto decode_fail;
+        free(req);
+        return -1;
     }
 
     neu_json_elem_t req_elems[] = {

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:

src/utils/json.c

@@ -17,6 +17,7 @@
  * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  **/
 
+#include <math.h>
 #include <stdlib.h>
 #include <string.h>
 
@@ -25,6 +26,44 @@
 #include "utils/log.h"
 #include "json/json.h"
 
+static double format_tag_value(float ele_value)
+{
+    double scale    = pow(10, 5);
+    double value    = ele_value;
+    int    negative = 1;
+
+    if (value < 0) {
+        value *= -1;
+        negative = -1;
+    }
+
+    int64_t integer_part = (int64_t)(value);
+    double  decimal_part = value - integer_part;
+    decimal_part *= scale;
+    decimal_part = round(decimal_part);
+
+    char str[6] = { 0 };
+    snprintf(str, sizeof(str), "%05" PRId64 "", (int64_t) decimal_part);
+    int i = 0, flag = 0;
+    for (; i < 4; i++) {
+        if (str[i] == '0' && str[i + 1] == '0') {
+            flag = 1;
+            break;
+        } else if (str[i] == '9' && str[i + 1] == '9') {
+            flag = 2;
+            break;
+        }
+    }
+    if (flag != 0 && i != 0) {
+        decimal_part = round(decimal_part / pow(10, 5 - i));
+        value        = (double) integer_part + decimal_part / pow(10, i);
+    } else {
+        value = (double) integer_part + decimal_part / scale;
+    }
+
+    return value * negative;
+}
+
 static json_t *encode_object_value(neu_json_elem_t *ele)
 {
     json_t *ob = NULL;
@@ -42,9 +81,14 @@ static json_t *encode_object_value(neu_json_elem_t *ele)
     case NEU_JSON_DOUBLE:
         ob = json_realp(ele->v.val_double, ele->precision);
         break;
-    case NEU_JSON_FLOAT:
-        ob = json_realp(ele->v.val_float, ele->precision);
+    case NEU_JSON_FLOAT: {
+        double t = ele->v.val_float;
+        if (ele->precision == 0) {
+            t = format_tag_value(ele->v.val_float);
+        }
+        ob = json_realp(t, ele->precision);
         break;
+    }
     case NEU_JSON_BOOL:
         ob = json_boolean(ele->v.val_bool);
         break;
@@ -69,10 +113,14 @@ static json_t *encode_object(json_t *object, neu_json_elem_t ele)
     case NEU_JSON_STR:
         json_object_set_new(ob, ele.name, json_string(ele.v.val_str));
         break;
-    case NEU_JSON_FLOAT:
-        json_object_set_new(ob, ele.name,
-                            json_realp(ele.v.val_float, ele.precision));
+    case NEU_JSON_FLOAT: {
+        double t = ele.v.val_float;
+        if (ele.precision == 0) {
+            t = format_tag_value(ele.v.val_float);
+        }
+        json_object_set_new(ob, ele.name, json_realp(t, ele.precision));
         break;
+    }
     case NEU_JSON_DOUBLE:
         json_object_set_new(ob, ele.name,
                             json_realp(ele.v.val_double, ele.precision));

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

@@ -295,8 +295,8 @@ def get_version():
     return requests.get(url=config.BASE_URL + "/api/v2/version", headers={"Authorization": config.default_jwt})
 
 
-def prgfile_upload(driver, path, params):
-    return requests.post(url=config.BASE_URL + "/api/v2/prgfile", headers={"Authorization": config.default_jwt}, json={"driver": driver, "path": path, "parameters": params})
+def prgfile_upload(driver, path, del_flag, params):
+    return requests.post(url=config.BASE_URL + "/api/v2/prgfile", headers={"Authorization": config.default_jwt}, json={"driver": driver, "path": path, "del_flag": del_flag, "parameters": params})
 
 
 def prgfile_process(driver):
@@ -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",

tests/ut/CMakeLists.txt

@@ -64,6 +64,13 @@ target_include_directories(rolling_counter_test PRIVATE
 )
 target_link_libraries(rolling_counter_test neuron-base gtest_main gtest)
 
+add_executable(mqtt_client_test mqtt_client_test.cc)
+target_include_directories(mqtt_client_test PRIVATE 
+	${CMAKE_SOURCE_DIR}/src
+	${CMAKE_SOURCE_DIR}/include
+)
+target_link_libraries(mqtt_client_test neuron-base gtest_main gtest)
+
 include(GoogleTest)
 gtest_discover_tests(json_test)
 gtest_discover_tests(http_test)
@@ -73,3 +80,4 @@ gtest_discover_tests(tag_sort_test)
 gtest_discover_tests(modbus_test)
 gtest_discover_tests(async_queue_test)
 gtest_discover_tests(rolling_counter_test)
+gtest_discover_tests(mqtt_client_test)

tests/ut/mqtt_client_test.cc

@@ -0,0 +1,129 @@
+#include <gtest/gtest.h>
+
+#include "connection/mqtt_client.h"
+#include "utils/log.h"
+
+zlog_category_t *neuron = NULL;
+
+TEST(MQTTClientTest, test_topic_filter_is_valid)
+{
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid(NULL));
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid(""));
+
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid("sport/tennis#"));
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid("sport/tennis/#/ranking"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("#"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport/#"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport/tennis/player/#"));
+
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid("sport+"));
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_valid("sport/+player"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("+"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("+/+"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport/+/player"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("+/tennis/#"));
+
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("$SYS"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("$SYS/#"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport/tennis"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_valid("sport/tennis/player"));
+    ASSERT_EQ(true,
+              neu_mqtt_topic_filter_is_valid("sport/tennis/player/ranking"));
+}
+
+TEST(MQTTClientTest, test_topic_filter_is_match)
+{
+    const char *filter = "#";
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "$sport"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/tennis"));
+    ASSERT_EQ(true,
+              neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player"));
+
+    filter = "sport/tennis/player/#";
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport"));
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/tennis"));
+    ASSERT_EQ(false,
+              neu_mqtt_topic_filter_is_match(filter, "sport/tennis/play"));
+    ASSERT_EQ(true,
+              neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player"));
+    ASSERT_EQ(true,
+              neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/"));
+    ASSERT_EQ(
+        true,
+        neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/ranking"));
+    ASSERT_EQ(true,
+              neu_mqtt_topic_filter_is_match(
+                  filter, "sport/tennis/player/ranking/score/wimbledon"));
+
+    filter = "+";
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "$finance"));
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "/finance"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "finance"));
+
+    filter = "/+";
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "finance"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "/finance"));
+
+    filter = "+/+";
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "$finance"));
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "finance"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "/finance"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/tennis"));
+
+    filter = "sport/+";
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/tennis"));
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/tennis/"));
+    ASSERT_EQ(false,
+              neu_mqtt_topic_filter_is_match(filter, "sport/tennis/ranking"));
+
+    filter = "sport/tennis/+";
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport"));
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/"));
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/tennis"));
+    ASSERT_EQ(
+        false,
+        neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/ranking"));
+    ASSERT_EQ(true,
+              neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player1"));
+    ASSERT_EQ(true,
+              neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player2"));
+
+    filter = "sport/+/player";
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport"));
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/"));
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/tennis"));
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/tennis/"));
+    ASSERT_EQ(false,
+              neu_mqtt_topic_filter_is_match(filter, "sport/tennis/ranking"));
+    ASSERT_EQ(true,
+              neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player"));
+    ASSERT_EQ(false,
+              neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/"));
+    ASSERT_EQ(
+        false,
+        neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/ranking"));
+
+    filter = "+/tennis/#";
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport"));
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/"));
+    ASSERT_EQ(false, neu_mqtt_topic_filter_is_match(filter, "sport/baseball"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/tennis"));
+    ASSERT_EQ(true, neu_mqtt_topic_filter_is_match(filter, "sport/tennis/"));
+    ASSERT_EQ(true,
+              neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player"));
+    ASSERT_EQ(
+        true,
+        neu_mqtt_topic_filter_is_match(filter, "sport/tennis/player/ranking"));
+}
+
+int main(int argc, char **argv)
+{
+    zlog_init("./config/dev.conf");
+    neuron = zlog_get_category("neuron");
+    ::testing::InitGoogleTest(&argc, argv);
+    return RUN_ALL_TESTS();
+}

version

@@ -1 +1 @@
-2.7.0-dev
+2.8.2