Merge pull request #817 from fengzeroz/v2.2

version: 2.2.3

Install-dependencies.md

@@ -12,7 +12,7 @@ $ cd zlog
 $ make && sudo make install
 ```
 
-[jansson](https://github.com/akheron/jansson)
+[jansson](https://github.com/neugates/jansson.git)
 ```shell
 $ git clone https://github.com/neugates/jansson.git
 $ cd jansson && mkdir build && cd build

ft/tag_test.robot

@@ -72,15 +72,15 @@ Update tag, it should return success.
 	${len} =	Get Length	${res}[tags]
  	Should Be Equal As Integers	${len}		2
 	
-	Should Be Equal As Strings	${res}[tags][1][name]		tag1
-	Should Be Equal As Strings	${res}[tags][1][address]	1!400002
-	Should Be Equal As Strings	${res}[tags][1][attribute]	${TAG_ATTRIBUTE_RW}
-	Should Be Equal As Strings	${res}[tags][1][type]		${TAG_TYPE_INT32}
-
-	Should Be Equal As Strings	${res}[tags][0][name]		tag2
-	Should Be Equal As Strings	${res}[tags][0][address]	1!00001
+	Should Be Equal As Strings	${res}[tags][0][name]		tag1
+	Should Be Equal As Strings	${res}[tags][0][address]	1!400002
 	Should Be Equal As Strings	${res}[tags][0][attribute]	${TAG_ATTRIBUTE_RW}
-	Should Be Equal As Strings	${res}[tags][0][type]		${TAG_TYPE_BIT}
+	Should Be Equal As Strings	${res}[tags][0][type]		${TAG_TYPE_INT32}
+
+	Should Be Equal As Strings	${res}[tags][1][name]		tag2
+	Should Be Equal As Strings	${res}[tags][1][address]	1!00001
+	Should Be Equal As Strings	${res}[tags][1][attribute]	${TAG_ATTRIBUTE_RW}
+	Should Be Equal As Strings	${res}[tags][1][type]		${TAG_TYPE_BIT}
 
 Delete tag from non-existent group, it should return success
 	${res}=		Del Tags  modbus-node  group1  "tag1"
@@ -100,15 +100,15 @@ Update non-existent tag, it should return failure.
 	${len} =	Get Length	${res}[tags]
  	Should Be Equal As Integers	${len}		2
 	
-	Should Be Equal As Strings	${res}[tags][1][name]		tag1
-	Should Be Equal As Strings	${res}[tags][1][address]	1!400002
-	Should Be Equal As Strings	${res}[tags][1][attribute]	${TAG_ATTRIBUTE_RW}
-	Should Be Equal As Strings	${res}[tags][1][type]		${TAG_TYPE_INT32}
-
-	Should Be Equal As Strings	${res}[tags][0][name]		tag2
-	Should Be Equal As Strings	${res}[tags][0][address]	1!00001
+	Should Be Equal As Strings	${res}[tags][0][name]		tag1
+	Should Be Equal As Strings	${res}[tags][0][address]	1!400002
 	Should Be Equal As Strings	${res}[tags][0][attribute]	${TAG_ATTRIBUTE_RW}
-	Should Be Equal As Strings	${res}[tags][0][type]		${TAG_TYPE_BIT}
+	Should Be Equal As Strings	${res}[tags][0][type]		${TAG_TYPE_INT32}
+
+	Should Be Equal As Strings	${res}[tags][1][name]		tag2
+	Should Be Equal As Strings	${res}[tags][1][address]	1!00001
+	Should Be Equal As Strings	${res}[tags][1][attribute]	${TAG_ATTRIBUTE_RW}
+	Should Be Equal As Strings	${res}[tags][1][type]		${TAG_TYPE_BIT}
 
 Delete tags, it should return success
 	Del Tags  modbus-node  group  "tag1","tag2"

plugins/modbus/modbus-tcp.json

