Merge pull request #1070 from fengzeroz/v2.3

version: 2.3.7

.github/workflows/ci.yml

@@ -8,8 +8,8 @@ on:
       - published
 
 env:
-  VERSION: 2.3.2
-  UI_VERSION: 2.3.1
+  VERSION: 2.3.7
+  UI_VERSION: 2.3.6
   BRANCH: v2.3
 
 jobs:

CMakeLists.txt

@@ -147,6 +147,8 @@ 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)
 

include/neuron/errcodes.h

@@ -106,6 +106,9 @@ typedef enum {
     NEU_ERR_PLUGIN_TAG_NOT_READY           = 3011,
     NEU_ERR_PLUGIN_PACKET_OUT_OF_ORDER     = 3012,
 
+    NEU_ERR_FILE_OPEN_FAILURE = 4101,
+    NEU_ERR_FILE_READ_FAILURE = 4102,
+
     NEU_ERR_MQTT_FAILURE                        = 4000,
     NEU_ERR_MQTT_NO_CERTFILESET                 = 4001,
     NEU_ERR_MQTT_CERTFILE_LOAD_FAILURE          = 4002,

include/neuron/metrics.h

@@ -39,6 +39,16 @@ typedef enum {
     NEU_METRIC_TYPE_COUNTER_SET,
 } neu_metric_type_e;
 
+// node running state
+#define NEU_METRIC_RUNNING_STATE "running_state"
+#define NEU_METRIC_RUNNING_STATE_TYPE NEU_METRIC_TYPE_GAUAGE
+#define NEU_METRIC_RUNNING_STATE_HELP "Node running state"
+
+// node link state
+#define NEU_METRIC_LINK_STATE "link_state"
+#define NEU_METRIC_LINK_STATE_TYPE NEU_METRIC_TYPE_GAUAGE
+#define NEU_METRIC_LINK_STATE_HELP "Node link state"
+
 // last round trip time in millisends
 #define NEU_METRIC_LAST_RTT_MS "last_rtt_ms"
 #define NEU_METRIC_LAST_RTT_MS_TYPE NEU_METRIC_TYPE_GAUAGE

include/neuron/type.h

@@ -94,7 +94,7 @@ inline static const char *neu_type_string(neu_type_e type)
     return "";
 }
 
