Merge pull request #799 from fengzeroz/v2.2

version: 2.2.2

ft/tag_test.robot

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

include/neuron/errcodes.h

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

include/neuron/utils/mem_cache.h

@@ -45,6 +45,8 @@ cache_item_t     neu_mem_cache_earliest(neu_mem_cache_t *cache);
 cache_item_t     neu_mem_cache_latest(neu_mem_cache_t *cache);
 void             neu_mem_cache_used(neu_mem_cache_t *cache, size_t *used_bytes,
                                     size_t *used_items);
+void             neu_mem_cache_resize(neu_mem_cache_t *cache, size_t new_bytes,
+                                      size_t new_items);
 int  neu_mem_cache_dump(neu_mem_cache_t *cache, cache_dump callback, void *ctx);
 int  neu_mem_cache_clear(neu_mem_cache_t *cache);
 void neu_mem_cache_destroy(neu_mem_cache_t *cache);

plugins/mqtt/command/read_write.c

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

plugins/mqtt/mqtt.json

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

plugins/mqtt/mqtt_plugin.c

@@ -19,9 +19,6 @@
 
 #include <assert.h>
 
-#include <nng/nng.h>
-#include <nng/supplemental/util/platform.h>
-
 #include "command/command.h"
 #include "mqtt_plugin.h"
 #include "mqtt_util.h"
@@ -53,14 +50,10 @@ struct topic_pair {
 };
 
 struct mqtt_routine {
-    neu_plugin_t *     plugin;
-    neu_mqtt_option_t  option;
-    neu_mqtt_client_t  client;
-    UT_array *         topics;
-    neu_mem_cache_t *  cache;
-    nng_mtx *          mutex;
-    neu_events_t *     events;
-    neu_event_timer_t *send;
+    neu_plugin_t *    plugin;
+    neu_mqtt_option_t option;
+    neu_mqtt_client_t client;
+    UT_array *        topics;
 };
 
 static char *topics_format(char *format, char *name)
@@ -171,46 +164,6 @@ static struct topic_pair *topics_find_type(UT_array *topics, int type)
     return NULL;
 }
 
-static int mqtt_routine_cache_send(void *data)
-{
-    mqtt_routine_t *routine = (mqtt_routine_t *) data;
-
-    // size_t used_bytes = 0;
-    // size_t used_items = 0;
-    // neu_mem_cache_used(routine->cache, &used_bytes, &used_items);
-    // plog_info(routine->plugin, "used bytes:%lu, used items:%lu", used_bytes,
-    //           used_items);
-
-    neu_err_code_e rc = neu_mqtt_client_is_connected(routine->client);
-    if (0 != rc) {
-        return -1;
-    }
-
-    nng_mtx_lock(routine->mutex);
-    cache_item_t item = neu_mem_cache_earliest(routine->cache);
-    if (NULL == item.data) {
-        nng_mtx_unlock(routine->mutex);
-        return -1;
-    }
-
-    int                type     = TOPIC_TYPE_UPLOAD;
-    struct topic_pair *pair     = topics_find_type(routine->topics, type);
-    char *             json_str = item.data;
-    const char *       topic    = pair->topic_response;
-    const int          qos      = pair->qos_response;
-    neu_err_code_e     error =
-        neu_mqtt_client_publish(routine->client, topic, qos,
-                                (unsigned char *) json_str, strlen(json_str));
-    if (NEU_ERR_SUCCESS != error) {
-        plog_error(routine->plugin, "cache publish error code :%d, topoic:%s",
-                   error, topic);
-    }
-
-    free(json_str);
-    nng_mtx_unlock(routine->mutex);
-    return 0;
-}
-
 static void mqtt_routine_response(const char *topic_name, size_t topic_len,
                                   void *payload, const size_t len,
                                   void *context)
@@ -299,21 +252,6 @@ static mqtt_routine_t *mqtt_routine_start(neu_plugin_t *plugin,
     topics_generate(routine->topics, plugin, &routine->option);
     topics_subscribe(routine->topics, routine->client);
 
-    // cache send
-    const size_t max_bytes = routine->option.cache * 1024 * 1024;
-    const size_t max_items = 0;
-    routine->cache         = neu_mem_cache_create(max_bytes, max_items);
-
-    nng_mtx_alloc(&routine->mutex);
-    routine->events               = neu_event_new();
-    neu_event_timer_param_t param = {
-        .second      = 0,
-        .millisecond = 100,
-        .usr_data    = routine,
-        .cb          = mqtt_routine_cache_send,
-    };
-
-    routine->send = neu_event_add_timer(routine->events, param);
     return routine;
 }
 