@@ -2,27 +2,27 @@
 	"tag_regex": [
 		{
 			"type": 3,
-			"regex": "[0-9]+![3-4][0-9]+(#B|#L|)$"
+			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
 		},
 		{
 			"type": 4,
-			"regex": "[0-9]+![3-4][0-9]+(#B|#L|)$"
+			"regex": "^[0-9]+![3-4][0-9]+(#B|#L|)$"
 		},
 		{
 			"type": 5,
-			"regex": "[0-9]+![3-4][0-9]+(#BB|#BL|#LL|#LB|)$"
+			"regex": "^[0-9]+![3-4][0-9]+(#BB|#BL|#LL|#LB|)$"
 		},
 		{
 			"type": 6,
-			"regex": "[0-9]+![3-4][0-9]+(#BB|#BL|#LL|#LB|)$"
+			"regex": "^[0-9]+![3-4][0-9]+(#BB|#BL|#LL|#LB|)$"
 		},
 		{
 			"type": 9,
-			"regex": "[0-9]+![3-4][0-9]+(#BB|#BL|#LL|#LB|)$"
+			"regex": "^[0-9]+![3-4][0-9]+(#BB|#BL|#LL|#LB|)$"
 		},
 		{
 			"type": 11,
-			"regex": "[0-9]+!([0-1][0-9]+|[3-4][0-9]+.([0-9]|[0-1][0-5]))$"
+			"regex": "^[0-9]+!([0-1][0-9]+|[3-4][0-9]+\\.([0-9]|[0-1][0-5]))$"
 		}
 	],
 	"host": {

plugins/modbus/modbus_req.c

@@ -168,7 +168,7 @@ int modbus_value_handle(void *ctx, uint8_t slave_id, uint16_t n_byte,
                     neu_value16_u v16 = { 0 };
                     v16.value         = htons(*(uint16_t *) dvalue.value.bytes);
                     dvalue.value.u8 =
-                        neu_value16_get_bit(v16, 15 - (*p_tag)->option.bit.bit);
+                        neu_value16_get_bit(v16, (*p_tag)->option.bit.bit);
                     break;
                 }
                 case MODBUS_AREA_COIL:

plugins/mqtt/mqtt.json

@@ -21,7 +21,7 @@
 	},
 	"format": {
 		"name": "upload format",
-		"description": "Upload format on mqtt",
+		"description": "The json format of the data reported in the upload topic. In the values mode, all items are contained in the values object or the errors object, respectively. In tags mode, all items are contained in an array",
 		"attribute": "optional",
 		"type": "map",
 		"default": 0,
@@ -40,7 +40,7 @@
 	},
 	"cache": {
 		"name": "cache size(MB)",
-		"description": "Upload data memory cache",
+		"description": "The maximum byte limit in MB for the data backlog when an MQTT connection exception occurs",
 		"type": "int",
 		"attribute": "optional",
 		"default": 64,

plugins/restful/license_handle.c

@@ -19,6 +19,7 @@
 
 #define _POSIX_C_SOURCE 200809L
 #include <sys/stat.h>
+#include <sys/types.h>
 #include <time.h>
 #include <unistd.h>
 
@@ -43,10 +44,95 @@ int        get_plugin_names(const license_t *lic, UT_array *plugin_names);
 static inline char *get_license_path()
 {
     char *s = NULL;
-    neu_asprintf(&s, "%s/%s", g_config_dir, LICENSE_FNAME);
+    neu_asprintf(&s, "persistence/%s", LICENSE_FNAME);
     return s;
 }
 
+static int copy_file(const char *from, const char *to)
+{
+    int         rv      = 0;
+    char *      buf     = NULL;
+    FILE *      fp      = NULL;
+    struct stat statbuf = {};
+
+    if (0 != stat(from, &statbuf)) {
+        return NEU_ERR_EINTERNAL;
+    }
+
+    buf = malloc(statbuf.st_size);
+    if (NULL == buf) {
+        return NEU_ERR_EINTERNAL;
+    }
+
+    fp = fopen(from, "rb");
+    if (NULL == fp) {
+        rv = NEU_ERR_EINTERNAL;
+        goto end;
+    }
+
+    if ((size_t) statbuf.st_size != fread(buf, 1, statbuf.st_size, fp)) {
+        rv = NEU_ERR_EINTERNAL;
+        goto end;
+    }
+
+    fclose(fp);
+    fp = fopen(to, "w");
+    if (NULL == fp) {
+        rv = NEU_ERR_EINTERNAL;
+        goto end;
+    }
+
+    if ((size_t) statbuf.st_size != fwrite(buf, 1, statbuf.st_size, fp)) {
+        rv = NEU_ERR_EINTERNAL;
+        goto end;
+    }
+
+end:
+    if (fp) {
+        fclose(fp);
+    }
+    free(buf);
+    return rv;
+}
+
+int copy_license_file_if_necessary()
+{
+    struct stat statbuf      = {};
+    char *      license_path = get_license_path();
+
+    if (NULL == license_path) {
+        return NEU_ERR_EINTERNAL;
+    }
+
+    if (0 == stat(license_path, &statbuf)) {
+        free(license_path);
+        return 0;
+    }
+
+    nlog_warn("license `%s` not found", license_path);
+
+    char *fallback_path = NULL;
+    neu_asprintf(&fallback_path, "%s/%s", g_config_dir, LICENSE_FNAME);
+    if (NULL == fallback_path) {
+        free(license_path);
+        return NEU_ERR_EINTERNAL;
+    }
+
+    if (0 != stat(fallback_path, &statbuf)) {
+        // no fallback license file
+        free(license_path);
+        free(fallback_path);
+        return 0;
+    }
+
+    nlog_warn("fallback to `%s`", fallback_path);
+    int rv = copy_file(fallback_path, license_path);
+
+    free(license_path);
+    free(fallback_path);
+    return rv;
+}
+
 void handle_get_license(nng_aio *aio)
 {
     int                         rv              = 0;

plugins/restful/license_handle.h

@@ -22,6 +22,7 @@
 
 #include <nng/nng.h>
 
+int  copy_license_file_if_necessary();
 void handle_get_license(nng_aio *aio);
 void handle_set_license(nng_aio *aio);
 

plugins/restful/rest.c

@@ -31,6 +31,7 @@
 #include "group_config_handle.h"
 #include "handle.h"
 #include "http.h"
+#include "license_handle.h"
 #include "plugin_handle.h"
 #include "proxy.h"
 #include "rest.h"
@@ -145,6 +146,12 @@ static neu_plugin_t *dashb_plugin_open(void)
     const struct neu_rest_handler *rest_handlers = NULL;
     const struct neu_rest_handler *cors          = NULL;
 
+    // copy license file for backward compatibility
+    if (0 != copy_license_file_if_necessary()) {
+        free(plugin);
+        return NULL;
+    }
+
     neu_plugin_common_init(&plugin->common);
 
     plugin->handle_ctx = neu_rest_init_ctx(plugin);
@@ -302,6 +309,8 @@ static int dashb_plugin_request(neu_plugin_t *      plugin,
         handle_read_resp(header->ctx, (neu_resp_read_group_t *) data);
         break;
     default:
+        nlog_fatal("recv unhandle msg: %s",
+                   neu_reqresp_type_string(header->type));
         assert(false);
         break;
     }

src/adapter/adapter.c

@@ -542,16 +542,20 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
                     neu_adapter_driver_add_tag((neu_adapter_driver_t *) adapter,
                                                cmd->group, &cmd->tags[i]);
                 if (ret == 0) {
-                    adapter_storage_add_tag(adapter->persister, cmd->driver,
-                                            cmd->group, &cmd->tags[i]);
                     resp.index += 1;
-
                 } else {
                     resp.error = ret;
                     break;
                 }
             }
         }