-#define NEU_VALUE_SIZE 128
+#define NEU_VALUE_SIZE 512
 typedef union {
     bool     boolean;
     int8_t   i8;

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 file_req.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,18 @@
+{
+    "tag_regex": [
+        {
+            "type": 13,
+            "regex": ""
+        }
+    ],
+    "file_length": {
+        "name": "file_length",
+        "description": "",
+        "attribute": "required",
+        "type": "int",
+        "default": 512,
+        "valid": {
+            "min": 1
+        }
+    }
+}

plugins/file/file_plugin.c

@@ -0,0 +1,155 @@
+/**
+ * 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 "file_req.h"
+
+static neu_plugin_t *driver_open(void);
+
+static int driver_close(neu_plugin_t *plugin);
+static int driver_init(neu_plugin_t *plugin);
+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_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group);
+static int driver_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag);
+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.write_tag    = driver_write,
+};
+
+const neu_plugin_module_t neu_plugin_module = {
+    .version         = NEURON_PLUGIN_VER_1_0,
+    .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)
+{
+    plugin->events            = neu_event_new();
+    plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED;
+
+    return 0;
+}
+
+static int driver_uninit(neu_plugin_t *plugin)
+{
+    neu_event_close(plugin->events);
+    plog_info(plugin, "node; file uininit ok");
+
+    return 0;
+}
+
+static int driver_start(neu_plugin_t *plugin)
+{
+    (void) plugin;
+
+    return 0;
+}
+
+static int driver_stop(neu_plugin_t *plugin)
+{
+    (void) plugin;
+
+    return 0;
+}
+
+static int driver_config(neu_plugin_t *plugin, const char *config)
+{
+    char *          err_param   = NULL;
+    neu_json_elem_t file_length = { .name = "file_length", .t = NEU_JSON_INT };
+
+    int ret = neu_parse_param((char *) config, &err_param, 1, &file_length);
+    if (ret != 0) {
+        plog_warn(plugin, "config:%s, decode error: %s", config, err_param);
+        free(err_param);
+        return -1;
+    }
+
+    plugin->file_length = file_length.v.val_int;
+
+    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_validate_tag(neu_plugin_t *plugin, neu_datatag_t *tag)
+{
+    plog_debug(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)
+{
+    return file_group_timer(plugin, group);
+}
+
+static int driver_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
+                        neu_value_u value)
+{
+    return file_write(plugin, req, tag, value);
+}

plugins/file/file_req.c

@@ -0,0 +1,128 @@
+/**
+ * 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 <stdio.h>
+#include <sys/stat.h>
+#include <unistd.h>
+
+#include "file_req.h"
+
+struct file_group_data {
+    UT_array *tags;
+    char *    group;
+};
+
+static void plugin_group_free(neu_plugin_group_t *pgp);
+
+int file_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
+{
+    struct file_group_data *gd     = NULL;
+    int                     length = plugin->file_length;
+
+    if (group->user_data == NULL) {
+        gd                = calloc(1, sizeof(struct file_group_data));
+        group->user_data  = gd;
+        group->group_free = plugin_group_free;
+        utarray_new(gd->tags, &ut_ptr_icd);
+
+        gd->group = strdup(group->group_name);
+    }
+
+    utarray_foreach(group->tags, neu_datatag_t *, tag)
+    {
+        FILE *       f      = NULL;
+        struct stat  st     = { 0 };
+        char *       buf    = calloc(1, length);
+        neu_dvalue_t dvalue = { 0 };
+
+        if (stat(tag->address, &st) != 0) {
+            dvalue.type      = NEU_TYPE_ERROR;
+            dvalue.value.i32 = NEU_ERR_FILE_NOT_EXIST;
+            goto dvalue_result;
+        }
+        if ((f = fopen(tag->address, "rb+")) == NULL) {
+            nlog_error("open file:%s", tag->address);
+            dvalue.type      = NEU_TYPE_ERROR;
+            dvalue.value.i32 = NEU_ERR_PLUGIN_READ_FAILURE;
+            goto dvalue_result;
+        }
+
+        int ret = fread(buf, sizeof(char), length, f);
+        if (ret == 0) {
+            nlog_error("read file failed:%s", tag->address);
+            dvalue.type      = NEU_TYPE_ERROR;
+            dvalue.value.i32 = NEU_ERR_PLUGIN_READ_FAILURE;
+            goto dvalue_result;
+        }
+
+        fseek(f, 0, SEEK_END);
+        int size = ftell(f);
+        if (size > length) {
+            dvalue.type      = NEU_TYPE_ERROR;
+            dvalue.value.i32 = NEU_ERR_PLUGIN_READ_FAILURE;
+            goto dvalue_result;
+        }
+
+        dvalue.type = NEU_TYPE_STRING;
+        strncpy(dvalue.value.str, buf, strlen(buf));
+        fclose(f);
+
+    dvalue_result:
+        plugin->common.adapter_callbacks->driver.update(
+            plugin->common.adapter, group->group_name, tag->name, dvalue);
+
+        free(buf);
+    }
+
+    return 0;
+}
+
+static void plugin_group_free(neu_plugin_group_t *pgp)
+{
+    struct file_group_data *gd = (struct file_group_data *) pgp->user_data;
+
+    utarray_foreach(gd->tags, neu_datatag_t **, tag) { free(*tag); }
+
+    utarray_free(gd->tags);
+    free(gd->group);
+
+    free(gd);
+}
+
+int file_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
+               neu_value_u value)
+{
+    FILE *fp  = fopen(tag->address, "w");
+    int   ret = 0;
+
+    if (0 == fwrite(value.str, sizeof(char), strlen(value.str), fp)) {
+        nlog_error("write file failed:%s", tag->address);
+        ret = NEU_ERR_PLUGIN_WRITE_FAILURE;
+        goto dvalue_result;
+    }
+
+    neu_datatag_string_ltoh(value.str, strlen(value.str));
+
+dvalue_result:
+    fclose(fp);
+    plugin->common.adapter_callbacks->driver.write_response(
+        plugin->common.adapter, req, ret);
+
+    return 0;
+}

plugins/file/file_req.h

@@ -0,0 +1,37 @@
+/**
+ * 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.
+ **/
+
+#ifndef _NEU_M_PLUGIN_file_REQ_H_
+#define _NEU_M_PLUGIN_file_REQ_H_
+
+#include <neuron.h>
+
+struct neu_plugin {
+    neu_plugin_common_t common;
+
+    neu_events_t *events;
+
+    int file_length;
+};
+
+int file_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group);
+int file_write(neu_plugin_t *plugin, void *req, neu_datatag_t *tag,
+               neu_value_u value);
+
+#endif

