Merge pull request #2405 from hxy7yx/v2.10-1

[v2.10]allow float-tag to write integers

.github/workflows/ci.yml

@@ -8,7 +8,7 @@ on:
       - published
 
 env:
-  VERSION: 2.10.1
+  VERSION: 2.10.3
   UI_VERSION: 2.6.1
   BRANCH: v2.10
 

CMakeLists.txt

@@ -166,8 +166,6 @@ file(COPY ${CMAKE_SOURCE_DIR}/dev.conf DESTINATION ${CMAKE_BINARY_DIR}/config)
 file(COPY ${CMAKE_SOURCE_DIR}/neuron.json 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)
 file(COPY ${SQL_SCHEMAS} DESTINATION ${CMAKE_BINARY_DIR}/config)
 file(COPY ${CMAKE_SOURCE_DIR}/default_plugins.json DESTINATION ${CMAKE_BINARY_DIR}/config)
@@ -175,6 +173,7 @@ file(COPY ${CMAKE_SOURCE_DIR}/default_plugins.json DESTINATION ${CMAKE_BINARY_DI
 add_subdirectory(plugins/modbus)
 add_subdirectory(plugins/mqtt)
 add_subdirectory(plugins/ekuiper)
+add_subdirectory(plugins/file)
 
 add_subdirectory(simulator)
 

default_plugins.json

@@ -5,6 +5,7 @@
 		"libplugin-azure-iot.so",
 		"libplugin-ekuiper.so",
 		"libplugin-modbus-tcp.so",
-		"libplugin-modbus-rtu.so"
+		"libplugin-modbus-rtu.so",
+		"libplugin-file.so"
 	]
 }

include/neuron/connection/neu_connection.h

@@ -254,6 +254,8 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len);
 ssize_t neu_conn_udp_recvfrom(neu_conn_t *conn, uint8_t *buf, ssize_t len,
                               void *src);
 
