Merge pull request #920 from fengzeroz/v2.2

version: 2.2.10

CMakeLists.txt

@@ -131,6 +131,7 @@ target_link_libraries(neuron dl neuron-base sqlite3 -lm)
 file(COPY ${CMAKE_SOURCE_DIR}/zlog.conf DESTINATION ${CMAKE_BINARY_DIR}/config)
 file(COPY ${CMAKE_SOURCE_DIR}/dev.conf DESTINATION ${CMAKE_BINARY_DIR}/config)
 file(COPY ${CMAKE_SOURCE_DIR}/.gitkeep DESTINATION ${CMAKE_BINARY_DIR}/logs)
+file(COPY ${CMAKE_SOURCE_DIR}/.gitkeep DESTINATION ${CMAKE_BINARY_DIR}/persistence)
 file(COPY ${CMAKE_SOURCE_DIR}/neuron.key DESTINATION ${CMAKE_BINARY_DIR}/config)
 file(COPY ${CMAKE_SOURCE_DIR}/neuron.pem DESTINATION ${CMAKE_BINARY_DIR}/config)
 file(GLOB SQL_SCHEMAS ${CMAKE_SOURCE_DIR}/persistence/*.sql)

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/define.h

@@ -20,8 +20,8 @@
 #ifndef _NEU_DEFINE_H_
 #define _NEU_DEFINE_H_
 
-#define NEU_TAG_NAME_LEN 32
-#define NEU_TAG_ADDRESS_LEN 64
+#define NEU_TAG_NAME_LEN 48
+#define NEU_TAG_ADDRESS_LEN 128
 #define NEU_TAG_DESCRIPTION_LEN 128
 #define NEU_GROUP_NAME_LEN 32
 #define NEU_GROUP_INTERVAL_LIMIT 100

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/ekuiper/json_rw.c

@@ -155,9 +155,10 @@ int json_encode_read_resp_tags(void *json_object, void *param)
         }
 
         neu_json_elem_t tag_elem = {
-            .name = json_tag.name,
-            .t    = json_tag.t,
-            .v    = json_tag.value,
+            .name      = json_tag.name,
+            .t         = json_tag.t,
+            .v         = json_tag.value,
+            .precision = trans_data->tags[i].value.precision,
         };
 
         ret = neu_json_encode_field((0 != json_tag.error) ? errors_object

plugins/ekuiper/plugin_ekuiper.c

@@ -114,6 +114,8 @@ static int ekuiper_plugin_start(neu_plugin_t *plugin)
 
     nng_pipe_notify(plugin->sock, NNG_PIPE_EV_ADD_POST, pipe_add_cb, plugin);
     nng_pipe_notify(plugin->sock, NNG_PIPE_EV_REM_POST, pipe_rm_cb, plugin);
+    nng_socket_set_int(plugin->sock, NNG_OPT_SENDBUF, 2048);
+    nng_socket_set_int(plugin->sock, NNG_OPT_RECVBUF, 2048);
 
     if ((rv = nng_listen(plugin->sock, EKUIPER_PLUGIN_URL, NULL, 0)) != 0) {
         plog_error(plugin, "nng_listen: %s", nng_strerror(rv));

plugins/ekuiper/read_write.c

@@ -56,7 +56,7 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data)
     rv = nng_sendmsg(plugin->sock, msg,
                      NNG_FLAG_NONBLOCK); // TODO: use aio to send message
     if (0 != rv) {
-        plog_error(plugin, "nng cannot send msg");
+        plog_error(plugin, "nng cannot send msg: %s", nng_strerror(rv));
         nng_msg_free(msg);
     }
 }

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

@@ -23,14 +23,6 @@
 
 #include "read_write.h"
 
-// static uint64_t current_time()
-// {
-//     struct timeval tv;
-//     gettimeofday(&tv, NULL);
-//     uint64_t ms = tv.tv_sec;
-//     return ms * 1000 + tv.tv_usec / 1000;
-// }
-
 int command_rw_read_once_request(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt,
                                  neu_json_read_req_t *req)
 {
@@ -112,6 +104,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/adapter_handle.c

@@ -37,7 +37,7 @@ void handle_add_adapter(nng_aio *aio)
 
     REST_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
         aio, neu_json_add_node_req_t, neu_json_decode_add_node_req, {
-            if (strlen(req->name) > NEU_NODE_NAME_LEN) {
+            if (strlen(req->name) >= NEU_NODE_NAME_LEN) {
                 NEU_JSON_RESPONSE_ERROR(NEU_ERR_NODE_NAME_TOO_LONG, {
                     http_response(aio, NEU_ERR_NODE_NAME_TOO_LONG,
                                   result_error);

plugins/restful/group_config_handle.c

@@ -36,7 +36,7 @@ void handle_add_group_config(nng_aio *aio)
     REST_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
         aio, neu_json_add_group_config_req_t,
         neu_json_decode_add_group_config_req, {
-            if (strlen(req->group) > NEU_GROUP_NAME_LEN) {
+            if (strlen(req->group) >= NEU_GROUP_NAME_LEN) {
                 NEU_JSON_RESPONSE_ERROR(NEU_ERR_GROUP_NAME_TOO_LONG, {
                     http_response(aio, NEU_ERR_GROUP_NAME_TOO_LONG,
                                   result_error);

plugins/restful/license.c

@@ -36,6 +36,7 @@
 #include <time.h>
 
 #include "errcodes.h"
+#include "utils/asprintf.h"
 #include "utils/log.h"
 
 #include "license.h"
@@ -65,7 +66,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;
@@ -81,7 +82,7 @@ static uint64_t mktimestamp(struct tm dt)
     return (ret);
 }
 
-static uint64_t get_asn1_ts(ASN1_TIME *time)
+static uint64_t get_asn1_ts(ASN1_TIME *time, char **ts_str)
 {
     struct tm   t;
     const char *str = (const char *) time->data;
@@ -97,7 +98,11 @@ static uint64_t get_asn1_ts(ASN1_TIME *time)
                &t.tm_hour, &t.tm_min, &t.tm_sec);
     }
 
+    neu_asprintf(ts_str, "%4d-%02d-%02d %02d:%02d:%02d", t.tm_year, t.tm_mon,
+                 t.tm_mday, t.tm_hour, t.tm_min, t.tm_sec);
+
     t.tm_mon -= 1;
+
     /* Note: we did not adjust the time based on time zone information */
     return mktimestamp(t);
 }
