Merge pull request #780 from fengzeroz/v2.2

version: 2.2.1

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_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/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:

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/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;
@@ -56,7 +57,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/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.1