+void neu_conn_clear_recv_buffer(neu_conn_t *conn);
+
 /**
  * @brief Specify the client to send data.
  *

include/neuron/define.h

@@ -24,7 +24,7 @@
 
 #define NEU_VERSION_MAJOR 2
 #define NEU_VERSION_MINOR 10
-#define NEU_VERSION_FIX 1
+#define NEU_VERSION_FIX 3
 
 #define NEU_GET_VERSION_MAJOR(V) ((0xffff0000 & V) >> 16)
 #define NEU_GET_VERSION_MINOR(V) ((0x0000ff00 & V) >> 8)

include/neuron/json/neu_json_rw.h

@@ -63,6 +63,7 @@ typedef struct {
 
 int neu_json_encode_read_resp(void *json_object, void *param);
 int neu_json_encode_read_resp1(void *json_object, void *param);
+int neu_json_encode_read_resp2(void *json_object, void *param);
 int neu_json_encode_read_paginate_resp(void *json_object, void *param);
 
 typedef struct {

include/neuron/type.h

@@ -146,7 +146,7 @@ typedef struct {
     uint8_t *  ptr;
 } neu_value_ptr_t;
 
-#define NEU_VALUE_SIZE 128
+#define NEU_VALUE_SIZE 1024
 
 typedef struct {
     uint8_t bytes[NEU_VALUE_SIZE];

package-sdk.sh

@@ -51,8 +51,6 @@ cp persistence/*.sql ${package_name}/config/
 cp sdk-zlog.conf ${package_name}/config/
 cp zlog.conf ${package_name}/config/
 cp dev.conf ${package_name}/config/
-cp neuron.key ${package_name}/config/
-cp neuron.pem ${package_name}/config/
 cp default_plugins.json ${package_name}/config/
 cp neuron.json ${package_name}/config/
 

plugins/ekuiper/plugin_ekuiper.c

@@ -316,7 +316,7 @@ static int ekuiper_plugin_request(neu_plugin_t *      plugin,
 
     neu_otel_trace_ctx trace = NULL;
     neu_otel_scope_ctx scope = NULL;
-    if (neu_otel_control_is_started()) {
+    if (neu_otel_control_is_started() && header->ctx) {
         trace = neu_otel_find_trace(header->ctx);
         if (trace) {
             scope = neu_otel_add_span(trace);

plugins/file/CMakeLists.txt

@@ -0,0 +1,11 @@
+set(LIBRARY_OUTPUT_PATH "${CMAKE_BINARY_DIR}/plugins")
+
+set(CMAKE_BUILD_RPATH ./)
+file(COPY ${CMAKE_SOURCE_DIR}/plugins/file/file.json DESTINATION ${CMAKE_BINARY_DIR}/plugins/schema/)
+
+set(PLUGIN_NAME plugin-file)
+set(PLUGIN_SOURCES file_plugin.c)
+add_library(${PLUGIN_NAME} SHARED)
+target_include_directories(${PLUGIN_NAME} PRIVATE ${CMAKE_SOURCE_DIR}/include/neuron)
+target_sources(${PLUGIN_NAME} PRIVATE ${PLUGIN_SOURCES})
+target_link_libraries(${PLUGIN_NAME} neuron-base)

plugins/file/file.json

@@ -0,0 +1,9 @@
+{
+	"tag_regex": [
+		{
+			"type": 13,
+			"regex": ""
+		}
+	],
+	"group_interval": 1000
+}

plugins/file/file_plugin.c

@@ -0,0 +1,247 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 3 of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with this program; if not, write to the Free Software Foundation,
+ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+ **/
+#include <errno.h>
+#include <stdlib.h>
+
+#include <neuron.h>
+
+#include "errcodes.h"
+
+struct neu_plugin {
+    neu_plugin_common_t common;
+
+    // neu_events_t *events;
+
+    // int file_length;
+};
+
+static neu_plugin_t *driver_open(void);
+
+static int driver_close(neu_plugin_t *plugin);
+static int driver_init(neu_plugin_t *plugin, bool load);
+static int driver_uninit(neu_plugin_t *plugin);
+static int driver_start(neu_plugin_t *plugin);
+static int driver_stop(neu_plugin_t *plugin);
+static int driver_config(neu_plugin_t *plugin, const char *config);
+static int driver_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
+                          void *data);
+
+static int driver_tag_validator(const neu_datatag_t *tag);
+static int driver_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag);
+static int driver_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group);
+static int driver_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
+                        neu_value_u value);
+
+static const neu_plugin_intf_funs_t plugin_intf_funs = {
+    .open    = driver_open,
+    .close   = driver_close,
+    .init    = driver_init,
+    .uninit  = driver_uninit,
+    .start   = driver_start,
+    .stop    = driver_stop,
+    .setting = driver_config,
+    .request = driver_request,
+
+    .driver.validate_tag  = driver_validate_tag,
+    .driver.group_timer   = driver_group_timer,
+    .driver.group_sync    = NULL,
+    .driver.write_tag     = driver_write,
+    .driver.tag_validator = driver_tag_validator,
+    .driver.write_tags    = NULL,
+    .driver.test_read_tag = NULL,
+    .driver.add_tags      = NULL,
+    .driver.load_tags     = NULL,
+    .driver.del_tags      = NULL,
+};
+
+const neu_plugin_module_t neu_plugin_module = {
+    .version         = NEURON_PLUGIN_VER_1_0,
+    .schema          = "file",
+    .module_name     = "File",
+    .module_descr    = "The plugin is used to read and write files.",
+    .module_descr_zh = "该插件用于读写文件。",
+    .intf_funs       = &plugin_intf_funs,
+    .kind            = NEU_PLUGIN_KIND_SYSTEM,
+    .type            = NEU_NA_TYPE_DRIVER,
+    .display         = true,
+    .single          = false,
+};
+
+static neu_plugin_t *driver_open(void)
+{
+    neu_plugin_t *plugin = calloc(1, sizeof(neu_plugin_t));
+
+    neu_plugin_common_init(&plugin->common);
+
+    return plugin;
+}
+
+static int driver_close(neu_plugin_t *plugin)
+{
+    free(plugin);
+
+    return 0;
+}
+
+static int driver_init(neu_plugin_t *plugin, bool load)
+{
+    (void) load;
+    plog_notice(plugin, "%s init success", plugin->common.name);
+    return 0;
+}
+
+static int driver_uninit(neu_plugin_t *plugin)
+{
+    plog_notice(plugin, "%s uninit success", plugin->common.name);
+
+    return 0;
+}
+
+static int driver_start(neu_plugin_t *plugin)
+{
+    neu_adapter_update_metric_cb_t update_metric =
+        plugin->common.adapter_callbacks->update_metric;
+    update_metric(plugin->common.adapter, NEU_METRIC_LAST_RTT_MS, 1, NULL);
+
+    plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
+    plog_notice(plugin, "start `%s` plugin success", plugin->common.name);
+
+    return 0;
+}
+
+static int driver_stop(neu_plugin_t *plugin)
+{
+    neu_adapter_update_metric_cb_t update_metric =
+        plugin->common.adapter_callbacks->update_metric;
+    update_metric(plugin->common.adapter, NEU_METRIC_LAST_RTT_MS, 9999, NULL);
+
+    plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
+    plog_notice(plugin, "stop `%s` plugin success", plugin->common.name);
+
+    return 0;
+}
+static int driver_config(neu_plugin_t *plugin, const char *config)
+{
+    (void) plugin;
+    (void) config;
+
+    return 0;
+}
+
+static int driver_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
+                          void *data)
+{
+    (void) plugin;
+    (void) head;
+    (void) data;
+    return 0;
+}
+
+static int driver_tag_validator(const neu_datatag_t *tag)
+{
+    (void) tag;
+    return 0;
+}
+
+static int driver_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag)
+{
+    if (tag->type != NEU_TYPE_STRING) {
+        return NEU_ERR_TAG_TYPE_NOT_SUPPORT;
+    }
+
+    plog_notice(plugin, "validate tag success, name:%s, address:%s, type:%d ",
+                tag->name, tag->address, tag->type);
+
+    return 0;
+}
+
+static int driver_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
+{
+    neu_adapter_update_metric_cb_t update_metric =
+        plugin->common.adapter_callbacks->update_metric;
+
+    utarray_foreach(group->tags, neu_datatag_t *, tag)
+    {
+        neu_dvalue_t dvalue = { 0 };
+
+        FILE *fp = fopen(tag->address, "r");
+        if (fp == NULL) {
+            dvalue.type = NEU_TYPE_ERROR;
+            if (errno == ENOENT) {
+                dvalue.value.i32 = NEU_ERR_FILE_NOT_EXIST;
+            } else {
+                dvalue.value.i32 = NEU_ERR_FILE_OPEN_FAILURE;
+            }
+        } else {
+            char *buf = calloc(1, 1024);
+            int   ret = fread(buf, 1, 1024, fp);
+            if (ret < 0) {
+                dvalue.type      = NEU_TYPE_ERROR;
+                dvalue.value.i32 = NEU_ERR_FILE_READ_FAILURE;
+            } else {
+                dvalue.type = NEU_TYPE_STRING;
+                strncpy(dvalue.value.str, buf, NEU_VALUE_SIZE - 1);
+            }
+
+            fclose(fp);
+            free(buf);
+        }
+
+        plugin->common.adapter_callbacks->driver.update(
+            plugin->common.adapter, group->group_name, tag->name, dvalue);
+    }
+
+    update_metric(plugin->common.adapter, NEU_METRIC_SEND_BYTES, 0, NULL);
+    update_metric(plugin->common.adapter, NEU_METRIC_RECV_BYTES, 0, NULL);
+    update_metric(plugin->common.adapter, NEU_METRIC_LAST_RTT_MS, 1, NULL);
+    update_metric(plugin->common.adapter, NEU_METRIC_GROUP_LAST_SEND_MSGS, 1,
+                  group->group_name);
+    return 0;
+}
+
+static int driver_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
+                        neu_value_u value)
+{
+    FILE *fp = fopen(tag->address, "w");
+    if (fp == NULL) {
+        if (errno == ENOENT) {
+            plugin->common.adapter_callbacks->driver.write_response(
+                plugin->common.adapter, req, NEU_ERR_FILE_NOT_EXIST);
+        } else {
+            plugin->common.adapter_callbacks->driver.write_response(
+                plugin->common.adapter, req, NEU_ERR_FILE_OPEN_FAILURE);
+        }
+    } else {
+        if (tag->type == NEU_TYPE_STRING) {
+            size_t ret = fwrite(value.str, 1, strlen(value.str), fp);
+            if (ret != strlen(value.str)) {
+                plugin->common.adapter_callbacks->driver.write_response(
+                    plugin->common.adapter, req, NEU_ERR_FILE_WRITE_FAILURE);
+            }
+        } else {
+            plugin->common.adapter_callbacks->driver.write_response(
+                plugin->common.adapter, req, NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH);
+        }
+
+        fclose(fp);
+        plugin->common.adapter_callbacks->driver.write_response(
+            plugin->common.adapter, req, 0);
+    }
+    return 0;
+}