plugins/restful/http.c

@@ -327,6 +327,8 @@ int http_response(nng_aio *aio, neu_err_code_e code, char *content)
         status = NNG_HTTP_STATUS_BAD_REQUEST;
         break;
     case NEU_ERR_FILE_NOT_EXIST:
+    case NEU_ERR_FILE_OPEN_FAILURE:
+    case NEU_ERR_FILE_READ_FAILURE:
     case NEU_ERR_LIBRARY_NOT_FOUND:
     case NEU_ERR_NODE_NOT_EXIST:
     case NEU_ERR_TAG_NOT_EXIST:

plugins/restful/log_handle.c

@@ -44,7 +44,6 @@
 
 UT_array *collect_log_files();
 
-// void      handle_get_log_files(nng_aio *aio);
 int        read_file(const char *file_name, void **datap, size_t *lenp);
 static int http_resp_files(nng_aio *aio, void *data, size_t len,
                            const char *disposition);
@@ -185,106 +184,6 @@ void handle_log_level(nng_aio *aio)
         })
 }
 
-// void handle_get_log(nng_aio *aio)
-// {
-//     int  rv                               = 0;
-//     char log_file[NEU_NODE_NAME_LEN + 32] = { 0 };
-
-//     VALIDATE_JWT(aio);
-
-//     // optional params
-//     rv = http_get_param_str(aio, "log_file", log_file, sizeof(log_file));
-//     if (-1 == rv) {
-//         nlog_error("parse query param `log_file` fail");
-//         NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
-//             http_response(aio, error_code.error, result_error);
-//         });
-//         return;
-//     } else if (-2 == rv) {
-//         // handle no param
-//         handle_get_log_files(aio);
-//         return;
-//     } else if ((size_t) rv >= sizeof(log_file)) {
-//         nlog_error("query param overflow, `log_file`:%s", log_file);
-//         NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
-//             http_response(aio, error_code.error, result_error);
-//         });
-//         return;
-//     }
-
-//     void *        data = NULL;
-//     size_t        len  = 0;
-//     nng_http_res *res  = NULL;
-//     rv                 = read_log_file(log_file, &data, &len);
-//     if (0 != rv) {
-//         NEU_JSON_RESPONSE_ERROR(
-//             rv, { http_response(aio, error_code.error, result_error); });
-//         return;
-//     }
-
-//     if (((rv = nng_http_res_alloc(&res)) != 0) ||
-//         ((rv = nng_http_res_set_status(res, NNG_HTTP_STATUS_OK)) != 0) ||
-//         ((rv = nng_http_res_set_header(res, "Content-Type",
-//                                        "application/octet-stream")) != 0) ||
-//         ((rv = nng_http_res_set_header(res, "Access-Control-Allow-Origin",
-//                                        "*")) != 0) ||
-//         ((rv = nng_http_res_set_header(res, "Access-Control-Allow-Methods",
-//                                        "POST,GET,PUT,DELETE,OPTIONS")) != 0)
-//                                        ||
-//         ((rv = nng_http_res_set_header(res, "Access-Control-Allow-Headers",
-//                                        "*")) != 0) ||
-//         ((rv = nng_http_res_copy_data(res, data, len)) != 0)) {
-//         nng_http_res_free(res);
-//         free(data);
-//         nng_aio_finish(aio, rv);
-//         return;
-//     }
-
-//     free(data);
-//     nng_aio_set_output(aio, 0, res);
-//     nng_aio_finish(aio, 0);
-// }
-
-// void handle_get_log_files(nng_aio *aio)
-// {
-//     UT_array *log_files = collect_log_files();
-//     if (NULL == log_files) {
-//         NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
-//             http_response(aio, error_code.error, result_error);
-//         });
-//         return;
-//     }
-
-//     size_t len = 1 + strlen("{log_files:[  ]}");
-//     utarray_foreach(log_files, char **, log_file)
-//     {
-//         // one byte for the comma,
-//         // the last one will be for the terminating NULL byte
-//         len += strlen(*log_file) + 1;
-//     }
-
-//     char *res = malloc(len);
-//     if (NULL == res) {
-//         NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
-//             http_response(aio, error_code.error, result_error);
-//         });
-//         utarray_free(log_files);
-//         return;
-//     }
-
-//     // with the space char to simplify code
-//     int i = snprintf(res, len, "{log_files:[ ");
-//     utarray_foreach(log_files, char **, log_file)
-//     {
-//         i += snprintf(res + i, len - i, "%s,", *log_file);
-//     }
-//     // will write over the last comma/space
-//     snprintf(res + i - 1, len, "]}");
-//     http_ok(aio, res);
-//     free(res);
-//     utarray_free(log_files);
-// }
-
 static int http_resp_files(nng_aio *aio, void *data, size_t len,
                            const char *disposition)
 {
@@ -365,7 +264,7 @@ int read_file(const char *file_name, void **datap, size_t *lenp)
 
     if ((f = fopen(file_name, "rb")) == NULL) {
         nlog_error("open fail: %s", file_name);
-        return NEU_ERR_EINTERNAL;
+        return NEU_ERR_FILE_OPEN_FAILURE;
     }
 
     len = st.st_size;
@@ -375,8 +274,8 @@ int read_file(const char *file_name, void **datap, size_t *lenp)
             goto done;
         }
         if (fread(data, 1, len, f) != len) {
-            rv = errno;
-            nlog_info("rv = %d", rv);
+            nlog_error("file read failued, errno = %d", errno);
+            rv = NEU_ERR_FILE_READ_FAILURE;
             free(data);
             goto done;
         }

