Merge pull request #1042 from fengzeroz/v2.2

version: 2.2.13

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

include/neuron/define.h

@@ -20,12 +20,12 @@
 #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 64
+#define NEU_TAG_ADDRESS_LEN 128
 #define NEU_TAG_DESCRIPTION_LEN 128
-#define NEU_GROUP_NAME_LEN 32
+#define NEU_GROUP_NAME_LEN 128
 #define NEU_GROUP_INTERVAL_LIMIT 100
-#define NEU_NODE_NAME_LEN 32
+#define NEU_NODE_NAME_LEN 128
 #define NEU_PLUGIN_NAME_LEN 32
 #define NEU_PLUGIN_LIBRARY_LEN 32
 #define NEU_PLUGIN_DESCRIPTION_LEN 512

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)
 {

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

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

@@ -502,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;
     }
 
@@ -529,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);
@@ -569,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;
     }
 

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

@@ -256,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) {
+        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);
     }
 
@@ -509,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;
     }
 }
@@ -758,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:
@@ -804,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

@@ -559,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

@@ -40,6 +40,7 @@ static void sig_handler(int sig)
 
     if (sig != SIGABRT) {
         neu_manager_destroy(g_manager);
+        neu_persister_destroy();
     }
     zlog_fini();
     exit_flag = true;
@@ -47,7 +48,8 @@ static void sig_handler(int sig)
 
 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);
@@ -59,6 +61,9 @@ static int neuron_run(const neu_cli_args_t *args)
         nlog_warn("neuron process failed enable core dump, ignore");
     }
 
+    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);

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,32 +918,29 @@ 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",
                        driver_name, group_info->name, group_info->interval,
@@ -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
 }

ui_version

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

version

@@ -1 +1 @@
-2.2.2
+2.2.13