plugins/modbus/modbus_req.c

@@ -113,6 +113,8 @@ int modbus_send_msg(void *ctx, uint16_t n_byte, uint8_t *bytes)
     neu_plugin_t *plugin = (neu_plugin_t *) ctx;
     int           ret    = 0;
 
+    neu_conn_clear_recv_buffer(plugin->conn);
+
     plog_send_protocol(plugin, bytes, n_byte);
 
     if (plugin->is_server) {

plugins/mqtt/mqtt_handle.c

@@ -82,7 +82,7 @@ char *generate_upload_json(neu_plugin_t *plugin, neu_reqresp_trans_data_t *data,
                                   neu_json_encode_read_periodic_resp,
                                   &json_str);
     } else if (MQTT_UPLOAD_FORMAT_TAGS == format) { // tags
-        neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp, &header,
+        neu_json_encode_with_mqtt(&json, neu_json_encode_read_resp2, &header,
                                   neu_json_encode_read_periodic_resp,
                                   &json_str);
     } else {

plugins/mqtt/mqtt_plugin_intf.c

@@ -463,7 +463,7 @@ int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
     neu_otel_trace_ctx trace           = NULL;
     neu_otel_scope_ctx scope           = NULL;
     char               new_span_id[36] = { 0 };
-    if (neu_otel_control_is_started()) {
+    if (neu_otel_control_is_started() && head->ctx) {
         trace = neu_otel_find_trace(head->ctx);
         if (trace) {
             scope = neu_otel_add_span(trace);

plugins/restful/normal_handle.c

@@ -307,7 +307,7 @@ void handle_get_plugin_schema(nng_aio *aio)
         return;
     }
 
-    if (0 > neu_asprintf(&schema_path, "%s/schema/%s.json", g_plugin_dir,
+    if (0 > neu_asprintf(&schema_path, "%s/system/schema/%s.json", g_plugin_dir,
                          schema_name)) {
         NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
             neu_http_response(aio, error_code.error, result_error);
@@ -332,8 +332,8 @@ void handle_get_plugin_schema(nng_aio *aio)
 
     if (NULL == buf) {
         free(schema_path);
-        if (0 > neu_asprintf(&schema_path, "%s/system/schema/%s.json",
-                             g_plugin_dir, schema_name)) {
+        if (0 > neu_asprintf(&schema_path, "%s/schema/%s.json", g_plugin_dir,
+                             schema_name)) {
             NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
                 neu_http_response(aio, error_code.error, result_error);
             });

plugins/restful/rest.c

@@ -176,7 +176,7 @@ static int dashb_plugin_request(neu_plugin_t *      plugin,
 
     neu_otel_trace_ctx trace = NULL;
     neu_otel_scope_ctx scope = NULL;
-    if (neu_otel_control_is_started()) {
+    if (neu_otel_control_is_started() && header->ctx) {
         trace = neu_otel_find_trace(header->ctx);
         if (trace) {
             scope = neu_otel_add_span(trace);

plugins/restful/rw_handle.c

@@ -284,7 +284,7 @@ void handle_write(nng_aio *aio)
             }
 
             if (req->t == NEU_JSON_ARRAY_INT64 &&
-                req->value.val_array_int64.length > NEU_VALUE_SIZE) {
+                req->value.val_array_int64.length > 128) {
                 NEU_JSON_RESPONSE_ERROR(NEU_ERR_STRING_TOO_LONG, {
                     neu_http_response(aio, NEU_ERR_STRING_TOO_LONG,
                                       result_error);

sdk-install.sh

@@ -25,8 +25,6 @@ mkdir -p ${neuron_bin}/plugins
 cp ./neuron ${neuron_bin}/
 cp config/sdk-zlog.conf ${neuron_bin}/config/zlog.conf
 cp config/dev.conf ${neuron_bin}/config/
-cp config/neuron.key ${neuron_bin}/config/
-cp config/neuron.pem ${neuron_bin}/config/
 cp config/default_plugins.json ${neuron_bin}/config/
 cp config/*.sql ${neuron_bin}/config/
 

src/adapter/driver/driver.c

@@ -834,7 +834,6 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type,
         break;
     case NEU_TYPE_BYTES:
         if (value->type == NEU_TYPE_ARRAY_INT64) {
-            value->value.bytes.length = value->value.i64s.length;
             for (int i = 0; i < value->value.i64s.length; i++) {
                 value->value.bytes.bytes[i] =
                     (uint8_t) value->value.i64s.i64s[i];
@@ -842,44 +841,44 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type,
             for (int i = value->value.i64s.length; i < NEU_VALUE_SIZE; i++) {
                 value->value.bytes.bytes[i] = 0;
             }
+            value->value.bytes.length = value->value.i64s.length;
         }
         break;
     case NEU_TYPE_ARRAY_INT8:
         if (value->type == NEU_TYPE_ARRAY_INT64) {
-            value->value.i8s.length = value->value.i64s.length;
             for (int i = 0; i < value->value.i64s.length; i++) {
                 value->value.i8s.i8s[i] = (int8_t) value->value.i64s.i64s[i];
             }
             for (int i = value->value.i64s.length; i < NEU_VALUE_SIZE; i++) {
                 value->value.i8s.i8s[i] = 0;
             }
+            value->value.i8s.length = value->value.i64s.length;
         }
         break;
     case NEU_TYPE_ARRAY_UINT8:
         if (value->type == NEU_TYPE_ARRAY_INT64) {
-            value->value.u8s.length = value->value.i64s.length;
             for (int i = 0; i < value->value.i64s.length; i++) {
                 value->value.u8s.u8s[i] = (uint8_t) value->value.i64s.i64s[i];
             }
             for (int i = value->value.i64s.length; i < NEU_VALUE_SIZE; i++) {
                 value->value.u8s.u8s[i] = 0;
             }
+            value->value.u8s.length = value->value.i64s.length;
         }
         break;
     case NEU_TYPE_ARRAY_INT16:
         if (value->type == NEU_TYPE_ARRAY_INT64) {
-            value->value.i16s.length = value->value.i64s.length;
             for (int i = 0; i < value->value.i64s.length; i++) {
                 value->value.i16s.i16s[i] = (int16_t) value->value.i64s.i64s[i];
             }
             for (int i = value->value.i64s.length; i < NEU_VALUE_SIZE; i++) {
                 value->value.i16s.i16s[i] = 0;
             }
+            value->value.i16s.length = value->value.i64s.length;
         }
         break;
     case NEU_TYPE_ARRAY_UINT16:
         if (value->type == NEU_TYPE_ARRAY_INT64) {
-            value->value.u16s.length = value->value.i64s.length;
             for (int i = 0; i < value->value.i64s.length; i++) {
                 value->value.u16s.u16s[i] =
                     (uint16_t) value->value.i64s.i64s[i];
@@ -887,22 +886,22 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type,
             for (int i = value->value.i64s.length; i < NEU_VALUE_SIZE; i++) {
                 value->value.u16s.u16s[i] = 0;
             }
+            value->value.u16s.length = value->value.i64s.length;
         }
         break;
     case NEU_TYPE_ARRAY_INT32:
         if (value->type == NEU_TYPE_ARRAY_INT64) {
-            value->value.i32s.length = value->value.i64s.length;
             for (int i = 0; i < value->value.i64s.length; i++) {
                 value->value.i32s.i32s[i] = (int32_t) value->value.i64s.i64s[i];
             }
             for (int i = value->value.i64s.length; i < NEU_VALUE_SIZE; i++) {
                 value->value.i32s.i32s[i] = 0;
             }
+            value->value.i32s.length = value->value.i64s.length;
         }
         break;
     case NEU_TYPE_ARRAY_UINT32:
         if (value->type == NEU_TYPE_ARRAY_INT64) {
-            value->value.u32s.length = value->value.i64s.length;
             for (int i = 0; i < value->value.i64s.length; i++) {
                 value->value.u32s.u32s[i] =
                     (uint32_t) value->value.i64s.i64s[i];
@@ -910,11 +909,11 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type,
             for (int i = value->value.i64s.length; i < NEU_VALUE_SIZE; i++) {
                 value->value.u32s.u32s[i] = 0;
             }
+            value->value.u32s.length = value->value.i64s.length;
         }
         break;
     case NEU_TYPE_ARRAY_UINT64:
         if (value->type == NEU_TYPE_ARRAY_INT64) {
-            value->value.u64s.length = value->value.i64s.length;
             for (int i = 0; i < value->value.i64s.length; i++) {
                 value->value.u64s.u64s[i] =
                     (uint64_t) value->value.i64s.i64s[i];
@@ -922,37 +921,41 @@ static void fix_value(neu_datatag_t *tag, neu_type_e value_type,
             for (int i = value->value.i64s.length; i < NEU_VALUE_SIZE; i++) {
                 value->value.u64s.u64s[i] = 0;
             }
+            value->value.u64s.length = value->value.i64s.length;
         }
         break;
     case NEU_TYPE_ARRAY_FLOAT:
         if (value->type == NEU_TYPE_ARRAY_INT64) {
-            value->value.f32s.length = value->value.i64s.length;
+
             for (int i = 0; i < value->value.i64s.length; i++) {
                 value->value.f32s.f32s[i] = (float) value->value.i64s.i64s[i];
             }
             for (int i = value->value.i64s.length; i < NEU_VALUE_SIZE; i++) {
                 value->value.f32s.f32s[i] = 0;
             }
+            value->value.f32s.length = value->value.i64s.length;
         }
         if (value->type == NEU_TYPE_ARRAY_DOUBLE) {
-            value->value.f32s.length = value->value.f64s.length;
+
             for (int i = 0; i < value->value.f64s.length; i++) {
                 value->value.f32s.f32s[i] = (float) value->value.f64s.f64s[i];
             }
             for (int i = value->value.f64s.length; i < NEU_VALUE_SIZE; i++) {
                 value->value.f32s.f32s[i] = 0;
             }
+            value->value.f32s.length = value->value.f64s.length;
         }
         break;
     case NEU_TYPE_ARRAY_DOUBLE:
         if (value->type == NEU_TYPE_ARRAY_INT64) {
-            value->value.f64s.length = value->value.i64s.length;
+
             for (int i = 0; i < value->value.i64s.length; i++) {
                 value->value.f64s.f64s[i] = (double) value->value.i64s.i64s[i];
             }
             for (int i = value->value.i64s.length; i < NEU_VALUE_SIZE; i++) {
                 value->value.f64s.f64s[i] = 0;
             }
+            value->value.f64s.length = value->value.i64s.length;
         }
         break;
     default:
@@ -1109,21 +1112,13 @@ int is_value_in_range(neu_type_e tag_type, int64_t value, double value_d,
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
             }
         case NEU_TYPE_FLOAT:
-            if (write_type != NEU_TYPE_DOUBLE) {
-                return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
-            } else if (result < (int64_t) -FLT_MAX ||
-                       result > (int64_t) FLT_MAX) {
+            if (result < (int64_t) -FLT_MAX || result > (int64_t) FLT_MAX) {
                 return NEU_ERR_PLUGIN_TAG_VALUE_OUT_OF_RANGE;
             } else {
                 return NEU_ERR_SUCCESS;
             }
         case NEU_TYPE_DOUBLE:
-            if (write_type != NEU_TYPE_DOUBLE) {
-                return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
-            } else {
-                return NEU_ERR_SUCCESS;
-            }
-            return check_value(write_type, result, INT16_MIN, INT16_MAX);
+            return NEU_ERR_SUCCESS;
         case NEU_TYPE_STRING:
         case NEU_TYPE_BOOL:
         case NEU_TYPE_BIT:
@@ -1167,20 +1162,13 @@ int is_value_in_range(neu_type_e tag_type, int64_t value, double value_d,
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
             }
         case NEU_TYPE_FLOAT:
-            if (write_type != NEU_TYPE_DOUBLE) {
-                return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
-            } else if (value < (int64_t) -FLT_MAX ||
-                       value > (int64_t) FLT_MAX) {
+            if (value < (int64_t) -FLT_MAX || value > (int64_t) FLT_MAX) {
                 return NEU_ERR_PLUGIN_TAG_VALUE_OUT_OF_RANGE;
             } else {
                 return NEU_ERR_SUCCESS;
             }
         case NEU_TYPE_DOUBLE:
-            if (write_type != NEU_TYPE_DOUBLE) {
-                return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;
-            } else {
-                return NEU_ERR_SUCCESS;
-            }
+            return NEU_ERR_SUCCESS;
         case NEU_TYPE_STRING:
             if (write_type != NEU_TYPE_STRING) {
                 return NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH;

src/connection/connection.c

@@ -407,6 +407,40 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len)
     return ret;
 }
 
+void neu_conn_clear_recv_buffer(neu_conn_t *conn)
+{
+    if (!conn->is_connected) {
+        return;
+    }
+
+    uint8_t temp_buf[256];
+    ssize_t ret;
+
+    switch (conn->param.type) {
+    case NEU_CONN_TCP_CLIENT:
+        do {
+            ret = recv(conn->fd, temp_buf, sizeof(temp_buf), MSG_DONTWAIT);
+
+            if (ret > 0) {
+                continue;
+            } else if (ret == 0) {
+                zlog_info(conn->param.log,
+                          "Connection closed while clearing buffer.");
+                break;
+            } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                break;
+            } else {
+                zlog_error(conn->param.log, "Error clearing buffer: %s",
+                           strerror(errno));
+                break;
+            }
+        } while (ret > 0);
+        break;
+    default:
+        break;
+    }
+}
+
 ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
 {
     ssize_t ret = 0;

src/core/manager.c

@@ -1538,6 +1538,47 @@ static char *file_save_tmp(const char *data, const char *suffix)
     return tmp_path;
 }
 
+static int copy_file(const char *src, const char *dst)
+{
+    FILE * source, *destination;
+    char   buffer[1024];
+    size_t bytes_read;
+
+    source = fopen(src, "rb");
+    if (source == NULL) {
+        nlog_error("failed to open source file!");
+        return -1;
+    }
+
+    destination = fopen(dst, "wb");
+    if (destination == NULL) {
+        nlog_error("failed to open destination file!");
+        fclose(source);
+        return -1;
+    }
+
+    while ((bytes_read = fread(buffer, 1, sizeof(buffer), source)) > 0) {
+        if (fwrite(buffer, 1, bytes_read, destination) != bytes_read) {
+            nlog_error("failed to write to destination file!");
+            fclose(source);
+            fclose(destination);
+            return -1;
+        }
+    }
+
+    if (ferror(source)) {
+        nlog_error("error reading source file");
+        fclose(source);
+        fclose(destination);
+        return -1;
+    }
+
+    fclose(source);
+    fclose(destination);
+
+    return 0;
+}
+
 static bool mv_tmp_library_file(neu_plugin_kind_e kind, const char *tmp_path,
                                 const char *library)
 {
@@ -1564,9 +1605,8 @@ static bool mv_tmp_library_file(neu_plugin_kind_e kind, const char *tmp_path,
         }
     }
 
-    if (rename(tmp_path, file_name) != 0) {
-        nlog_error("%s rename %s fail, err:%s", tmp_path, file_name,
-                   strerror(errno));
+    if (copy_file(tmp_path, file_name) != 0) {
+        nlog_error("%s copy_file %s fail!", tmp_path, file_name);
         return false;
     } else {
         return true;
@@ -1604,9 +1644,8 @@ static bool mv_tmp_schema_file(neu_plugin_kind_e kind, const char *tmp_path,
         }
     }
 
-    if (rename(tmp_path, file_name) != 0) {
-        nlog_error("%s rename %s fail, err:%s", tmp_path, file_name,
-                   strerror(errno));
+    if (copy_file(tmp_path, file_name) != 0) {
+        nlog_error("%s copy_file %s fail!", tmp_path, file_name);
         return false;
     } else {
         return true;

src/otel/otel_manager.c

@@ -186,6 +186,20 @@ void neu_otel_free_span(Opentelemetry__Proto__Trace__V1__Span *span)
 neu_otel_trace_ctx neu_otel_create_trace(const char *trace_id, void *req_ctx,
                                          uint32_t flags, const char *tracestate)
 {
+    trace_ctx_table_ele_t *find = NULL;
+
+    pthread_mutex_lock(&table_mutex);
+
+    HASH_FIND(hh, traces_table, &req_ctx, sizeof(req_ctx), find);
+
+    if (find) {
+        HASH_DEL(traces_table, find);
+        neu_otel_free_trace(find->ctx);
+        free(find);
+    }
+
+    pthread_mutex_unlock(&table_mutex);
+
     trace_ctx_t *ctx = calloc(1, sizeof(trace_ctx_t));
     opentelemetry__proto__trace__v1__traces_data__init(&ctx->trace_data);
     strncpy((char *) ctx->trace_id, trace_id, 64);
@@ -458,7 +472,16 @@ neu_otel_trace_ctx neu_otel_find_trace(void *req_ctx)
     pthread_mutex_unlock(&table_mutex);
 
     if (find) {
-        return (void *) find->ctx;
+        if (find->ctx->final &&
+            find->ctx->span_num ==
+                find->ctx->trace_data.resource_spans[0]
+                    ->scope_spans[0]
+                    ->n_spans &&
+            find->ctx->expected_span_num <= 0) {
+            return NULL;
+        } else {
+            return (void *) find->ctx;
+        }
     } else {
         return NULL;
     }
@@ -880,7 +903,8 @@ void neu_otel_scope_set_status_code(neu_otel_scope_ctx     ctx,
     scope->span->status =
         calloc(1, sizeof(Opentelemetry__Proto__Trace__V1__Status));
     opentelemetry__proto__trace__v1__status__init(scope->span->status);
-    scope->span->status->code    = code;
+    scope->span->status->code =
+        (Opentelemetry__Proto__Trace__V1__Status__StatusCode) code;
     scope->span->status->message = strdup(desc);
 }
 
@@ -893,7 +917,8 @@ void neu_otel_scope_set_status_code2(neu_otel_scope_ctx     ctx,
     opentelemetry__proto__trace__v1__status__init(scope->span->status);
     char *error_buf = calloc(1, 16);
     sprintf(error_buf, "%d", errorno);
-    scope->span->status->code    = code;
+    scope->span->status->code =
+        (Opentelemetry__Proto__Trace__V1__Status__StatusCode) code;
     scope->span->status->message = error_buf;
 }
 
@@ -930,8 +955,10 @@ int neu_otel_trace_pack(neu_otel_trace_ctx ctx, uint8_t *out)
 
 void neu_otel_new_span_id(char *id)
 {
+    int64_t p_id = (int64_t) pthread_self();
     for (int i = SPAN_ID_LENGTH - 1; i >= 0; i--) {
-        id[i] = ID_CHARSET[rand() % 16];
+        int64_t rand_id = (int64_t) rand() + neu_time_ns() + p_id;
+        id[i]           = ID_CHARSET[rand_id % 16];
     }
     id[SPAN_ID_LENGTH] = '\0';
 }
@@ -1007,6 +1034,8 @@ static int otel_timer_cb(void *data)
                 HASH_DEL(traces_table, el);
                 neu_otel_free_trace(el->ctx);
                 free(el);
+            } else {
+                break;
             }
         } else if (neu_time_ms() - el->ctx->ts >= TRACE_TIME_OUT) {
             nlog_debug("trace:%s time out", (char *) el->ctx->trace_id);
@@ -1033,7 +1062,7 @@ void neu_otel_start()
         param.usr_data = NULL;
 
         param.second      = 0;
-        param.millisecond = 100;
+        param.millisecond = 80;
         param.cb          = otel_timer_cb;
         param.type        = NEU_EVENT_TIMER_BLOCK;
 

src/parser/neu_json_rw.c

@@ -291,6 +291,53 @@ int neu_json_encode_read_resp1(void *json_object, void *param)
     return ret;
 }
 
+int neu_json_encode_read_resp2(void *json_object, void *param)
+{
+    int                   ret  = 0;
+    neu_json_read_resp_t *resp = (neu_json_read_resp_t *) param;
+
+    void *                    tag_array = neu_json_array();
+    neu_json_read_resp_tag_t *p_tag     = resp->tags;
+    for (int i = 0; i < resp->n_tag; i++) {
+        neu_json_elem_t tag_elems[2 + NEU_TAG_META_SIZE] = { 0 };
+
+        tag_elems[0].name      = "name";
+        tag_elems[0].t         = NEU_JSON_STR;
+        tag_elems[0].v.val_str = p_tag->name;
+
+        if (p_tag->error != 0) {
+            tag_elems[1].name      = "error";
+            tag_elems[1].t         = NEU_JSON_INT;
+            tag_elems[1].v.val_int = p_tag->error;
+        } else {
+            tag_elems[1].name      = "value";
+            tag_elems[1].t         = p_tag->t;
+            tag_elems[1].v         = p_tag->value;
+            tag_elems[1].precision = p_tag->precision;
+        }
+
+        for (int k = 0; k < p_tag->n_meta; k++) {
+            tag_elems[2 + k].name = p_tag->metas[k].name;
+            tag_elems[2 + k].t    = p_tag->metas[k].t;
+            tag_elems[2 + k].v    = p_tag->metas[k].value;
+        }
+
+        tag_array =
+            neu_json_encode_array(tag_array, tag_elems, 2 + p_tag->n_meta);
+        p_tag++;
+    }
+
+    neu_json_elem_t resp_elems[] = { {
+        .name         = "tags",
+        .t            = NEU_JSON_OBJECT,
+        .v.val_object = tag_array,
+    } };
+    ret = neu_json_encode_field(json_object, resp_elems,
+                                NEU_JSON_ELEM_SIZE(resp_elems));
+
+    return ret;
+}
+
 int neu_json_decode_write_req(char *buf, neu_json_write_req_t **result)
 {
     void *json_obj = neu_json_decode_new(buf);

src/utils/neu_jwt.c

@@ -43,6 +43,46 @@ static struct public_key_store key_store;
 static char                    neuron_private_key[2048] = { 0 };
 static char                    neuron_public_key[2048]  = { 0 };
 
+static const char *private_key_content =
+    "-----BEGIN RSA PRIVATE KEY-----\n"
+    "MIIEpAIBAAKCAQEAzrO7FUz4GGKl4nP5lwKMh5uageJAAHnLZpVxTR1wXA9E870s\n"
+    "U03gHnPWsh2bXCCx85ymYaiu/5VlptgnU0rQB1I3xGRG8eHmHPrhHNWDRKpY6oY8\n"
+    "AnTPqXvfa/Z4u+njJzDSqi4qM8Gwfqlgpjlg8DUF+sheLx8hL+x9XFzovTxafrmy\n"
+    "JRRAVQTkhB7esKKqRF8BMjdOdYZxJEv9jxa0BwXcjprPDlj5TV6k3wJqbq4UtnEE\n"
+    "VbjEQKqFVkfY4FuJ5NSPHpGEix94PGQGRsiWkGX0vx5udHJ0SlXM6qpMLny18FhW\n"
+    "bwKNX3urc3mImn/j3zcJ5x1d+FlXAXs1NBSeTQIDAQABAoIBAQDJOT8PZXbAhohn\n"
+    "A/AeimS0P08S0mbsD6VroGBEajxP4q2FeswD7PQZsTt4+kmcTlfuiLmQqN50AcSL\n"
+    "wDHIbDRIbEnN7rECGKAj5jfwEgtQdWVKKpOQ8JaYr/a466BtjyuLo4PyGC8NY6mm\n"
+    "JM3qBEHSlkvT7+uAhBWSye7gU7JfRPoCeFRpaMMC3Ad0DaIt/Pdt9CddNwj56+9j\n"
+    "gl//blNnBImjvA7/kh9gL69SYVrQoG9vm/UgFbdIwd+w2hhhpx6kuHjRJ21xIV/y\n"
+    "Yh9k4QhPr7zIkhQ8yn11ZhyZkMC8XAMy7NKCmAFIPNJ8jwvgBkunPEk2aYBeDUaq\n"
+    "a30VwlIhAoGBAOnm8DW5DROcfSj4Q8zsvgmvhGM0zzF5zmA+4E+tEbf4t6sDUjSs\n"
+    "ADo2EaiJZz75vLsD7UZHKzFEV/wnuT7SiNCbrJafx8Tp2UgijTkW8gEhAWREJDn5\n"
+    "FS375RKFdEcpmVHWgsuz4z/0L3BGxmyYWQxx5pD2NL49ZpTIHAHm/XETAoGBAOI6\n"
+    "8Y08TmM2bnXbL/MggYMJlqOyv27ueVYz9of3FYFAMIe7ebSgrlSCOwsoDZYmegBM\n"
+    "OrowR3AAnoP4fjGNbIPjNYbiv1qLOdtnvPxYotEz+ZaUBSn9V/gUEVFlgrKxgTKr\n"
+    "sqi/QMyyJplBTUkUuqYRDuv6TZ3OWKiDUoXTdv8fAoGBAIbbyf+PlESMY3Vtvtm2\n"
+    "XdODyRbR6ewiyKShW/9UT/T8iBknrwwDZ5YoeoHrxwV+RBynpPRyMCsVto0B8kKQ\n"
+    "bKWqPBYURb/4/Hgkw7v4yMtx1jWTPDfYryd2JptJKsOk7mtK/Nqp+wpypa9cfyc5\n"
+    "p44PVdqauco96Jk7zzohjlrVAoGAfjQDrJaH5DDpTjYIeckYdtFSh9+fi3LdnYk/\n"
+    "bnoYNRJqAE5Fhs5ccih0Z7TgX3L3fFMKL/Pe5kxyIYzuWRZcAvctVSIJPamNjShB\n"
+    "9UQ9EBe+lJHej54VBP+s6YuHbcg4GtxNvnVy4L5Bah1T0AEQXrQFbv8jbXU/YEJi\n"
+    "NuXQ7GECgYA0UNvuZxVSDn8fBxARcX3ub5aok5owdLL3SOCPMpNrmOBQj8iJGT+5\n"
+    "nv7fH1TWVh9yUdRwhd81s5H/PO3/myLmgq0TcPvewtEEwY+++QpsYZwHTDXgxbbw\n"
+    "++lmJali4fzmd/72JqocGABYhZFt0jbcQ1QEcqUYcD380+cNVZPO+w==\n"
+    "-----END RSA PRIVATE KEY-----\n";
+
+const char *public_key_content =
+    "-----BEGIN PUBLIC KEY-----\n"
+    "MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAzrO7FUz4GGKl4nP5lwKM\n"
+    "h5uageJAAHnLZpVxTR1wXA9E870sU03gHnPWsh2bXCCx85ymYaiu/5VlptgnU0rQ\n"
+    "B1I3xGRG8eHmHPrhHNWDRKpY6oY8AnTPqXvfa/Z4u+njJzDSqi4qM8Gwfqlgpjlg\n"
+    "8DUF+sheLx8hL+x9XFzovTxafrmyJRRAVQTkhB7esKKqRF8BMjdOdYZxJEv9jxa0\n"
+    "BwXcjprPDlj5TV6k3wJqbq4UtnEEVbjEQKqFVkfY4FuJ5NSPHpGEix94PGQGRsiW\n"
+    "kGX0vx5udHJ0SlXM6qpMLny18FhWbwKNX3urc3mImn/j3zcJ5x1d+FlXAXs1NBSe\n"
+    "TQIDAQAB\n"
+    "-----END PUBLIC KEY-----\n";
+
 static int find_key(const char *name)
 {
     for (int i = 0; i < key_store.size; i++) {
@@ -95,15 +135,13 @@ static char *load_key(const char *dir, char *name)
 
 static void load_neuron_key(const char *dir_path)
 {
-    char *content = load_key(dir_path, "neuron.key");
-    assert(content != NULL);
 
-    strncpy(neuron_private_key, content, sizeof(neuron_private_key) - 1);
+    (void) dir_path;
+    strncpy(neuron_private_key, private_key_content,
+            sizeof(neuron_private_key) - 1);
 
-    content = load_key(dir_path, "neuron.pem");
-    assert(content != NULL);
-
-    strncpy(neuron_public_key, content, sizeof(neuron_public_key) - 1);
+    strncpy(neuron_public_key, public_key_content,
+            sizeof(neuron_public_key) - 1);
 }
 
 static void scanf_key(const char *dir_path)

tests/ft/driver/test_modbus.py

@@ -635,12 +635,6 @@ class TestModbus:
         assert error.NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH == response.json()[
             'error']
 
-        response = api.write_tag(
-            node=param[0], group='group', tag=hold_float[0]['name'], value=1)
-        assert 400 == response.status_code
-        assert error.NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH == response.json()[
-            'error']
-
         response = api.write_tags(node=param[0], group='group', tag_values=[
             {"tag": hold_int16[0]['name'], "value": 32768},
             {"tag": hold_uint16[0]['name'], "value": 2},

tests/ft/persist/test_persist.py

@@ -463,19 +463,12 @@ class TestPersist:
 
     @description(given="upload pem in certs", when="login", then="success")
     def test_jwt(self):
-        if not os.path.exists('./build/certs/'):
-            os.makedirs('./build/certs/')
-
-        shutil.copyfile('./build/config/neuron.pem', './build/certs/neuron.pem')
-
         p = process.NeuronProcess()
         p.start()
 
         response = api.login()
         assert 200 == response.status_code
 
-        shutil.rmtree('./build/certs/')
-
         p.stop()
 
     @description(given="new schema", when="upload schema , restart and add tags", then="success")

tests/ut/CMakeLists.txt

@@ -19,8 +19,6 @@ target_include_directories(http_test PRIVATE
 	${CMAKE_SOURCE_DIR}/plugins/restful)
 target_link_libraries(http_test neuron-base gtest_main gtest jansson nng)
 
-file(COPY ${CMAKE_SOURCE_DIR}/neuron.key DESTINATION ${UT_DIRECTORY}/config)
-file(COPY ${CMAKE_SOURCE_DIR}/neuron.pem DESTINATION ${UT_DIRECTORY}/config)
 add_executable(jwt_test jwt_test.cc)
 target_include_directories(jwt_test PRIVATE 
     ${CMAKE_SOURCE_DIR}/src 

version

@@ -1 +1 @@
-2.10.1
+2.10.3