src/adapter/adapter.c

@@ -67,6 +67,10 @@ static const adapter_callbacks_t callback_funs = {
     adapter_register_metric(adapter, name, name##_HELP, name##_TYPE, init);
 
 #define REGISTER_DRIVER_METRICS(adapter)                     \
+    REGISTER_METRIC(adapter, NEU_METRIC_LINK_STATE,          \
+                    NEU_NODE_LINK_STATE_DISCONNECTED);       \
+    REGISTER_METRIC(adapter, NEU_METRIC_RUNNING_STATE,       \
+                    NEU_NODE_RUNNING_STATE_INIT);            \
     REGISTER_METRIC(adapter, NEU_METRIC_LAST_RTT_MS,         \
                     NEU_METRIC_LAST_RTT_MS_MAX);             \
     REGISTER_METRIC(adapter, NEU_METRIC_SEND_BYTES, 0);      \
@@ -75,6 +79,10 @@ static const adapter_callbacks_t callback_funs = {
     REGISTER_METRIC(adapter, NEU_METRIC_TAG_READ_ERRORS_TOTAL, 0);
 
 #define REGISTER_APP_METRICS(adapter)                              \
+    REGISTER_METRIC(adapter, NEU_METRIC_LINK_STATE,                \
+                    NEU_NODE_LINK_STATE_DISCONNECTED);             \
+    REGISTER_METRIC(adapter, NEU_METRIC_RUNNING_STATE,             \
+                    NEU_NODE_RUNNING_STATE_INIT);                  \
     REGISTER_METRIC(adapter, NEU_METRIC_SEND_MSGS_TOTAL, 0);       \
     REGISTER_METRIC(adapter, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 0); \
     REGISTER_METRIC(adapter, NEU_METRIC_RECV_MSGS_TOTAL, 0);

src/adapter/metrics.c

@@ -163,6 +163,11 @@ void neu_metrics_visist(neu_metrics_cb_t cb, void *data)
         neu_plugin_common_t *common =
             neu_plugin_to_plugin_common(n->adapter->plugin);
 
+        n->adapter->cb_funs.update_metric(n->adapter, NEU_METRIC_RUNNING_STATE,
+                                          n->adapter->state, NULL);
+        n->adapter->cb_funs.update_metric(n->adapter, NEU_METRIC_LINK_STATE,
+                                          common->link_state, NULL);
+
         if (NEU_NA_TYPE_DRIVER == n->adapter->module->type) {
             ++g_metrics_.south_nodes;
             if (NEU_NODE_RUNNING_STATE_RUNNING == n->adapter->state) {

src/connection/connection.c

@@ -842,7 +842,7 @@ static void conn_tcp_server_add_client(neu_conn_t *conn, int fd,
 static void conn_tcp_server_del_client(neu_conn_t *conn, int fd)
 {
     for (int i = 0; i < conn->param.params.tcp_server.max_link; i++) {
-        if (conn->tcp_server.clients[i].fd == fd) {
+        if (fd > 0 && conn->tcp_server.clients[i].fd == fd) {
             close(fd);
             conn->tcp_server.clients[i].fd = 0;
             conn->tcp_server.n_client -= 1;

version

@@ -1 +1 @@
-2.3.2
+2.3.7