+
+        if (resp.index) {
+            // we have added some tags, try to persist
+            adapter_storage_add_tags(adapter->persister, cmd->driver,
+                                     cmd->group, cmd->tags, resp.index);
+        }
+
         for (int i = 0; i < cmd->n_tag; i++) {
             free(cmd->tags[i].address);
             free(cmd->tags[i].name);

src/adapter/driver/driver.c

@@ -751,8 +751,14 @@ static int read_callback(void *usr_data)
     }
 
     if (group->grp.tags != NULL) {
+        int64_t spend = group->driver->adapter.timestamp;
+
         group->driver->adapter.module->intf_funs->driver.group_timer(
             group->driver->adapter.plugin, &group->grp);
+
+        spend = group->driver->adapter.timestamp - spend;
+        nlog_info("%s-%s timer: %" PRId64, group->driver->adapter.name,
+                  group->name, spend);
     }
 
     return 0;
@@ -856,21 +862,33 @@ static int read_report_group(int64_t timestamp, int64_t timeout,
                 datas[index].value.type = NEU_TYPE_DOUBLE;
                 switch (tag->type) {
                 case NEU_TYPE_INT8:
+                    datas[index].value.value.d64 =
+                        (double) datas[index].value.value.i8 * tag->decimal;
+                    break;
                 case NEU_TYPE_UINT8:
                     datas[index].value.value.d64 =
                         (double) datas[index].value.value.u8 * tag->decimal;
                     break;
                 case NEU_TYPE_INT16:
+                    datas[index].value.value.d64 =
+                        (double) datas[index].value.value.i16 * tag->decimal;
+                    break;
                 case NEU_TYPE_UINT16:
                     datas[index].value.value.d64 =
                         (double) datas[index].value.value.u16 * tag->decimal;
                     break;
                 case NEU_TYPE_INT32:
+                    datas[index].value.value.d64 =
+                        (double) datas[index].value.value.i32 * tag->decimal;
+                    break;
                 case NEU_TYPE_UINT32:
                     datas[index].value.value.d64 =
                         (double) datas[index].value.value.u32 * tag->decimal;
                     break;
                 case NEU_TYPE_INT64:
+                    datas[index].value.value.d64 =
+                        (double) datas[index].value.value.i64 * tag->decimal;
+                    break;
                 case NEU_TYPE_UINT64:
                     datas[index].value.value.d64 =
                         (double) datas[index].value.value.u64 * tag->decimal;
@@ -982,21 +1000,33 @@ static void read_group(int64_t timestamp, int64_t timeout,
                 datas[index].value.type = NEU_TYPE_DOUBLE;
                 switch (tag->type) {
                 case NEU_TYPE_INT8:
+                    datas[index].value.value.d64 =
+                        (double) datas[index].value.value.i8 * tag->decimal;
+                    break;
                 case NEU_TYPE_UINT8:
                     datas[index].value.value.d64 =
                         (double) datas[index].value.value.u8 * tag->decimal;
                     break;
                 case NEU_TYPE_INT16:
+                    datas[index].value.value.d64 =
+                        (double) datas[index].value.value.i16 * tag->decimal;
+                    break;
                 case NEU_TYPE_UINT16:
                     datas[index].value.value.d64 =
                         (double) datas[index].value.value.u16 * tag->decimal;
                     break;
                 case NEU_TYPE_INT32:
+                    datas[index].value.value.d64 =
+                        (double) datas[index].value.value.i32 * tag->decimal;
+                    break;
                 case NEU_TYPE_UINT32:
                     datas[index].value.value.d64 =
                         (double) datas[index].value.value.u32 * tag->decimal;
                     break;
                 case NEU_TYPE_INT64:
+                    datas[index].value.value.d64 =
+                        (double) datas[index].value.value.i64 * tag->decimal;
+                    break;
                 case NEU_TYPE_UINT64:
                     datas[index].value.value.d64 =
                         (double) datas[index].value.value.u64 * tag->decimal;

src/adapter/storage.c

@@ -77,6 +77,25 @@ void adapter_storage_add_tag(neu_persister_t *persister, const char *node,
     }
 }
 
+void adapter_storage_add_tags(neu_persister_t *persister, const char *node,
+                              const char *group, const neu_datatag_t *tags,
+                              size_t n)
+{
+    if (0 == n) {
+        return;
+    }
+
+    if (1 == n) {
+        return adapter_storage_add_tag(persister, node, group, &tags[0]);
+    }
+
+    int rv = neu_persister_store_tags(persister, node, group, tags, n);
+    if (0 != rv) {
+        nlog_error("fail store %zu tags:[%s ... %s] adapter:%s grp:%s", n,
+                   tags->name, tags[n - 1].name, node, group);
+    }
+}
+
 void adapter_storage_update_tag(neu_persister_t *persister, const char *node,
                                 const char *group, const neu_datatag_t *tag)
 {

src/adapter/storage.h

@@ -36,6 +36,9 @@ void adapter_storage_del_group(neu_persister_t *persister, const char *node,
                                const char *group);
 void adapter_storage_add_tag(neu_persister_t *persister, const char *node,
                              const char *group, const neu_datatag_t *tag);
+void adapter_storage_add_tags(neu_persister_t *persister, const char *node,
+                              const char *group, const neu_datatag_t *tags,
+                              size_t n);
 void adapter_storage_update_tag(neu_persister_t *persister, const char *node,
                                 const char *group, const neu_datatag_t *tag);
 void adapter_storage_del_tag(neu_persister_t *persister, const char *node,

src/base/group.c

@@ -142,28 +142,19 @@ int neu_group_update_tag(neu_group_t *group, neu_datatag_t *tag)
 
     HASH_FIND_STR(group->tags, tag->name, el);
     if (el != NULL) {
-        HASH_DEL(group->tags, el);
-        free(el->name);
-        free(el->tag->name);
         free(el->tag->address);
-        free(el->tag->description);
-        free(el->tag);
-        free(el);
+        el->tag->address = strdup(tag->address);
 
-        el                   = calloc(1, sizeof(tag_elem_t));
-        el->name             = strdup(tag->name);
-        el->tag              = calloc(1, sizeof(neu_datatag_t));
-        el->tag->name        = strdup(tag->name);
-        el->tag->address     = strdup(tag->address);
+        free(el->tag->description);
         el->tag->description = strdup(tag->description);
-        el->tag->type        = tag->type;
-        el->tag->attribute   = tag->attribute;
-        el->tag->precision   = tag->precision;
-        el->tag->decimal     = tag->decimal;
-        el->tag->option      = tag->option;
+
+        el->tag->type      = tag->type;
+        el->tag->attribute = tag->attribute;
+        el->tag->precision = tag->precision;
+        el->tag->decimal   = tag->decimal;
+        el->tag->option    = tag->option;
         memcpy(el->tag->meta, tag->meta, sizeof(tag->meta));
 
-        HASH_ADD_STR(group->tags, name, el);
         update_timestamp(group);
         ret = NEU_ERR_SUCCESS;
     }

src/connection/mqtt_c_client.c

@@ -509,9 +509,19 @@ static void client_reconnect(struct mqtt_client *mqtt, void **p_reconnect_state)
     const char *will_payload = state->will_payload;
     size_t      payload_len  = state->will_payload_len;
 
-    enum MQTTErrors rc = mqtt_connect(
-        mqtt, client_id, will_topic, will_payload, payload_len, state->username,
-        state->password, connect_flags, state->keepalive);
+    char *username = NULL;
+    if (NULL != state->username && 0 < strlen(state->username)) {
+        username = state->username;
+    }
+
+    char *password = NULL;
+    if (NULL != state->password && 0 < strlen(state->password)) {
+        password = state->password;
+    }
+
+    enum MQTTErrors rc =
+        mqtt_connect(mqtt, client_id, will_topic, will_payload, payload_len,
+                     username, password, connect_flags, state->keepalive);
 
     if (MQTT_OK != rc) {
         return;

src/main.c

@@ -38,7 +38,9 @@ static void sig_handler(int sig)
 {
     nlog_warn("recv sig: %d", sig);
 
-    neu_manager_destroy(g_manager);
+    if (sig != SIGABRT) {
+        neu_manager_destroy(g_manager);
+    }
     zlog_fini();
     exit_flag = true;
 }

src/persist/persist.c

@@ -629,6 +629,14 @@ neu_persister_t *neu_persister_create(const char *dir_name)
         sqlite3_close(persister->db);
         goto error;
     }
+    if (SQLITE_OK !=
+        sqlite3_exec(persister->db, "PRAGMA journal_mode=WAL", NULL, NULL,
+                     NULL)) {
+        nlog_error("db journal_mode WAL fail: %s",
+                   sqlite3_errmsg(persister->db));
+        sqlite3_close(persister->db);
+        goto error;
+    }
     if (0 != apply_schemas(persister->db, g_config_dir)) {
         goto error;
     }
@@ -908,6 +916,109 @@ int neu_persister_store_tag(neu_persister_t *persister, const char *driver_name,
                        tag->description);
 }
 
+int neu_persister_store_tags(neu_persister_t *persister,
+                             const char *driver_name, const char *group_name,
+                             const neu_datatag_t *tags, size_t n)
+{
+    sqlite3_stmt *stmt  = NULL;
+    const char *  query = "INSERT INTO tags (driver_name, group_name, name, "
+                        "address, attribute, precision, type, decimal, "
+                        "description) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?)";
+
+    if (SQLITE_OK != sqlite3_exec(persister->db, "BEGIN", NULL, NULL, NULL)) {
+        nlog_error("begin transaction fail: %s", sqlite3_errmsg(persister->db));
+        return NEU_ERR_EINTERNAL;
+    }
+
+    if (SQLITE_OK !=
+        sqlite3_prepare_v2(persister->db, query, -1, &stmt, NULL)) {
+        nlog_error("prepare `%s` fail: %s", query,
+                   sqlite3_errmsg(persister->db));
+        goto error;
+    }
+
+    for (size_t i = 0; i < n; ++i) {
+        const neu_datatag_t *tag = &tags[i];
+
+        sqlite3_reset(stmt);
+
+        if (SQLITE_OK != sqlite3_bind_text(stmt, 1, driver_name, -1, NULL)) {
+            nlog_error("bind `%s` with driver_name=`%s` fail: %s", query,
+                       driver_name, sqlite3_errmsg(persister->db));
+            goto error;
+        }
+
+        if (SQLITE_OK != sqlite3_bind_text(stmt, 2, group_name, -1, NULL)) {
+            nlog_error("bind `%s` with group_name=`%s` fail: %s", query,
+                       group_name, sqlite3_errmsg(persister->db));
+            goto error;
+        }
+
+        if (SQLITE_OK != sqlite3_bind_text(stmt, 3, tag->name, -1, NULL)) {
+            nlog_error("bind `%s` with name=`%s` fail: %s", query, tag->name,
+                       sqlite3_errmsg(persister->db));
+            goto error;
+        }
+
+        if (SQLITE_OK != sqlite3_bind_text(stmt, 4, tag->address, -1, NULL)) {
+            nlog_error("bind `%s` with address=`%s` fail: %s", query,
+                       tag->address, sqlite3_errmsg(persister->db));
+            goto error;
+        }
+
+        if (SQLITE_OK != sqlite3_bind_int(stmt, 5, tag->attribute)) {
+            nlog_error("bind `%s` with attribute=`%i` fail: %s", query,
+                       tag->attribute, sqlite3_errmsg(persister->db));
+            goto error;
+        }
+
+        if (SQLITE_OK != sqlite3_bind_int(stmt, 6, tag->precision)) {
+            nlog_error("bind `%s` with precision=`%i` fail: %s", query,
+                       tag->precision, sqlite3_errmsg(persister->db));
+            goto error;
+        }
+
+        if (SQLITE_OK != sqlite3_bind_int(stmt, 7, tag->type)) {
+            nlog_error("bind `%s` with type=`%i` fail: %s", query, tag->type,
+                       sqlite3_errmsg(persister->db));
+            goto error;
+        }
+
+        if (SQLITE_OK != sqlite3_bind_double(stmt, 8, tag->decimal)) {
+            nlog_error("bind `%s` with decimal=`%f` fail: %s", query,
+                       tag->decimal, sqlite3_errmsg(persister->db));
+            goto error;
+        }
+
+        if (SQLITE_OK !=
+            sqlite3_bind_text(stmt, 9, tag->description, -1, NULL)) {
+            nlog_error("bind `%s` with description=`%s` fail: %s", query,
+                       tag->description, sqlite3_errmsg(persister->db));
+            goto error;
+        }
+
+        if (SQLITE_DONE != sqlite3_step(stmt)) {
+            nlog_error("sqlite3_step fail: %s", sqlite3_errmsg(persister->db));
+            goto error;
+        }
+    }
+
+    if (SQLITE_OK != sqlite3_exec(persister->db, "COMMIT", NULL, NULL, NULL)) {
+        nlog_error("commit transaction fail: %s",
+                   sqlite3_errmsg(persister->db));
+        goto error;
+    }
+
+    sqlite3_finalize(stmt);
+    return 0;
+
+error:
+    nlog_info("rollback transaction");
+    sqlite3_exec(persister->db, "ROLLBACK", NULL, NULL, NULL);
+    sqlite3_finalize(stmt);
+    return NEU_ERR_EINTERNAL;
+}
+
 int neu_persister_load_tags(neu_persister_t *persister, const char *driver_name,
                             const char *group_name, UT_array **tags)
 {

src/persist/persist.h

@@ -161,6 +161,10 @@ int neu_persister_load_plugins(neu_persister_t *persister,
 int neu_persister_store_tag(neu_persister_t *persister, const char *driver_name,
                             const char *group_name, const neu_datatag_t *tag);
 
+int neu_persister_store_tags(neu_persister_t *persister,
+                             const char *driver_name, const char *group_name,
+                             const neu_datatag_t *tags, size_t n);
+
 /**
  * Load node tag infos.
  * @param persister                 persiter object.

src/utils/async_queue.c

@@ -26,7 +26,7 @@ config_ **/
 #include "utils/utextend.h"
 
 typedef struct element {
-    uint32_t key;
+    uint64_t key;
 
     void *data;
 
@@ -107,39 +107,26 @@ void neu_async_queue_push(neu_async_queue_t *q, void *elem)
 
 int neu_async_queue_pop(neu_async_queue_t *q, uint64_t key, void **elem)
 {
-    element *elt = NULL, *first = NULL;
+    element *elt = NULL, *tmp = NULL;
     int      ret = -1;
 
     nng_mtx_lock(q->mtx);
-    first = q->list;
-    elt   = DL_LAST(q->list);
-
-    while (elt != NULL) {
+    DL_FOREACH_SAFE(q->list, elt, tmp)
+    {
         if (elt->key == key) {
             DL_DELETE(q->list, elt);
             *elem = elt->data;
             free(elt);
             ret = 0;
             break;
-        }
-
-        if (first == elt) {
-            break;
-        }
-
-        if (q->expire_fn(elt->data)) {
-            element *tmp = DL_PREV(elt);
-
-            DL_DELETE(q->list, elt);
-            q->free_fn(elt->data);
-            free(elt);
-
-            elt = tmp;
         } else {
-            elt = DL_PREV(elt);
+            if (q->expire_fn(elt->data)) {
+                DL_DELETE(q->list, elt);
+                q->free_fn(elt->data);
+                free(elt);
+            }
         }
     }
-
     nng_mtx_unlock(q->mtx);
 
     return ret;

ui_version

@@ -1 +1 @@
-2.2.1
+2.2.3

version

@@ -1 +1 @@
-2.2.1
+2.2.3