Merge pull request #835 from fengzeroz/v2.2

version: 2.2.5

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"

include/neuron/errcodes.h

@@ -87,6 +87,7 @@ typedef enum {
     NEU_ERR_LICENSE_MAX_NODES       = 2404,
     NEU_ERR_LICENSE_MAX_TAGS        = 2405,
     NEU_ERR_LICENSE_TOKEN_NOT_MATCH = 2406,
+    NEU_ERR_LICENSE_BAD_CLOCK       = 2407,
 
     // common error codes for plugins
     NEU_ERR_PLUGIN_READ_FAILURE            = 3000,

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/command/read_write.c

@@ -112,6 +112,7 @@ static void wrap_read_response_json(neu_resp_tag_value_t *tags, uint16_t len,
         case NEU_TYPE_DOUBLE:
             json->tags[i].t                = NEU_JSON_DOUBLE;
             json->tags[i].value.val_double = tag->value.value.d64;
+            json->tags[i].precision        = tag->value.precision;
             break;
         case NEU_TYPE_BOOL:
             json->tags[i].t              = NEU_JSON_BOOL;

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

@@ -65,7 +65,7 @@ static uint64_t mktimestamp(struct tm dt)
     uint64_t ret;
     int      i = 0;
 
-    ret = (dt.tm_year - 1970) * 365 * 24 * 3600;
+    ret = ((uint64_t) dt.tm_year - 1970) * 365 * 24 * 3600;
     ret +=
         (mon_yday[isleap(dt.tm_year)][dt.tm_mon] + dt.tm_mday - 1) * 24 * 3600;
     ret += dt.tm_hour * 3600 + dt.tm_min * 60 + dt.tm_sec;

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

plugins/restful/rw_handle.c

@@ -153,6 +153,7 @@ void handle_read_resp(nng_aio *aio, neu_resp_read_group_t *resp)
         case NEU_TYPE_DOUBLE:
             api_res.tags[i].t                = NEU_JSON_DOUBLE;
             api_res.tags[i].value.val_double = resp->tags[i].value.value.d64;
+            api_res.tags[i].precision        = resp->tags[i].value.precision;
             break;
         case NEU_TYPE_BIT:
             api_res.tags[i].t             = NEU_JSON_BIT;

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

@@ -253,6 +253,11 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
             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.u64;
+            }
+        }
         switch (tag->type) {
         case NEU_TYPE_BOOL:
         case NEU_TYPE_STRING:
@@ -746,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;
@@ -851,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;
@@ -977,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/connection.c

@@ -366,7 +366,9 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
 
 void neu_conn_disconnect(neu_conn_t *conn)
 {
+    pthread_mutex_lock(&conn->mtx);
     conn_disconnect(conn);
+    pthread_mutex_unlock(&conn->mtx);
 }
 
 static void conn_free_param(neu_conn_t *conn)

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;
@@ -549,7 +559,8 @@ static void publish_callback(void **                       p_reconnect_state,
     topic_name[published->topic_name_size] = '\0';
 
     if (NULL != client->log) {
-        zlog_debug(client->log, "received publish('%s'): %s", topic_name,
+        zlog_debug(client->log, "received publish('%s'): %.*s", topic_name,
+                   (int) published->application_message_size,
                    (const char *) published->application_message);
     }
 

src/core/manager.c

@@ -311,6 +311,10 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
         }
 
         manager_storage_del_node(manager, cmd->node);
+        if (neu_adapter_get_type(adapter) == NEU_NA_TYPE_APP) {
+            neu_subscribe_manager_unsub_all(manager->subscribe_manager,
+                                            cmd->node);
+        }
         header->type = NEU_REQ_NODE_UNINIT;
         forward_msg(manager, msg, header->receiver);
         if (neu_adapter_get_type(adapter) == NEU_NA_TYPE_DRIVER) {

src/core/subscribe.c

@@ -127,6 +127,23 @@ UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app)
     return groups;
 }
 
+void neu_subscribe_manager_unsub_all(neu_subscribe_mgr_t *mgr, const char *app)
+{
+    sub_elem_t *el = NULL, *tmp = NULL;
+
+    HASH_ITER(hh, mgr->ss, el, tmp)
+    {
+        utarray_foreach(el->apps, neu_app_subscribe_t *, sub_app)
+        {
+            if (strcmp(sub_app->app_name, app) == 0) {
+                neu_subscribe_manager_unsub(mgr, el->key.driver, app,
+                                            el->key.group);
+                break;
+            }
+        }
+    }
+}
+
 int neu_subscribe_manager_sub(neu_subscribe_mgr_t *mgr, const char *driver,
                               const char *app, const char *group, nng_pipe pipe)
 {

src/core/subscribe.h

@@ -41,6 +41,7 @@ UT_array *neu_subscribe_manager_find(neu_subscribe_mgr_t *mgr,
 UT_array *neu_subscribe_manager_find_by_driver(neu_subscribe_mgr_t *mgr,
                                                const char *         driver);
 UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app);
+void neu_subscribe_manager_unsub_all(neu_subscribe_mgr_t *mgr, const char *app);
 int  neu_subscribe_manager_sub(neu_subscribe_mgr_t *mgr, const char *driver,
                                const char *app, const char *group,
                                nng_pipe pipe);

src/main.c

@@ -27,6 +27,7 @@
 
 #include "argparse.h"
 #include "daemon.h"
+#include "version.h"
 
 static bool           exit_flag   = false;
 static neu_manager_t *g_manager   = NULL;
@@ -37,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;
 }
@@ -56,7 +59,9 @@ static int neuron_run(const neu_cli_args_t *args)
         nlog_warn("neuron process failed enable core dump, ignore");
     }
 
-    zlog_notice(neuron, "neuron process, daemon: %d", args->daemonized);
+    zlog_notice(neuron, "neuron process, daemon: %d, version: %s (%s %s)",
+                args->daemonized, NEURON_VERSION,
+                NEURON_GIT_REV NEURON_GIT_DIFF, NEURON_BUILD_DATE);
     g_manager = neu_manager_create();
     if (g_manager == NULL) {
         nlog_fatal("neuron process failed to create neuron manager, exit!");

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)
 {
@@ -1095,7 +1206,7 @@ int neu_persister_update_group(neu_persister_t *         persister,
 {
     return execute_sql(persister->db,
                        "UPDATE groups SET driver_name=%Q, name=%Q, "
-                       "interval=%i, WHERE driver_name=%Q AND name=%Q",
+                       "interval=%i WHERE driver_name=%Q AND name=%Q",
                        driver_name, group_info->name, group_info->interval,
                        driver_name, group_info->name);
 }

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;

src/utils/json.c

@@ -131,10 +131,18 @@ static int decode_object(json_t *root, neu_json_elem_t *ele)
         ele->v.val_str = strdup(json_string_value(ob));
         break;
     case NEU_JSON_FLOAT:
-        ele->v.val_float = json_real_value(ob);
+        if (json_is_integer(ob)) {
+            ele->v.val_float = (float) json_integer_value(ob);
+        } else {
+            ele->v.val_float = json_real_value(ob);
+        }
         break;
     case NEU_JSON_DOUBLE:
-        ele->v.val_double = json_real_value(ob);
+        if (json_is_integer(ob)) {
+            ele->v.val_double = (double) json_integer_value(ob);
+        } else {
+            ele->v.val_double = json_real_value(ob);
+        }
         break;
     case NEU_JSON_BOOL:
         ele->v.val_bool = json_boolean_value(ob);

ui_version

@@ -1 +1 @@
-2.1.2
+2.2.3

version

@@ -1 +1 @@
-2.2.0
+2.2.5