@@ -321,12 +259,6 @@ static void mqtt_routine_stop(mqtt_routine_t *routine)
 {
     assert(NULL != routine);
 
-    // stop cache send timer
-    neu_event_del_timer(routine->events, routine->send);
-    neu_event_close(routine->events);
-    nng_mtx_free(routine->mutex);
-    neu_mem_cache_destroy(routine->cache);
-
     // stop recevied
     neu_mqtt_client_suspend(routine->client);
 
@@ -358,6 +290,8 @@ static int mqtt_plugin_close(neu_plugin_t *plugin)
     return NEU_ERR_SUCCESS;
 }
 
+static void plugin_cache_init(neu_plugin_t *plugin);
+
 static int mqtt_plugin_init(neu_plugin_t *plugin)
 {
     assert(NULL != plugin);
@@ -365,6 +299,7 @@ static int mqtt_plugin_init(neu_plugin_t *plugin)
     plugin->routine = NULL;
     plugin->running = false;
     plugin->config  = NULL;
+    plugin_cache_init(plugin);
 
     const char *name = neu_plugin_module.module_name;
     plog_info(plugin, "initialize plugin: %s", name);
@@ -418,11 +353,24 @@ static void plugin_config_save(neu_plugin_t *plugin, const char *config)
     plugin->config = strdup(config);
 }
 
+static void plugin_cache_init(neu_plugin_t *plugin)
+{
+    plugin->cache = neu_mem_cache_create(0, 0);
+    nng_mtx_alloc(&plugin->mutex);
+}
+
+static void plugin_cache_uninit(neu_plugin_t *plugin)
+{
+    nng_mtx_free(plugin->mutex);
+    neu_mem_cache_destroy(plugin->cache);
+}
+
 static int mqtt_plugin_uninit(neu_plugin_t *plugin)
 {
     assert(NULL != plugin);
 
     plugin_stop_running(plugin);
+    plugin_cache_uninit(plugin);
     plugin_config_free(plugin);
 
     const char *name = neu_plugin_module.module_name;
@@ -435,11 +383,17 @@ static int mqtt_plugin_config(neu_plugin_t *plugin, const char *config)
     assert(NULL != plugin);
     assert(NULL != config);
 
+    // Option setting
     if (NEU_ERR_SUCCESS != mqtt_option_validate(plugin, config)) {
         return NEU_ERR_NODE_SETTING_INVALID;
     }
 
     plugin_config_save(plugin, config);
+    size_t cache_size = mqtt_option_item_cache(plugin, config);
+    size_t max_bytes  = cache_size * 1024 * 1024;
+    nng_mtx_lock(plugin->mutex);
+    neu_mem_cache_resize(plugin->cache, max_bytes, 0);
+    nng_mtx_unlock(plugin->mutex);
 
     // Routine not running
     if (!plugin->running) {
@@ -544,6 +498,35 @@ static void cache_string_release(void *data)
     free(data);
 }
 
+static int cache_send(neu_plugin_t *plugin)
+{
+    mqtt_routine_t *routine = plugin->routine;
+
+    nng_mtx_lock(plugin->mutex);
+    cache_item_t item = neu_mem_cache_earliest(plugin->cache);
+    if (NULL == item.data) {
+        nng_mtx_unlock(plugin->mutex);
+        return -1;
+    }
+
+    int                type     = TOPIC_TYPE_UPLOAD;
+    struct topic_pair *pair     = topics_find_type(routine->topics, type);
+    char *             json_str = item.data;
+    const char *       topic    = pair->topic_response;
+    const int          qos      = pair->qos_response;
+    neu_err_code_e     error =
+        neu_mqtt_client_publish(routine->client, topic, qos,
+                                (unsigned char *) json_str, strlen(json_str));
+    if (NEU_ERR_SUCCESS != error) {
+        plog_error(plugin, "cache publish error code :%d, topoic:%s", error,
+                   topic);
+    }
+
+    free(json_str);
+    nng_mtx_unlock(plugin->mutex);
+    return 0;
+}
+
 static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data)
 {
     neu_reqresp_trans_data_t *trans_data = data;
@@ -569,15 +552,22 @@ static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data)
             .release = cache_string_release,
         };
 
-        nng_mtx_lock(routine->mutex);
-        rc = neu_mem_cache_add(routine->cache, &item);
-        nng_mtx_unlock(routine->mutex);
+        nng_mtx_lock(plugin->mutex);
+        rc = neu_mem_cache_add(plugin->cache, &item);
+        nng_mtx_unlock(plugin->mutex);
 
         if (0 != rc) {
             item.release(item.data);
         }
 
         return NEU_ERR_MQTT_CONNECT_FAILURE;