@@ -245,9 +250,14 @@ static int extract_license(const char *license_fname, const char *pub_key,
 
     not_before      = X509_get_notBefore(cert);
     not_after       = X509_get_notAfter(cert);
-    license->since_ = get_asn1_ts(not_before);
-    license->until_ = get_asn1_ts(not_after);
+    license->since_ = get_asn1_ts(not_before, &license->since_str_);
+    license->until_ = get_asn1_ts(not_after, &license->until_str_);
     nlog_info("certificate expire date: %s", not_after->data);
+    if (NULL == license->since_str_ || NULL == license->until_str_) {
+        nlog_warn("cannot strdup");
+        rv = NEU_ERR_EINTERNAL;
+        goto final;
+    }
 
     subj = X509_get_subject_name(cert);
     for (i = 0; i < X509_NAME_entry_count(subj); i++) {
@@ -273,6 +283,7 @@ static int extract_license(const char *license_fname, const char *pub_key,
         goto final;
     }
 
+    unsigned mark = 0;
     for (i = 0; i < sk_X509_EXTENSION_num(ext_list); i++) {
         ext = sk_X509_EXTENSION_value(ext_list, i);
         obj = X509_EXTENSION_get_object(ext);
@@ -286,25 +297,36 @@ static int extract_license(const char *license_fname, const char *pub_key,
         if (0 == strcmp(buf, LICENSE_KEY_TYPE)) {
             license->type_ = scan_licese_type(val);
             if (LICENSE_TYPE_MAX == license->type_) {
-                zlog_error(neuron, "invalid license type:`%s`", val);
+                nlog_error("invalid license type:`%s`", val);
                 rv = NEU_ERR_LICENSE_INVALID;
             }
+            mark |= 0x01;
         } else if (0 == strcmp(buf, LICENSE_KEY_MAX_NODES)) {
             license->max_nodes_ = strtoul(val, NULL, 0);
+            mark |= 0x02;
         } else if (0 == strcmp(buf, LICENSE_KEY_MAX_NODE_TAGS)) {
             license->max_node_tags_ = strtoul(val, NULL, 0);
+            mark |= 0x04;
         } else if (0 == strcmp(buf, LICENSE_KEY_PLUGIN_FLAG)) {
             if (0 !=
                 scan_bitvec(license->plugin_flag_,
                             sizeof(license->plugin_flag_), val)) {
-                zlog_error(neuron, "fail to scan license plugin flag:`%s`",
-                           val);
+                nlog_error("fail to scan license plugin flag:`%s`", val);
                 rv = NEU_ERR_LICENSE_INVALID;
             }
+            mark |= 0x08;
+        } else if (0 == strcmp(buf, LICENSE_KEY_TOKEN)) {
+            license->token_ = strdup(val);
+            mark |= 0x10;
         }
     }
 
-    nlog_info("certification verification success");
+    if (0x0F == (mark & 0x0F)) {
+        nlog_info("certification verification success");
+    } else {
+        rv = NEU_ERR_LICENSE_INVALID;
+        nlog_error("missing license key fields, mark:%u", mark);
+    }
 
 final:
     if (pkey)
@@ -330,26 +352,15 @@ const char *plugin_bit_str(plugin_bit_e b)
 
 void license_print(license_t *license)
 {
-    struct tm   tm        = {};
-    char        start[16] = {};
-    char        end[16]   = {};
-    time_t      since     = license->since_;
-    time_t      until     = license->until_;
-    const char *fmt       = "%Y%m%dT%H%M%S";
-
-    gmtime_r(&since, &tm);
-    strftime(start, sizeof(start), fmt, &tm);
-    gmtime_r(&until, &tm);
-    strftime(end, sizeof(end), fmt, &tm);
-
     char flag[sizeof(license->plugin_flag_) * 2 + 1] = {};
     print_bitvec(flag, sizeof(flag), license->plugin_flag_,
                  sizeof(license->plugin_flag_));
 
     nlog_info("license `%s` for %s, from %s to %s, max_nodes:%" PRIu32
               ", max_node_tags:%" PRIu32 ", flag:0x%s",
-              license->fname_, license_type_str(license->type_), start, end,
-              license->max_nodes_, license->max_node_tags_, flag);
+              license->fname_, license_type_str(license->type_),
+              license->since_str_, license->until_str_, license->max_nodes_,
+              license->max_node_tags_, flag);
 }
 
 void license_init(license_t *license)
@@ -363,7 +374,10 @@ void license_init(license_t *license)
 
 void license_fini(license_t *license)
 {
+    free(license->since_str_);
+    free(license->until_str_);
     free(license->fname_);
+    free(license->token_);
 }
 
 license_t *license_new()
@@ -410,7 +424,10 @@ int license_read(license_t *license, const char *fname)
 
     license_print(&lic);
 
+    free(license->since_str_);
+    free(license->until_str_);
     free(license->fname_);
+    free(license->token_);
     *license = lic;
 
     return rv;

plugins/restful/license.h

@@ -50,6 +50,8 @@
 //
 #define LICENSE_KEY_PLUGIN_FLAG "1.3.6.1.4.1.52509.13"
 
+#define LICENSE_KEY_TOKEN "1.3.6.1.4.1.52509.14"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -91,10 +93,13 @@ typedef struct {
     license_type_e type_;                 // license type
     char *         fname_;                // license file name
     uint64_t       since_;                // not before
+    char *         since_str_;            //
     uint64_t       until_;                // not after
+    char *         until_str_;            //
     uint32_t       max_nodes_;            // max number of nodes
     uint32_t       max_node_tags_;        // max number of tags per node
     uint8_t        plugin_flag_[200 / 8]; // plugin enable/disable bits
+    char *         token_;                // hardware token
 } license_t;
 
 // Initialize to be expired with no file.

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,23 +44,103 @@ 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;
-    char *                      license_path    = NULL;
-    license_t                   lic             = {};
-    neu_json_get_license_resp_t resp            = {};
-    time_t                      since           = 0;
-    time_t                      until           = 0;
-    char                        valid_since[20] = {};
-    char                        valid_until[20] = {};
-    struct tm                   tm              = {};
-    UT_array *                  plugin_names    = NULL;
-    char *                      result          = NULL;
+    int                         rv           = 0;
+    char *                      license_path = NULL;
+    license_t                   lic          = {};
+    neu_json_get_license_resp_t resp         = {};
+    UT_array *                  plugin_names = NULL;
+    char *                      result       = NULL;
 
     VALIDATE_JWT(aio);
 
@@ -77,26 +158,10 @@ void handle_get_license(nng_aio *aio)
         goto final;
     }
 
-    since = license_not_before(&lic);
-    gmtime_r(&since, &tm);
-    if (0 ==
-        strftime(valid_since, sizeof(valid_since), "%Y-%m-%d %H:%M:%S", &tm)) {
-        rv = NEU_ERR_EINTERNAL;
-        goto final;
-    }
-
-    until = license_not_after(&lic);
-    gmtime_r(&until, &tm);
-    if (0 ==
-        strftime(valid_until, sizeof(valid_until), "%Y-%m-%d %H:%M:%S", &tm)) {
-        rv = NEU_ERR_EINTERNAL;
-        goto final;
-    }
-
     resp.license_type  = (char *) license_type_str(license_get_type(&lic));
     resp.valid         = !license_is_expired(&lic);
-    resp.valid_since   = valid_since;
-    resp.valid_until   = valid_until;
+    resp.valid_since   = lic.since_str_;
+    resp.valid_until   = lic.until_str_;
     resp.max_nodes     = license_get_max_nodes(&lic);
     resp.max_node_tags = license_get_max_node_tags(&lic);
 

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

@@ -61,7 +61,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
     neu_event_io_param_t    param       = { 0 };
     neu_event_timer_param_t timer_param = {
         .second      = 0,
-        .millisecond = 100,
+        .millisecond = 10,
         .cb          = update_timestamp,
     };
 
@@ -81,8 +81,6 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
     adapter->cb_funs.command  = callback_funs.command;
     adapter->cb_funs.response = callback_funs.response;
     adapter->module           = info->module;
-    adapter->persister        = neu_persister_create("persistence");
-    assert(adapter->persister != NULL);
 
     rv = nng_pair1_open(&adapter->sock);
     assert(rv == 0);
@@ -110,8 +108,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
     strcpy(common->name, adapter->name);
 
     adapter->module->intf_funs->init(adapter->plugin);
-    if (adapter_load_setting(adapter->persister, adapter->name,
-                             &adapter->setting) == 0) {
+    if (adapter_load_setting(adapter->name, &adapter->setting) == 0) {
         if (adapter->module->intf_funs->setting(adapter->plugin,
                                                 adapter->setting) == 0) {
             adapter->state = NEU_NODE_RUNNING_STATE_READY;
@@ -122,8 +119,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
     }
 
     if (info->module->type == NEU_NA_TYPE_DRIVER) {
-        adapter_load_group_and_tag(adapter->persister,
-                                   (neu_adapter_driver_t *) adapter);
+        adapter_load_group_and_tag((neu_adapter_driver_t *) adapter);
     }
 
     param.fd       = adapter->recv_fd;
@@ -139,7 +135,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
 
     nlog_info("Success to create adapter: %s", adapter->name);
 
-    adapter_storage_state(adapter->persister, adapter->name, adapter->state);
+    adapter_storage_state(adapter->name, adapter->state);
     return adapter;
 }
 
@@ -334,8 +330,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
 
         error.error = neu_adapter_set_setting(adapter, cmd->setting);
         if (error.error == NEU_ERR_SUCCESS) {
-            adapter_storage_setting(adapter->persister, adapter->name,
-                                    cmd->setting);
+            adapter_storage_setting(adapter->name, cmd->setting);
         }
         free(cmd->setting);
 
@@ -429,8 +424,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
         }
 
         if (error.error == NEU_ERR_SUCCESS) {
-            adapter_storage_add_group(adapter->persister, adapter->name,
-                                      cmd->group, cmd->interval);
+            adapter_storage_add_group(adapter->name, cmd->group, cmd->interval);
         }
 
         neu_msg_exchange(header);
@@ -455,8 +449,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
         }
 
         if (error.error == NEU_ERR_SUCCESS) {
-            adapter_storage_update_group(adapter->persister, adapter->name,
-                                         cmd->group, cmd->interval);
+            adapter_storage_update_group(adapter->name, cmd->group,
+                                         cmd->interval);
         }
 
         neu_msg_exchange(header);
@@ -476,8 +470,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
         }
 
         if (error.error == NEU_ERR_SUCCESS) {
-            adapter_storage_del_group(adapter->persister, cmd->driver,
-                                      cmd->group);
+            adapter_storage_del_group(cmd->driver, cmd->group);
         }
 
         neu_msg_exchange(header);
@@ -514,8 +507,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
                 int ret = neu_adapter_driver_del_tag(
                     (neu_adapter_driver_t *) adapter, cmd->group, cmd->tags[i]);
                 if (0 == ret) {
-                    adapter_storage_del_tag(adapter->persister, cmd->driver,
-                                            cmd->group, cmd->tags[i]);
+                    adapter_storage_del_tag(cmd->driver, cmd->group,
+                                            cmd->tags[i]);
                 }
             }
         }
