Merge pull request #2023 from fengzeroz/v2.8

version: 2.8.2

.github/workflows/ci.yml

@@ -8,7 +8,7 @@ on:
       - published
 
 env:
-  VERSION: 2.8.1
+  VERSION: 2.8.2
   UI_VERSION: 2.6.1
   BRANCH: v2.8
 

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

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

@@ -24,7 +24,7 @@
 
 #define NEU_VERSION_MAJOR 2
 #define NEU_VERSION_MINOR 8
-#define NEU_VERSION_FIX 1
+#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/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

@@ -906,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/type.h

@@ -136,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:

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

@@ -426,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)
 {
@@ -1344,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);
 }
 
@@ -1389,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;
@@ -1427,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/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,
@@ -890,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);
 
@@ -910,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,
@@ -975,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) {
@@ -1005,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;
 }
@@ -1045,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);
         }
@@ -1689,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 {
@@ -1834,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);
@@ -1989,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/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/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++;
     }
 

src/parser/neu_json_tag.c

@@ -294,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[] = {
@@ -565,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[] = {
@@ -659,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",
@@ -775,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/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/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):

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.8.1
+2.8.2