+    } else {
+        for (int i = 0; i < 100; i++) {
+            int r = cache_send(plugin);
+            if (-1 == r) {
+                break;
+            }
+        }
     }
 
     const char *   topic = pair->topic_response;

plugins/mqtt/mqtt_plugin.h

@@ -26,6 +26,9 @@ extern "C" {
 
 #include <stdlib.h>
 
+#include <nng/nng.h>
+#include <nng/supplemental/util/platform.h>
+
 #include <neuron.h>
 
 typedef struct mqtt_routine mqtt_routine_t;
@@ -35,6 +38,8 @@ struct neu_plugin {
     bool                running;
     mqtt_routine_t *    routine;
     char *              config;
+    neu_mem_cache_t *   cache;
+    nng_mtx *           mutex;
 };
 
 #ifdef __cplusplus

plugins/mqtt/mqtt_util.c

@@ -341,4 +341,19 @@ int mqtt_option_validate(neu_plugin_t *plugin, const char *config)
 
     mqtt_option_uninit(plugin, &option);
     return NEU_ERR_SUCCESS;
+}
+
+size_t mqtt_option_item_cache(neu_plugin_t *plugin, const char *config)
+{
+    neu_mqtt_option_t option = { 0 };
+    int               rc = mqtt_option_init(plugin, (char *) config, &option);
+    if (0 != rc) {
+        plog_error(plugin, "%s", mqtt_option_error(rc));
+        mqtt_option_uninit(plugin, &option);
+        return 64;
+    }
+
+    size_t cache_size = option.cache;
+    mqtt_option_uninit(plugin, &option);
+    return cache_size;
 }

plugins/mqtt/mqtt_util.h

@@ -32,7 +32,8 @@ int mqtt_option_init(neu_plugin_t *plugin, char *config,
 
 const char *mqtt_option_error(int error);
 
-int mqtt_option_validate(neu_plugin_t *plugin, const char *config);
+int    mqtt_option_validate(neu_plugin_t *plugin, const char *config);
+size_t mqtt_option_item_cache(neu_plugin_t *plugin, const char *config);
 
 void mqtt_option_uninit(neu_plugin_t *plugin, neu_mqtt_option_t *option);
 

plugins/restful/rest.c

@@ -302,6 +302,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/driver/driver.c

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

src/base/group.c

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

src/connection/connection.c

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

src/connection/mqtt_c_client.c

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

src/main.c

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

src/persist/persist.c

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

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

src/utils/mem_cache.c

@@ -50,23 +50,26 @@ static bool queue_is_full(neu_mem_cache_t *cache, cache_item_t *item)
     return false;
 }
 
-int neu_mem_cache_add(neu_mem_cache_t *cache, cache_item_t *item)
+static void queue_reduce(neu_mem_cache_t *cache, cache_item_t *item)
 {
-    assert(NULL != cache);
-    assert(NULL != item);
-
     while (queue_is_full(cache, item)) {
         cache_item_t remove_item = neu_mem_cache_earliest(cache);
         if (NULL == remove_item.data) {
-            // size(item.data) >  cache->max_bytes
-            return -1;
+            break;
         }
 
         if (NULL != remove_item.release && NULL != remove_item.data) {
             remove_item.release(remove_item.data);
         }
     }
+}
 
+int neu_mem_cache_add(neu_mem_cache_t *cache, cache_item_t *item)
+{
+    assert(NULL != cache);
+    assert(NULL != item);
+
+    queue_reduce(cache, item);
     element *el = calloc(1, sizeof(element));
     if (NULL == el) {
         return -1;
@@ -126,6 +129,17 @@ void neu_mem_cache_used(neu_mem_cache_t *cache, size_t *bytes, size_t *msgs)
     *msgs  = cache->used_items;
 }
 
+void neu_mem_cache_resize(neu_mem_cache_t *cache, size_t new_bytes,
+                          size_t new_items)
+{
+    assert(NULL != cache);
+
+    cache->max_bytes  = new_bytes;
+    cache->max_items  = new_items;
+    cache_item_t item = { 0 };
+    queue_reduce(cache, &item);
+}
+
 int neu_mem_cache_dump(neu_mem_cache_t *cache, cache_dump callback, void *ctx)
 {
     assert(NULL != cache);

ui_version

@@ -1 +1 @@
-2.1.2
+2.2.1

version

@@ -1 +1 @@
-2.2.0
+2.2.2