@@ -542,16 +535,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(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);
@@ -576,8 +573,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
                     (neu_adapter_driver_t *) adapter, cmd->group,
                     &cmd->tags[i]);
                 if (ret == 0) {
-                    adapter_storage_update_tag(adapter->persister, cmd->driver,
-                                               cmd->group, &cmd->tags[i]);
+                    adapter_storage_update_tag(cmd->driver, cmd->group,
+                                               &cmd->tags[i]);
 
                     resp.index += 1;
                 } else {
@@ -637,7 +634,6 @@ void neu_adapter_destroy(neu_adapter_t *adapter)
 
     nng_dialer_close(adapter->dialer);
     nng_close(adapter->sock);
-    neu_persister_destroy(adapter->persister);
 
     if (adapter->name != NULL) {
         free(adapter->name);
@@ -693,8 +689,7 @@ int neu_adapter_start(neu_adapter_t *adapter)
     error = intf_funs->start(adapter->plugin);
     if (error == NEU_ERR_SUCCESS) {
         adapter->state = NEU_NODE_RUNNING_STATE_RUNNING;
-        adapter_storage_state(adapter->persister, adapter->name,
-                              adapter->state);
+        adapter_storage_state(adapter->name, adapter->state);
     }
 
     return error;
@@ -732,8 +727,7 @@ int neu_adapter_stop(neu_adapter_t *adapter)
     error = intf_funs->stop(adapter->plugin);
     if (error == NEU_ERR_SUCCESS) {
         adapter->state = NEU_NODE_RUNNING_STATE_STOPPED;
-        adapter_storage_state(adapter->persister, adapter->name,
-                              adapter->state);
+        adapter_storage_state(adapter->name, adapter->state);
     }
 
     return error;
@@ -821,7 +815,7 @@ static int update_timestamp(void *usr_data)
     uint64_t       timestamp = 0;
 
     gettimeofday(&tv, NULL);
-    timestamp          = tv.tv_sec * 1000 + tv.tv_usec / 1000;
+    timestamp = (int64_t) tv.tv_sec * 1000 + (int64_t) tv.tv_usec / 1000;
     adapter->timestamp = timestamp;
 
     neu_plugin_to_plugin_common(adapter->plugin)->timestamp = timestamp;

src/adapter/adapter_internal.h

@@ -38,7 +38,6 @@ struct neu_adapter {
 
     neu_node_running_state_e state;
 
-    neu_persister_t *   persister;
     adapter_callbacks_t cb_funs;
 
     void *               handle;

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:
@@ -497,15 +502,15 @@ int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
     int      ret  = NEU_ERR_SUCCESS;
     group_t *find = NULL;
 
-    if (strlen(tag->name) > NEU_TAG_NAME_LEN) {
+    if (strlen(tag->name) >= NEU_TAG_NAME_LEN) {
         return NEU_ERR_TAG_NAME_TOO_LONG;
     }
 
-    if (strlen(tag->address) > NEU_TAG_ADDRESS_LEN) {
+    if (strlen(tag->address) >= NEU_TAG_ADDRESS_LEN) {
         return NEU_ERR_TAG_ADDRESS_TOO_LONG;
     }
 
-    if (strlen(tag->description) > NEU_TAG_DESCRIPTION_LEN) {
+    if (strlen(tag->description) >= NEU_TAG_DESCRIPTION_LEN) {
         return NEU_ERR_TAG_DESCRIPTION_TOO_LONG;
     }
 
@@ -524,8 +529,7 @@ int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
     HASH_FIND_STR(driver->groups, group, find);
     if (find == NULL) {
         neu_adapter_driver_add_group(driver, group, 3000);
-        adapter_storage_add_group(driver->adapter.persister,
-                                  driver->adapter.name, group, 3000);
+        adapter_storage_add_group(driver->adapter.name, group, 3000);
     }
     HASH_FIND_STR(driver->groups, group, find);
     assert(find != NULL);
@@ -564,7 +568,7 @@ int neu_adapter_driver_update_tag(neu_adapter_driver_t *driver,
     int      ret  = NEU_ERR_SUCCESS;
     group_t *find = NULL;
 
-    if (strlen(tag->name) > NEU_TAG_NAME_LEN) {
+    if (strlen(tag->name) >= NEU_TAG_NAME_LEN) {
         return NEU_ERR_TAG_NAME_TOO_LONG;
     }
 
@@ -746,8 +750,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 +861,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 +999,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

@@ -23,84 +23,97 @@
 #include "driver/driver_internal.h"
 #include "storage.h"
 
-void adapter_storage_state(neu_persister_t *persister, const char *node,
-                           neu_node_running_state_e state)
+void adapter_storage_state(const char *node, neu_node_running_state_e state)
 {
-    neu_persister_update_node_state(persister, node, state);
+    neu_persister_update_node_state(node, state);
 }
 
-void adapter_storage_setting(neu_persister_t *persister, const char *node,
-                             const char *setting)
+void adapter_storage_setting(const char *node, const char *setting)
 {
-    neu_persister_store_node_setting(persister, node, setting);
+    neu_persister_store_node_setting(node, setting);
 }
 
-void adapter_storage_add_group(neu_persister_t *persister, const char *node,
-                               const char *group, uint32_t interval)
+void adapter_storage_add_group(const char *node, const char *group,
+                               uint32_t interval)
 {
     neu_persist_group_info_t info = {
         .name     = (char *) group,
         .interval = interval,
     };
 
-    neu_persister_store_group(persister, node, &info);
+    neu_persister_store_group(node, &info);
 }
 
-void adapter_storage_update_group(neu_persister_t *persister, const char *node,
-                                  const char *group, uint32_t interval)
+void adapter_storage_update_group(const char *node, const char *group,
+                                  uint32_t interval)
 {
     neu_persist_group_info_t info = {
         .name     = (char *) group,
         .interval = interval,
     };
 
-    int rv = neu_persister_update_group(persister, node, &info);
+    int rv = neu_persister_update_group(node, &info);
     if (0 != rv) {
         nlog_error("fail update adapter:%s group:%s, interval: %u", node, group,
                    interval);
     }
 }
 
-void adapter_storage_del_group(neu_persister_t *persister, const char *node,
-                               const char *group)
+void adapter_storage_del_group(const char *node, const char *group)
 {
-    neu_persister_delete_group(persister, node, group);
+    neu_persister_delete_group(node, 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_tag(const char *node, const char *group,
+                             const neu_datatag_t *tag)
 {
-    int rv = neu_persister_store_tag(persister, node, group, tag);
+    int rv = neu_persister_store_tag(node, group, tag);
     if (0 != rv) {
         nlog_error("fail store tag:%s adapter:%s grp:%s", tag->name, node,
                    group);
     }
 }
 
-void adapter_storage_update_tag(neu_persister_t *persister, const char *node,
-                                const char *group, const neu_datatag_t *tag)
+void adapter_storage_add_tags(const char *node, const char *group,
+                              const neu_datatag_t *tags, size_t n)
 {
-    int rv = neu_persister_update_tag(persister, node, group, tag);
+    if (0 == n) {
+        return;
+    }
+
+    if (1 == n) {
+        return adapter_storage_add_tag(node, group, &tags[0]);
+    }
+
+    int rv = neu_persister_store_tags(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(const char *node, const char *group,
+                                const neu_datatag_t *tag)
+{
+    int rv = neu_persister_update_tag(node, group, tag);
     if (0 != rv) {
         nlog_error("fail update tag:%s adapter:%s grp:%s", tag->name, node,
                    group);
     }
 }
 
-void adapter_storage_del_tag(neu_persister_t *persister, const char *node,
-                             const char *group, const char *name)
+void adapter_storage_del_tag(const char *node, const char *group,
+                             const char *name)
 {
-    int rv = neu_persister_delete_tag(persister, node, group, name);
+    int rv = neu_persister_delete_tag(node, group, name);
     if (0 != rv) {
         nlog_error("fail del tag:%s adapter:%s grp:%s", name, node, group);
     }
 }
 
-int adapter_load_setting(neu_persister_t *persister, const char *node,
-                         char **setting)
+int adapter_load_setting(const char *node, char **setting)
 {
-    int rv = neu_persister_load_node_setting(persister, node,
-                                             (const char **) setting);
+    int rv = neu_persister_load_node_setting(node, (const char **) setting);
     if (0 != rv) {
         nlog_info("load %s setting fail", node);
         return -1;
@@ -109,13 +122,12 @@ int adapter_load_setting(neu_persister_t *persister, const char *node,
     return 0;
 }
 
-int adapter_load_group_and_tag(neu_persister_t *     persister,
-                               neu_adapter_driver_t *driver)
+int adapter_load_group_and_tag(neu_adapter_driver_t *driver)
 {
     UT_array *     group_infos = NULL;
     neu_adapter_t *adapter     = (neu_adapter_t *) driver;
 
-    int rv = neu_persister_load_groups(persister, adapter->name, &group_infos);
+    int rv = neu_persister_load_groups(adapter->name, &group_infos);
     if (0 != rv) {
         nlog_warn("load %s group fail", adapter->name);
         return rv;
@@ -126,7 +138,7 @@ int adapter_load_group_and_tag(neu_persister_t *     persister,
         UT_array *tags = NULL;
         neu_adapter_driver_add_group(driver, p->name, p->interval);
 
-        rv = neu_persister_load_tags(persister, adapter->name, p->name, &tags);
+        rv = neu_persister_load_tags(adapter->name, p->name, &tags);
         if (0 != rv) {
             nlog_warn("load %s:%s tags fail", adapter->name, p->name);
             continue;

src/adapter/storage.h

@@ -24,26 +24,23 @@
 
 #include "adapter_internal.h"
 
-void adapter_storage_state(neu_persister_t *persister, const char *node,
-                           neu_node_running_state_e state);
-void adapter_storage_setting(neu_persister_t *persister, const char *node,
-                             const char *setting);
-void adapter_storage_add_group(neu_persister_t *persister, const char *node,
-                               const char *group, uint32_t interval);
-void adapter_storage_update_group(neu_persister_t *persister, const char *node,
-                                  const char *group, uint32_t interval);
-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_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,
-                             const char *group, const char *name);
+void adapter_storage_state(const char *node, neu_node_running_state_e state);
+void adapter_storage_setting(const char *node, const char *setting);
+void adapter_storage_add_group(const char *node, const char *group,
+                               uint32_t interval);
+void adapter_storage_update_group(const char *node, const char *group,
+                                  uint32_t interval);
+void adapter_storage_del_group(const char *node, const char *group);
+void adapter_storage_add_tag(const char *node, const char *group,
+                             const neu_datatag_t *tag);
+void adapter_storage_add_tags(const char *node, const char *group,
+                              const neu_datatag_t *tags, size_t n);
+void adapter_storage_update_tag(const char *node, const char *group,
+                                const neu_datatag_t *tag);
+void adapter_storage_del_tag(const char *node, const char *group,
+                             const char *name);
 
-int adapter_load_setting(neu_persister_t *persister, const char *node,
-                         char **setting);
-int adapter_load_group_and_tag(neu_persister_t *     persister,
-                               neu_adapter_driver_t *driver);
+int adapter_load_setting(const char *node, char **setting);
+int adapter_load_group_and_tag(neu_adapter_driver_t *driver);
 
 #endif

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;
     }
@@ -265,7 +256,7 @@ static void update_timestamp(neu_group_t *group)
 
     gettimeofday(&tv, NULL);
 
-    group->timestamp = tv.tv_sec * 1000 * 1000 + tv.tv_usec;
+    group->timestamp = (int64_t) tv.tv_sec * 1000 * 1000 + (int64_t) tv.tv_usec;
 }
 
 static UT_array *to_array(tag_elem_t *tags)

src/base/neu_plugin_common.c

@@ -32,7 +32,8 @@ void neu_plugin_common_init(neu_plugin_common_t *common)
     common->magic      = NEU_PLUGIN_MAGIC_NUMBER;
     common->link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
     common->tag_size   = 0;
-    common->timestamp  = time(NULL) * 1000; // don't need much precision at init
+    common->timestamp  = time(NULL);
+    common->timestamp *= 1000; // don't need much precision at init
 }
 
 bool neu_plugin_common_check(neu_plugin_t *plugin)

src/connection/connection.c

@@ -235,6 +235,11 @@ ssize_t neu_conn_tcp_server_send(neu_conn_t *conn, int fd, uint8_t *buf,
     conn_tcp_server_listen(conn);
     ret = send(fd, buf, len, MSG_NOSIGNAL | MSG_DONTWAIT);
 
+    if (ret <= 0 && errno != EAGAIN) {
+        conn->disconnected(conn->data, fd);
+        conn_tcp_server_del_client(conn, fd);
+    }
+
     pthread_mutex_unlock(&conn->mtx);
 
     return ret;
@@ -257,6 +262,11 @@ ssize_t neu_conn_tcp_server_recv(neu_conn_t *conn, int fd, uint8_t *buf,
         ret = recv(fd, buf, len, 0);
     }
 
+    if (ret <= 0 && errno != EAGAIN) {
+        conn->disconnected(conn->data, fd);
+        conn_tcp_server_del_client(conn, fd);
+    }
+
     pthread_mutex_unlock(&conn->mtx);
 
     return ret;
@@ -350,7 +360,7 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
                    conn->fd, len, ret, strerror(errno), errno);
     }
 
-    if (errno == EPIPE || ret <= 0) {
+    if (errno == EPIPE || ret == -1) {
         conn_disconnect(conn);
     }
 
@@ -366,7 +376,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)
@@ -507,6 +519,7 @@ static void conn_tcp_server_stop(neu_conn_t *conn)
             conn->fd = 0;
         }
 
+        conn->tcp_server.n_client  = 0;
         conn->tcp_server.is_listen = false;
     }
 }
@@ -756,6 +769,7 @@ static void conn_disconnect(neu_conn_t *conn)
                 conn->tcp_server.clients[i].fd = 0;
             }
         }
+        conn->tcp_server.n_client = 0;
         break;
     case NEU_CONN_TCP_CLIENT:
     case NEU_CONN_UDP:
@@ -802,8 +816,6 @@ static void conn_tcp_server_del_client(neu_conn_t *conn, int fd)
             return;
         }
     }
-
-    assert(1 == 0);
 }
 
 void neu_conn_stream_consume(neu_conn_t *conn, void *context,

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

@@ -77,8 +77,6 @@ neu_manager_t *neu_manager_create()
     manager->node_manager      = neu_node_manager_create();
     manager->subscribe_manager = neu_subscribe_manager_create();
     manager->sub_msg_manager   = neu_sub_msg_manager_create();
-    manager->persister         = neu_persister_create("persistence");
-    assert(manager->persister != NULL);
 
     rv = nng_pair1_open_poly(&manager->socket);
     assert(rv == 0);
@@ -149,7 +147,6 @@ void neu_manager_destroy(neu_manager_t *manager)
     }
 
     neu_sub_msg_manager_destroy(manager->sub_msg_manager);
-    neu_persister_destroy(manager->persister);
     neu_subscribe_manager_destroy(manager->subscribe_manager);
     neu_node_manager_destroy(manager->node_manager);
     neu_plugin_manager_destroy(manager->plugin_manager);
@@ -311,6 +308,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/manager_internal.h

@@ -41,8 +41,6 @@ typedef struct neu_manager {
     neu_node_manager_t *  node_manager;
     neu_subscribe_mgr_t * subscribe_manager;
     neu_sub_msg_mgr_t *   sub_msg_manager;
-
-    neu_persister_t *persister;
 } neu_manager_t;
 
 int       neu_manager_add_plugin(neu_manager_t *manager, const char *library);

src/core/storage.c

@@ -26,7 +26,7 @@ void manager_strorage_plugin(neu_manager_t *manager)
 
     plugin_infos = neu_manager_get_plugins(manager);
 
-    int rv = neu_persister_store_plugins(manager->persister, plugin_infos);
+    int rv = neu_persister_store_plugins(plugin_infos);
     if (0 != rv) {
         nlog_error("failed to store plugins infos");
     }
@@ -46,7 +46,7 @@ void manager_storage_add_node(neu_manager_t *manager, const char *node)
         return;
     }
 
-    rv = neu_persister_store_node(manager->persister, &node_info);
+    rv = neu_persister_store_node(&node_info);
     if (0 != rv) {
         nlog_error("failed to store adapter info");
     }
@@ -56,14 +56,15 @@ void manager_storage_add_node(neu_manager_t *manager, const char *node)
 
 void manager_storage_del_node(neu_manager_t *manager, const char *node)
 {
-    neu_persister_delete_node(manager->persister, node);
+    (void) manager;
+    neu_persister_delete_node(node);
 }
 
 void manager_storage_subscribe(neu_manager_t *manager, const char *app,
                                const char *driver, const char *group)
 {
-    int rv = neu_persister_store_subscription(manager->persister, app, driver,
-                                              group);
+    (void) manager;
+    int rv = neu_persister_store_subscription(app, driver, group);
     if (0 != rv) {
         nlog_error("fail store subscription app:%s driver:%s group:%s", app,
                    driver, group);
@@ -73,8 +74,8 @@ void manager_storage_subscribe(neu_manager_t *manager, const char *app,
 void manager_storage_unsubscribe(neu_manager_t *manager, const char *app,
                                  const char *driver, const char *group)
 {
-    int rv = neu_persister_delete_subscription(manager->persister, app, driver,
-                                               group);
+    (void) manager;
+    int rv = neu_persister_delete_subscription(app, driver, group);
     if (0 != rv) {
         nlog_error("fail delete subscription app:%s driver:%s group:%s", app,
                    driver, group);
@@ -85,7 +86,7 @@ int manager_load_plugin(neu_manager_t *manager)
 {
     UT_array *plugin_infos = NULL;
 
-    int rv = neu_persister_load_plugins(manager->persister, &plugin_infos);
+    int rv = neu_persister_load_plugins(&plugin_infos);
     if (rv != 0) {
         return rv;
     }
@@ -108,7 +109,7 @@ int manager_load_node(neu_manager_t *manager)
     UT_array *node_infos = NULL;
     int       rv         = 0;
 
-    rv = neu_persister_load_nodes(manager->persister, &node_infos);
+    rv = neu_persister_load_nodes(&node_infos);
     if (0 != rv) {
         nlog_error("failed to load adapter infos");
         return -1;
@@ -139,8 +140,7 @@ int manager_load_subscribe(neu_manager_t *manager)
         int       rv        = 0;
         UT_array *sub_infos = NULL;
 
-        rv = neu_persister_load_subscriptions(manager->persister, node->node,
-                                              &sub_infos);
+        rv = neu_persister_load_subscriptions(node->node, &sub_infos);
         if (0 != rv) {
             nlog_warn("load %s subscribetion infos error", node->node);
         } else {

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,14 +38,18 @@ 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);
+        neu_persister_destroy();
+    }
     zlog_fini();
     exit_flag = true;
 }
 
 static int neuron_run(const neu_cli_args_t *args)
 {
-    struct rlimit rl;
+    struct rlimit rl = { 0 };
+    int           rv = 0;
 
     signal(SIGINT, sig_handler);
     signal(SIGTERM, sig_handler);
@@ -56,7 +61,12 @@ 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);
+    rv = neu_persister_create();
+    assert(rv == 0);
+
+    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

@@ -49,6 +49,10 @@
 
 #define PATH_MAX_SIZE 128
 
+static const char *plugin_file = "persistence/plugins.json";
+static const char *db_file     = "persistence/sqlite.db";
+static sqlite3 *   global_db   = NULL;
+
 static inline bool ends_with(const char *str, const char *suffix)
 {
     size_t m = strlen(str);
@@ -56,74 +60,6 @@ static inline bool ends_with(const char *str, const char *suffix)
     return m >= n && !strcmp(str + m - n, suffix);
 }
 
-typedef struct neu_persister {
-    const char *persist_dir;
-    const char *plugins_fname;
-    sqlite3 *   db;
-} neu_persister_t;
-
-const char *neu_persister_get_persist_dir(neu_persister_t *persister)
-{
-    if (persister == NULL) {
-        return NULL;
-    }
-
-    return persister->persist_dir;
-}
-
-const char *neu_persister_get_plugins_fname(neu_persister_t *persister)
-{
-    if (persister == NULL) {
-        return NULL;
-    }
-
-    return persister->plugins_fname;
-}
-
-/**
- * Escape special characters in path string,
- *    '%' -> '%%', '/' => '%-', '.' -> '%.'
- *
- * @param buf     output buffer,
- * @param size    size of output buffer
- * @param path    path string
- *
- * @return if `buf` is NULL, return number of bytes needed to store the result.
- *         otherwise, return the number of bytes written excluding the
- *         terminating null byte, `size` indicates overflow.
- */
-static int path_escape(char *buf, size_t size, const char *path)
-{
-    size_t i = 0;
-    char   c;
-
-    if (NULL != buf) {
-        for (const char *s = path; (c = *s++) && i < size; ++i) {
-            if (PATH_SEP_CHAR == c || '.' == c || '%' == c) {
-                c        = (PATH_SEP_CHAR == c) ? '-' : c;
-                buf[i++] = '%';
-                if (size == i) {
-                    break;
-                }
-            }
-            buf[i] = c;
-        }
-        if (i < size) {
-            buf[i] = '\0';
-        } else if (i > 0) {
-            buf[i - 1] = '\0';
-        }
-    } else {
-        for (const char *s = path; (c = *s++); ++i) {
-            if (PATH_SEP_CHAR == c || '.' == c || '%' == c) {
-                ++i;
-            }
-        }
-    }
-
-    return i;
-}
-
 /**
  * Concatenate a path string to another.
  *
@@ -158,66 +94,6 @@ static int path_cat(char *dst, size_t len, size_t size, const char *src)
     return i;
 }
 
-/**
- * Escape and concatenate a path string to another.
- *
- * @param dst   destination path string buffer
- * @param len   destination path string len, not greater than size
- * @param size  destination path buffer size
- * @param src   path string
- *
- * @return length of the result path string excluding the terminating NULL
- *         byte, `size` indicates overflow.
- */
-static int path_cat_escaped(char *dst, size_t len, size_t size, const char *src)
-{
-    size_t i = len;
-
-    if (0 < i && i < size && (PATH_SEP_CHAR != dst[i - 1])) {
-        dst[i++] = PATH_SEP_CHAR;
-        dst[i]   = '\0';
-    }
-
-    size_t n = path_escape(dst + i, size - i, src);
-
-    return i + n;
-}
-
-static inline int create_dir(char *dir_name)
-{
-    int rv = mkdir(dir_name, 0777);
-    if (0 != rv && EEXIST == errno) {
-        rv = 0;
-    }
-    return rv;
-}
-
-/**
- * Ensure that a file with the given name exists.
- *
- * @param name              file name
- * @param default_content   default file content if file not exists
- */
-static inline int ensure_file_exist(const char *name,
-                                    const char *default_content)
-{
-    int fd = open(name, O_CREAT | O_WRONLY, S_IRUSR | S_IWUSR);
-    if (-1 == fd) {
-        return -1;
-    }
-    struct stat statbuf;
-    if (-1 == fstat(fd, &statbuf)) {
-        return -1;
-    }
-    if (0 == statbuf.st_size) {
-        ssize_t size = strlen(default_content);
-        if (size != write(fd, default_content, size)) {
-            return -1;
-        }
-    }
-    return 0;
-}
-
 static int write_file_string(const char *fn, const char *s)
 {
     char *tmp = NULL;
@@ -554,159 +430,49 @@ static int apply_schemas(sqlite3 *db, const char *dir)
     return rv;
 }
 
-neu_persister_t *neu_persister_create(const char *dir_name)
+int neu_persister_create()
 {
-    int   rv            = 0;
-    char *persist_dir   = NULL;
-    char *plugins_fname = NULL;
-    char *path          = NULL;
+    int rv = sqlite3_open(db_file, &global_db);
 
-    path = calloc(PATH_MAX_SIZE, sizeof(char));
-    if (NULL == path) {
-        nlog_error("malloc fail");
-        goto error;
-    }
-
-    int dir_len = path_cat(path, 0, PATH_MAX_SIZE, dir_name);
-    if (PATH_MAX_SIZE <= dir_len) {
-        nlog_error("path too long: %s", dir_name);
-        goto error;
-    }
-    persist_dir = strdup(dir_name);
-    if (NULL == persist_dir) {
-        nlog_error("fail to strdup: %s", dir_name);
-        goto error;
-    }
-
-    rv = create_dir(persist_dir);
-    if (rv != 0) {
-        nlog_error("failed to create directory: %s", persist_dir);
-        goto error;
-    }
-
-    int n = path_cat(path, dir_len, PATH_MAX_SIZE, "plugins.json");
-    if (PATH_MAX_SIZE <= n) {
-        nlog_error("path too long: %s", path);
-        goto error;
-    }
-    plugins_fname = strdup(path);
-    if (NULL == plugins_fname) {
-        nlog_error("fail to strdup: %s", path);
-        goto error;
-    }
-
-    if (0 != ensure_file_exist(plugins_fname, "{\"plugins\": []}")) {
-        nlog_error("persister failed to ensure file exist: %s", plugins_fname);
-        goto error;
-    }
-
-    neu_persister_t *persister = malloc(sizeof(neu_persister_t));
-    if (NULL == persister) {
-        nlog_error("failed to alloc memory for persister struct");
-        goto error;
-    }
-
-    persister->plugins_fname = plugins_fname;
-    persister->persist_dir   = persist_dir;
-
-    n = path_cat(path, dir_len, PATH_MAX_SIZE, "sqlite.db");
-    if (PATH_MAX_SIZE <= n) {
-        nlog_error("path too long: %s", path);
-        goto error;
-    }
-    rv = sqlite3_open(path, &persister->db);
     if (SQLITE_OK != rv) {
-        nlog_error("db `%s` fail: %s", path, sqlite3_errstr(rv));
-        goto error;
+        nlog_fatal("db `%s` fail: %s", db_file, sqlite3_errstr(rv));
+        return -1;
     }
-    sqlite3_busy_timeout(persister->db, 100 * 1000);
-    // we rely on sqlite foreign key support
-    if (SQLITE_OK !=
-        sqlite3_exec(persister->db, "PRAGMA foreign_keys=ON", NULL, NULL,
-                     NULL)) {
-        nlog_error("db foreign key support fail: %s",
-                   sqlite3_errmsg(persister->db));
-        sqlite3_close(persister->db);
-        goto error;
-    }
-    if (0 != apply_schemas(persister->db, g_config_dir)) {
-        goto error;
+    sqlite3_busy_timeout(global_db, 100 * 1000);
+
+    rv = sqlite3_exec(global_db, "PRAGMA foreign_keys=ON", NULL, NULL, NULL);
+    if (rv != SQLITE_OK) {
+        nlog_fatal("db foreign key support fail: %s",
+                   sqlite3_errmsg(global_db));
+        sqlite3_close(global_db);
+        return -1;
     }
 
-    free(path);
-    return persister;
+    rv = sqlite3_exec(global_db, "PRAGMA journal_mode=WAL", NULL, NULL, NULL);
+    if (rv != SQLITE_OK) {
+        nlog_fatal("db journal_mode WAL fail: %s", sqlite3_errmsg(global_db));
+        sqlite3_close(global_db);
+        return -1;
+    }
 
-error:
-    free(path);
-    free(plugins_fname);
-    free(persist_dir);
-    return NULL;
+    rv = apply_schemas(global_db, g_config_dir);
+    if (rv != 0) {
+        nlog_fatal("db apply schemas fail");
+        sqlite3_close(global_db);
+        return -1;
+    }
+
+    return 0;
 }
 
-static inline int persister_adapters_dir(char *buf, size_t size,
-                                         neu_persister_t *persister)
+void neu_persister_destroy()
 {
-    size_t n = path_cat(buf, 0, size, persister->persist_dir);
-    if (size == n) {
-        return n;
-    }
-
-    n = path_cat(buf, n, size, "adapters");
-    return n;
+    sqlite3_close(global_db);
 }
 
-static inline int persister_adapter_dir(char *buf, size_t size,
-                                        neu_persister_t *persister,
-                                        const char *     adapter_name)
+int neu_persister_store_node(neu_persist_node_info_t *info)
 {
-    size_t n = persister_adapters_dir(buf, size, persister);
-    if (size == n) {
-        return n;
-    }
-
-    n = path_cat_escaped(buf, n, size, adapter_name);
-    return n;
-}
-
-static inline int persister_group_configs_dir(char *buf, size_t size,
-                                              neu_persister_t *persister,
-                                              const char *     adapter_name)
-{
-    size_t n = persister_adapter_dir(buf, size, persister, adapter_name);
-    if (size == n) {
-        return n;
-    }
-
-    n = path_cat(buf, n, size, "groups");
-    return n;
-}
-
-static inline int persister_group_config_dir(char *buf, size_t size,
-                                             neu_persister_t *persister,
-                                             const char *     adapter_name,
-                                             const char *     group_config_name)
-{
-    size_t n = persister_group_configs_dir(buf, size, persister, adapter_name);
-    if (size == n) {
-        return n;
-    }
-
-    n = path_cat_escaped(buf, n, size, group_config_name);
-    return n;
-}
-
-void neu_persister_destroy(neu_persister_t *persister)
-{
-    sqlite3_close(persister->db);
-    free((char *) persister->plugins_fname);
-    free((char *) persister->persist_dir);
-    free(persister);
-}
-
-int neu_persister_store_node(neu_persister_t *        persister,
-                             neu_persist_node_info_t *info)
-{
-    return execute_sql(persister->db,
+    return execute_sql(global_db,
                        "INSERT INTO nodes (name, type, state, plugin_name) "
                        "VALUES (%Q, %i, %i, %Q)",
                        info->name, info->type, info->state, info->plugin_name);
@@ -719,7 +485,7 @@ static UT_icd node_info_icd = {
     (dtor_f *) neu_persist_node_info_fini,
 };
 
-int neu_persister_load_nodes(neu_persister_t *persister, UT_array **node_infos)
+int neu_persister_load_nodes(UT_array **node_infos)
 {
     int           rv    = 0;
     sqlite3_stmt *stmt  = NULL;
@@ -727,10 +493,8 @@ int neu_persister_load_nodes(neu_persister_t *persister, UT_array **node_infos)
 
     utarray_new(*node_infos, &node_info_icd);
 
-    if (SQLITE_OK !=
-        sqlite3_prepare_v2(persister->db, query, -1, &stmt, NULL)) {
-        nlog_error("prepare `%s` fail: %s", query,
-                   sqlite3_errmsg(persister->db));
+    if (SQLITE_OK != sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL)) {
+        nlog_error("prepare `%s` fail: %s", query, sqlite3_errmsg(global_db));
         utarray_free(*node_infos);
         *node_infos = NULL;
         return NEU_ERR_EINTERNAL;
@@ -760,7 +524,7 @@ int neu_persister_load_nodes(neu_persister_t *persister, UT_array **node_infos)
     }
 
     if (SQLITE_DONE != step) {
-        nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(persister->db));
+        nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(global_db));
         // do not set return code, return partial or empty result
     }
 
@@ -768,31 +532,27 @@ int neu_persister_load_nodes(neu_persister_t *persister, UT_array **node_infos)
     return rv;
 }
 
-int neu_persister_delete_node(neu_persister_t *persister, const char *node_name)
+int neu_persister_delete_node(const char *node_name)
 {
     // rely on foreign key constraints to remove settings, groups, tags and
     // subscriptions
-    return execute_sql(persister->db, "DELETE FROM nodes WHERE name=%Q;",
+    return execute_sql(global_db, "DELETE FROM nodes WHERE name=%Q;",
                        node_name);
 }
 
-int neu_persister_update_node(neu_persister_t *persister, const char *node_name,
-                              const char *new_name)
+int neu_persister_update_node(const char *node_name, const char *new_name)
 {
-    return execute_sql(persister->db, "UPDATE nodes SET name=%Q WHERE name=%Q;",
+    return execute_sql(global_db, "UPDATE nodes SET name=%Q WHERE name=%Q;",
                        new_name, node_name);
 }
 
-int neu_persister_update_node_state(neu_persister_t *persister,
-                                    const char *node_name, int state)
+int neu_persister_update_node_state(const char *node_name, int state)
 {
-    return execute_sql(persister->db,
-                       "UPDATE nodes SET state=%i WHERE name=%Q;", state,
-                       node_name);
+    return execute_sql(global_db, "UPDATE nodes SET state=%i WHERE name=%Q;",
+                       state, node_name);
 }
 
-int neu_persister_store_plugins(neu_persister_t *persister,
-                                UT_array *       plugin_infos)
+int neu_persister_store_plugins(UT_array *plugin_infos)
 {
     int                    index       = 0;
     neu_json_plugin_resp_t plugin_resp = {
@@ -820,7 +580,7 @@ int neu_persister_store_plugins(neu_persister_t *persister,
         return rv;
     }
 
-    rv = write_file_string(persister->plugins_fname, result);
+    rv = write_file_string(plugin_file, result);
 
     free(result);
     return rv;
@@ -857,8 +617,7 @@ static int ut_str_cmp(const void *a, const void *b)
     return strcmp(*(char **) a, *(char **) b);
 }
 
-int neu_persister_load_plugins(neu_persister_t *persister,
-                               UT_array **      plugin_infos)
+int neu_persister_load_plugins(UT_array **plugin_infos)
 {
     UT_array *default_plugins = NULL;
     UT_array *user_plugins    = NULL;
@@ -871,7 +630,7 @@ int neu_persister_load_plugins(neu_persister_t *persister,
         nlog_warn("cannot load default plugins");
     }
     // user plugins
-    if (0 != load_plugins_file(persister->plugins_fname, user_plugins)) {
+    if (0 != load_plugins_file(plugin_file, user_plugins)) {
         nlog_warn("cannot load user plugins");
     } else {
         // the following operation needs sorting
@@ -895,10 +654,10 @@ int neu_persister_load_plugins(neu_persister_t *persister,
     return 0;
 }
 
-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_tag(const char *driver_name, const char *group_name,
+                            const neu_datatag_t *tag)
 {
-    return execute_sql(persister->db,
+    return execute_sql(global_db,
                        "INSERT INTO tags (driver_name, group_name, name, "
                        "address, attribute, precision, "
                        "type, decimal, description) VALUES(%Q, %Q, %Q, %Q, %i, "
@@ -908,8 +667,107 @@ int neu_persister_store_tag(neu_persister_t *persister, const char *driver_name,
                        tag->description);
 }
 
-int neu_persister_load_tags(neu_persister_t *persister, const char *driver_name,
-                            const char *group_name, UT_array **tags)
+int neu_persister_store_tags(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(global_db, "BEGIN", NULL, NULL, NULL)) {
+        nlog_error("begin transaction fail: %s", sqlite3_errmsg(global_db));
+        return NEU_ERR_EINTERNAL;
+    }
+
+    if (SQLITE_OK != sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL)) {
+        nlog_error("prepare `%s` fail: %s", query, sqlite3_errmsg(global_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(global_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(global_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(global_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(global_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(global_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(global_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(global_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(global_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(global_db));
+            goto error;
+        }
+
+        if (SQLITE_DONE != sqlite3_step(stmt)) {
+            nlog_error("sqlite3_step fail: %s", sqlite3_errmsg(global_db));
+            goto error;
+        }
+    }
+
+    if (SQLITE_OK != sqlite3_exec(global_db, "COMMIT", NULL, NULL, NULL)) {
+        nlog_error("commit transaction fail: %s", sqlite3_errmsg(global_db));
+        goto error;
+    }
+
+    sqlite3_finalize(stmt);
+    return 0;
+
+error:
+    nlog_info("rollback transaction");
+    sqlite3_exec(global_db, "ROLLBACK", NULL, NULL, NULL);
+    sqlite3_finalize(stmt);
+    return NEU_ERR_EINTERNAL;
+}
+
+int neu_persister_load_tags(const char *driver_name, const char *group_name,
+                            UT_array **tags)
 {
     sqlite3_stmt *stmt  = NULL;
     const char *  query = "SELECT name, address, attribute, precision, type, "
@@ -918,22 +776,20 @@ int neu_persister_load_tags(neu_persister_t *persister, const char *driver_name,
 
     utarray_new(*tags, neu_tag_get_icd());
 
-    if (SQLITE_OK !=
-        sqlite3_prepare_v2(persister->db, query, -1, &stmt, NULL)) {
-        nlog_error("prepare `%s` fail: %s", query,
-                   sqlite3_errmsg(persister->db));
+    if (SQLITE_OK != sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL)) {
+        nlog_error("prepare `%s` fail: %s", query, sqlite3_errmsg(global_db));
         goto error;
     }
 
     if (SQLITE_OK != sqlite3_bind_text(stmt, 1, driver_name, -1, NULL)) {
         nlog_error("bind `%s` with `%s` fail: %s", query, driver_name,
-                   sqlite3_errmsg(persister->db));
+                   sqlite3_errmsg(global_db));
         goto error;
     }
 
     if (SQLITE_OK != sqlite3_bind_text(stmt, 2, group_name, -1, NULL)) {
         nlog_error("bind `%s` with `%s` fail: %s", query, group_name,
-                   sqlite3_errmsg(persister->db));
+                   sqlite3_errmsg(global_db));
         goto error;
     }
 
@@ -954,7 +810,7 @@ int neu_persister_load_tags(neu_persister_t *persister, const char *driver_name,
     }
 
     if (SQLITE_DONE != step) {
-        nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(persister->db));
+        nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(global_db));
         // do not set return code, return partial or empty result
     }
 
@@ -967,11 +823,10 @@ error:
     return NEU_ERR_EINTERNAL;
 }
 
-int neu_persister_update_tag(neu_persister_t *persister,
-                             const char *driver_name, const char *group_name,
+int neu_persister_update_tag(const char *driver_name, const char *group_name,
                              const neu_datatag_t *tag)
 {
-    return execute_sql(persister->db,
+    return execute_sql(global_db,
                        "UPDATE tags SET address=%Q, attribute=%i, "
                        "precision=%i, type=%i, decimal=%lf, description=%Q "
                        "WHERE driver_name=%Q AND group_name=%Q AND name=%Q",
@@ -980,23 +835,21 @@ int neu_persister_update_tag(neu_persister_t *persister,
                        tag->name);
 }
 
-int neu_persister_delete_tag(neu_persister_t *persister,
-                             const char *driver_name, const char *group_name,
+int neu_persister_delete_tag(const char *driver_name, const char *group_name,
                              const char *tag_name)
 {
     return execute_sql(
-        persister->db,
+        global_db,
         "DELETE FROM tags WHERE driver_name=%Q AND group_name=%Q AND name=%Q",
         driver_name, group_name, tag_name);
 }
 
-int neu_persister_store_subscription(neu_persister_t *persister,
-                                     const char *     app_name,
-                                     const char *     driver_name,
-                                     const char *     group_name)
+int neu_persister_store_subscription(const char *app_name,
+                                     const char *driver_name,
+                                     const char *group_name)
 {
     return execute_sql(
-        persister->db,
+        global_db,
         "INSERT INTO subscriptions (app_name, driver_name, group_name) "
         "VALUES (%Q, %Q, %Q)",
         app_name, driver_name, group_name);
@@ -1009,9 +862,8 @@ static UT_icd subscription_info_icd = {
     (dtor_f *) neu_persist_subscription_info_fini,
 };
 
-int neu_persister_load_subscriptions(neu_persister_t *persister,
-                                     const char *     app_name,
-                                     UT_array **      subscription_infos)
+int neu_persister_load_subscriptions(const char *app_name,
+                                     UT_array ** subscription_infos)
 {
     sqlite3_stmt *stmt = NULL;
     const char *  query =
@@ -1019,16 +871,14 @@ int neu_persister_load_subscriptions(neu_persister_t *persister,
 
     utarray_new(*subscription_infos, &subscription_info_icd);
 
-    if (SQLITE_OK !=
-        sqlite3_prepare_v2(persister->db, query, -1, &stmt, NULL)) {
-        nlog_error("prepare `%s` fail: %s", query,
-                   sqlite3_errmsg(persister->db));
+    if (SQLITE_OK != sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL)) {
+        nlog_error("prepare `%s` fail: %s", query, sqlite3_errmsg(global_db));
         goto error;
     }
 
     if (SQLITE_OK != sqlite3_bind_text(stmt, 1, app_name, -1, NULL)) {
         nlog_error("bind `%s` with `%s` fail: %s", query, app_name,
-                   sqlite3_errmsg(persister->db));
+                   sqlite3_errmsg(global_db));
         goto error;
     }
 
@@ -1055,7 +905,7 @@ int neu_persister_load_subscriptions(neu_persister_t *persister,
     }
 
     if (SQLITE_DONE != step) {
-        nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(persister->db));
+        nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(global_db));
         // do not set return code, return partial or empty result
     }
 
@@ -1068,34 +918,31 @@ error:
     return NEU_ERR_EINTERNAL;
 }
 
-int neu_persister_delete_subscription(neu_persister_t *persister,
-                                      const char *     app_name,
-                                      const char *     driver_name,
-                                      const char *     group_name)
+int neu_persister_delete_subscription(const char *app_name,
+                                      const char *driver_name,
+                                      const char *group_name)
 {
-    return execute_sql(persister->db,
+    return execute_sql(global_db,
                        "DELETE FROM subscriptions WHERE app_name=%Q AND "
                        "driver_name=%Q AND group_name=%Q",
                        app_name, driver_name, group_name);
 }
 
-int neu_persister_store_group(neu_persister_t *         persister,
-                              const char *              driver_name,
+int neu_persister_store_group(const char *              driver_name,
                               neu_persist_group_info_t *group_info)
 {
     return execute_sql(
-        persister->db,
+        global_db,
         "INSERT INTO groups (driver_name, name, interval) VALUES (%Q, %Q, %u)",
         driver_name, group_info->name, (unsigned) group_info->interval);
 }
 
-int neu_persister_update_group(neu_persister_t *         persister,
-                               const char *              driver_name,
+int neu_persister_update_group(const char *              driver_name,
                                neu_persist_group_info_t *group_info)
 {
-    return execute_sql(persister->db,
+    return execute_sql(global_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);
 }
@@ -1107,24 +954,21 @@ static UT_icd group_info_icd = {
     (dtor_f *) neu_persist_group_info_fini,
 };
 
-int neu_persister_load_groups(neu_persister_t *persister,
-                              const char *driver_name, UT_array **group_infos)
+int neu_persister_load_groups(const char *driver_name, UT_array **group_infos)
 {
     sqlite3_stmt *stmt = NULL;
     const char *query = "SELECT name, interval FROM groups WHERE driver_name=?";
 
     utarray_new(*group_infos, &group_info_icd);
 
-    if (SQLITE_OK !=
-        sqlite3_prepare_v2(persister->db, query, -1, &stmt, NULL)) {
-        nlog_error("prepare `%s` fail: %s", query,
-                   sqlite3_errmsg(persister->db));
+    if (SQLITE_OK != sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL)) {
+        nlog_error("prepare `%s` fail: %s", query, sqlite3_errmsg(global_db));
         goto error;
     }
 
     if (SQLITE_OK != sqlite3_bind_text(stmt, 1, driver_name, -1, NULL)) {
         nlog_error("bind `%s` with `%s` fail: %s", query, driver_name,
-                   sqlite3_errmsg(persister->db));
+                   sqlite3_errmsg(global_db));
         goto error;
     }
 
@@ -1144,7 +988,7 @@ int neu_persister_load_groups(neu_persister_t *persister,
     }
 
     if (SQLITE_DONE != step) {
-        nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(persister->db));
+        nlog_warn("query `%s` fail: %s", query, sqlite3_errmsg(global_db));
         // do not set return code, return partial or empty result
     }
 
@@ -1157,49 +1001,45 @@ error:
     return NEU_ERR_EINTERNAL;
 }
 
-int neu_persister_delete_group(neu_persister_t *persister,
-                               const char *driver_name, const char *group_name)
+int neu_persister_delete_group(const char *driver_name, const char *group_name)
 {
     // rely on foreign key constraints to delete tags and subscriptions
-    return execute_sql(persister->db,
+    return execute_sql(global_db,
                        "DELETE FROM groups WHERE driver_name=%Q AND name=%Q",
                        driver_name, group_name);
 }
 
-int neu_persister_store_node_setting(neu_persister_t *persister,
-                                     const char *node_name, const char *setting)
+int neu_persister_store_node_setting(const char *node_name, const char *setting)
 {
     return execute_sql(
-        persister->db,
+        global_db,
         "INSERT OR REPLACE INTO settings (node_name, setting) VALUES (%Q, %Q)",
         node_name, setting);
 }
 
-int neu_persister_load_node_setting(neu_persister_t *  persister,
-                                    const char *       node_name,
+int neu_persister_load_node_setting(const char *       node_name,
                                     const char **const setting)
 {
     int           rv    = 0;
     sqlite3_stmt *stmt  = NULL;
     const char *  query = "SELECT setting FROM settings WHERE node_name=?";
 
-    if (SQLITE_OK !=
-        sqlite3_prepare_v2(persister->db, query, -1, &stmt, NULL)) {
+    if (SQLITE_OK != sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL)) {
         nlog_error("prepare `%s` with `%s` fail: %s", query, node_name,
-                   sqlite3_errmsg(persister->db));
+                   sqlite3_errmsg(global_db));
         return NEU_ERR_EINTERNAL;
     }
 
     if (SQLITE_OK != sqlite3_bind_text(stmt, 1, node_name, -1, NULL)) {
         nlog_error("bind `%s` with `%s` fail: %s", query, node_name,
-                   sqlite3_errmsg(persister->db));
+                   sqlite3_errmsg(global_db));
         rv = NEU_ERR_EINTERNAL;
         goto end;
     }
 
     if (SQLITE_ROW != sqlite3_step(stmt)) {
         nlog_warn("SQL `%s` with `%s` fail: %s", query, node_name,
-                  sqlite3_errmsg(persister->db));
+                  sqlite3_errmsg(global_db));
         rv = NEU_ERR_EINTERNAL;
         goto end;
     }
@@ -1218,9 +1058,8 @@ end:
     return rv;
 }
 
-int neu_persister_delete_node_setting(neu_persister_t *persister,
-                                      const char *     node_name)
+int neu_persister_delete_node_setting(const char *node_name)
 {
-    return execute_sql(persister->db, "DELETE FROM settings WHERE node_name=%Q",
+    return execute_sql(global_db, "DELETE FROM settings WHERE node_name=%Q",
                        node_name);
 }

src/persist/persist.h

@@ -69,241 +69,197 @@ neu_persist_subscription_info_fini(neu_persist_subscription_info_t *info)
     free(info->group_name);
 }
 
-/**
- * Persister, provide methods to persist data */
-typedef struct neu_persister neu_persister_t;
-
-const char *neu_persister_get_persist_dir(neu_persister_t *persister);
-const char *neu_persister_get_plugins_fname(neu_persister_t *persister);
-
 /**
  * Create persister.
- * @return Pointer to heap allocated neu_persister_t on success, NULL otherwise.
+ * @return 0 on success, -1 otherwise.
  */
-neu_persister_t *neu_persister_create(const char *dir_name);
+int neu_persister_create();
 /**
  * Destroy perister.
- * @param persister                 persiter object to destroy.
  */
-void neu_persister_destroy(neu_persister_t *persiter);
+void neu_persister_destroy();
 
 /**
  * Persist nodes.
- * @param persister                 persiter object.
  * @param node_info                 neu_persist_node_info_t.
  * @return 0 on success, non-zero on failure
  */
-int neu_persister_store_node(neu_persister_t *        persister,
-                             neu_persist_node_info_t *info);
+int neu_persister_store_node(neu_persist_node_info_t *info);
 /**
  * Load node infos.
- * @param persister                 persiter object.
  * @param[out] node_infos           used to return pointer to heap allocated
  *                                  vector of neu_persist_node_info_t.
  * @return 0 on success, none-zero on failure
  */
-int neu_persister_load_nodes(neu_persister_t *persister, UT_array **node_infos);
+int neu_persister_load_nodes(UT_array **node_infos);
 /**
  * Delete node.
- * @param persister                 persiter object.
  * @param node_name                 name of the node to delete.
  * @return 0 on success, none-zero on failure
  */
-int neu_persister_delete_node(neu_persister_t *persister,
-                              const char *     node_name);
+int neu_persister_delete_node(const char *node_name);
 
 /**
  * Update node name.
- * @param persister                 persiter object.
  * @param node_name                 name of the node to update.
  * @param new_name                  new name of the adapter.
  * @return 0 on success, none-zero on failure
  */
-int neu_persister_update_node(neu_persister_t *persister, const char *node_name,
-                              const char *new_name);
+int neu_persister_update_node(const char *node_name, const char *new_name);
 
 /**
  * Update node state.
- * @param persister                 persiter object.
  * @param node_name                 name of the node to update.
  * @param state                     state of the adapter.
  * @return 0 on success, none-zero on failure
  */
-int neu_persister_update_node_state(neu_persister_t *persister,
-                                    const char *node_name, int state);
+int neu_persister_update_node_state(const char *node_name, int state);
 
 /**
  * Persist plugins.
- * @param persister                 persiter object.
  * @param plugin_infos              vector of neu_persist_plugin_info_t.
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_store_plugins(neu_persister_t *persister,
-                                UT_array *       plugin_infos);
+int neu_persister_store_plugins(UT_array *plugin_infos);
 /**
  * Load plugin infos.
- * @param persister                 persiter object.
  * @param[out] plugin_infos         used to return pointer to heap allocated
  * vector of neu_persist_plugin_info_t.
  * @return 0 on success, none-zero on failure
  */
-int neu_persister_load_plugins(neu_persister_t *persister,
-                               UT_array **      plugin_infos);
+int neu_persister_load_plugins(UT_array **plugin_infos);
 
 /**
  * Persist node tags.
- * @param persister                 persiter object.
  * @param driver_name               name of the driver who owns the tags
  * @param group_name                name of the group
  * @param tag                       the tag to store
  * @return 0 on success, non-zero otherwise
  */
-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_tag(const char *driver_name, const char *group_name,
+                            const neu_datatag_t *tag);
+
+int neu_persister_store_tags(const char *driver_name, const char *group_name,
+                             const neu_datatag_t *tags, size_t n);
 
 /**
  * Load node tag infos.
- * @param persister                 persiter object.
  * @param node_name                 name of the node who owns the tags
  * @param group_name                name of the group
  * @param[out] tag_infos            used to return pointer to heap allocated
  *                                  vector of neu_datatag_t
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_load_tags(neu_persister_t *persister, const char *driver_name,
-                            const char *group_name, UT_array **tag_infos);
+int neu_persister_load_tags(const char *driver_name, const char *group_name,
+                            UT_array **tag_infos);
 
 /**
  * Update node tags.
- * @param persister                 persiter object.
  * @param driver_name               name of the driver who owns the tags
  * @param group_name                name of the group
  * @param tag                       the tag to update
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_update_tag(neu_persister_t *persister,
-                             const char *driver_name, const char *group_name,
+int neu_persister_update_tag(const char *driver_name, const char *group_name,
                              const neu_datatag_t *tag);
 
 /**
  * Delete node tags.
- * @param persister                 persiter object.
  * @param driver_name               name of the driver who owns the tags
  * @param group_name                name of the group
  * @param tag_name                  name of the tag
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_delete_tag(neu_persister_t *persister,
-                             const char *driver_name, const char *group_name,
+int neu_persister_delete_tag(const char *driver_name, const char *group_name,
                              const char *tag_name);
 
 /**
  * Persist subscriptions.
- * @param persister                 persiter object.
  * @param app_name                  name of the app node
  * @param driver_name               name of the driver node
  * @param group_name                name of the group
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_store_subscription(neu_persister_t *persister,
-                                     const char *     app_name,
-                                     const char *     driver_name,
-                                     const char *     group_name);
+int neu_persister_store_subscription(const char *app_name,
+                                     const char *driver_name,
+                                     const char *group_name);
 
 /**
  * Load adapter subscriptions.
- * @param persister                 persiter object.
  * @param app_name                  name of the app node
  * @param[out] subscription_infos   used to return pointer to heap allocated
  *                                  vector of neu_persist_subscription_info_t.
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_load_subscriptions(neu_persister_t *persister,
-                                     const char *     app_name,
-                                     UT_array **      subscription_infos);
+int neu_persister_load_subscriptions(const char *app_name,
+                                     UT_array ** subscription_infos);
 
 /**
  * Persist subscriptions.
- * @param persister                 persiter object.
  * @param app_name                  name of the app node
  * @param driver_name               name of the driver node
  * @param group_name                name of the group
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_delete_subscription(neu_persister_t *persister,
-                                      const char *     app_name,
-                                      const char *     driver_name,
-                                      const char *     group_name);
+int neu_persister_delete_subscription(const char *app_name,
+                                      const char *driver_name,
+                                      const char *group_name);
 
 /**
  * Persist group config.
- * @param persister                 persiter object.
  * @param driver_name               name of the driver who owns the group
  * @param group_info                group info to persist.
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_store_group(neu_persister_t *         persister,
-                              const char *              driver_name,
+int neu_persister_store_group(const char *              driver_name,
                               neu_persist_group_info_t *group_info);
 
 /**
  * Update group config.
- * @param persister                 persiter object.
  * @param driver_name               name of the driver who owns the group
  * @param group_info                group info to persist.
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_update_group(neu_persister_t *         persister,
-                               const char *              driver_name,
+int neu_persister_update_group(const char *              driver_name,
                                neu_persist_group_info_t *group_info);
 /**
  * Load all group config infos under an adapter.
- * @param persister                 persiter object.
  * @param driver_name               name of the driver who owns the group
  * @param[out] group_infos          used to return pointer to heap allocated
  *                                  vector of neu_persist_group_info_t.
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_load_groups(neu_persister_t *persister,
-                              const char *driver_name, UT_array **group_infos);
+int neu_persister_load_groups(const char *driver_name, UT_array **group_infos);
 /**
  * Delete group config.
- * @param persister                 persiter object.
  * @param driver_name               name of the driver who owns the group
  * @param group_name                name of the group.
  * @return 0 on success, none-zero on failure
  */
-int neu_persister_delete_group(neu_persister_t *persister,
-                               const char *driver_name, const char *group_name);
+int neu_persister_delete_group(const char *driver_name, const char *group_name);
 
 /**
  * Persist node setting.
- * @param persister                 persiter object.
  * @param adapter_name              name of the adapter who owns the setting.
  * @param setting                   node setting string.
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_store_node_setting(neu_persister_t *persister,
-                                     const char *     node_name,
-                                     const char *     setting);
+int neu_persister_store_node_setting(const char *node_name,
+                                     const char *setting);
 /**
  * Load node setting.
- * @param persister                 persiter object.
  * @param node_name                 name of the node.
  * @param[out] setting              used to return node setting string.
  * @return 0 on success, non-zero otherwise
  */
-int neu_persister_load_node_setting(neu_persister_t *  persister,
-                                    const char *       node_name,
+int neu_persister_load_node_setting(const char *       node_name,
                                     const char **const setting);
 /**
  * Delete node setting.
- * @param persister                 persiter object.
  * @param node_name                 name of the node.
  * @return 0 on success, none-zero on failure
  */
-int neu_persister_delete_node_setting(neu_persister_t *persister,
-                                      const char *     node_name);
+int neu_persister_delete_node_setting(const char *node_name);
 
 #ifdef __cplusplus
 }

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