Merge pull request #1292 from fengzeroz/v2.3

version: 2.3.13

.github/workflows/ci.yml

@@ -8,9 +8,9 @@ on:
       - published
 
 env:
-  VERSION: 2.3.0
-  UI_VERSION: 2.3
-  BRANCH: main
+  VERSION: 2.3.13
+  UI_VERSION: 2.3.9
+  BRANCH: v2.3
 
 jobs:
   cross-build:
@@ -58,54 +58,6 @@ jobs:
           path: |
             neuron-sdk-*.tar.gz
 
-  cross-build-old:
-    runs-on: ubuntu-18.04
-    strategy:
-      matrix:
-        arch:
-          - [aarch64-linux-gnu, aarch64, arm64]
-          - [arm-linux-gnueabihf, armv4, armhf]
-          - [x86_64-linux-gnu, x86_64, amd64]
-    container: neugates/build_old:${{ matrix.arch[1] }}-2.3
-
-    steps:
-      - uses: actions/checkout@v3
-
-      - uses: actions/checkout@v3
-        with:
-          repository: troydhanson/uthash
-          path: extern/uthash
-
-      - name: compile
-        run: |
-          git config --global --add safe.directory $(pwd)
-          mkdir -p build && cd build
-          cmake .. -DCMAKE_TOOLCHAIN_FILE=../cmake/${{ matrix.arch[0] }}.cmake \
-            -DCMAKE_BUILD_TYPE=Release -DDISABLE_UT=ON
-          make -j4
-
-      - name: fetch dashboard
-        uses: dsaltares/fetch-gh-release-asset@master
-        with:
-          repo: "emqx/neuron-dashboard-src"
-          version: "tags/${{ env.UI_VERSION }}"
-          file: "neuron-dashboard.zip"
-          target: "neuron-dashboard.zip"
-
-      - name: unpack dashboard
-        run: |
-          unzip neuron-dashboard.zip -d ./build
-
-      - name: Create neuron sdk package
-        run: |
-          ./package-sdk.sh -p ${{ matrix.arch[2] }} -n neuron-sdk-old-$VERSION
-
-      - uses: actions/upload-artifact@v3
-        with:
-          name: neuron-sdk-old-${{ matrix.arch[2] }}
-          path: |
-            neuron-sdk-old-*.tar.gz
-
   build-debug-sdk:
     runs-on: ubuntu-20.04
     strategy:
@@ -144,7 +96,7 @@ jobs:
 
     steps:
       - name: delete old daily-pre-release
-        uses: dev-drprasad/delete-tag-and-release@v0.2.0
+        uses: dev-drprasad/delete-tag-and-release@v0.2.1
         with:
           delete_release: true
           tag_name: ${{ env.BRANCH }}-daily
@@ -178,7 +130,7 @@ jobs:
     strategy:
       matrix:
         arch: [arm64, armhf, amd64]
-        type: [neuron-sdk, neuron-sdk-old]
+        type: [neuron-sdk]
         package: [tar.gz]
 
     steps:
@@ -306,4 +258,4 @@ jobs:
           upload_url: ${{ github.event.release.upload_url }}
           asset_path: ./${{ steps.package.outputs.name}}
           asset_name: ${{ matrix.type }}-${{ env.VERSION }}-linux-${{ matrix.arch }}.${{ matrix.package }}
-          asset_content_type: application/octet-stream
+          asset_content_type: application/octet-stream

CMakeLists.txt

@@ -59,6 +59,9 @@ else()
   link_directories(/usr/local/lib)
 endif()
 
+set(PERSIST_SOURCES
+    src/persist/persist.c
+    src/persist/json/persist_json_plugin.c)
 aux_source_directory(src/parser NEURON_SRC_PARSE)
 set(NEURON_BASE_SOURCES
     src/base/tag.c
@@ -74,7 +77,8 @@ set(NEURON_BASE_SOURCES
     src/utils/neu_jwt.c
     src/utils/base64.c
     src/utils/async_queue.c
-    src/utils/mem_cache.c)
+    src/utils/mem_cache.c
+    ${PERSIST_SOURCES})
 
 add_library(neuron-base SHARED)
 target_sources(neuron-base PRIVATE ${NEURON_BASE_SOURCES} ${NEURON_SRC_PARSE}) 
@@ -89,9 +93,6 @@ add_dependencies(neuron-base neuron-version)
 #find_package(MbedTLS)
 target_link_libraries(neuron-base mbedtls mbedx509 mbedcrypto)
 
-set(PERSIST_SOURCES
-    src/persist/persist.c
-    src/persist/json/persist_json_plugin.c)
 set(NEURON_SOURCES
     src/main.c
     src/argparse.c
@@ -123,8 +124,7 @@ set(NEURON_SOURCES
     plugins/restful/proxy.c
     plugins/restful/websocket.c
     plugins/monitor/monitor.c
-    plugins/monitor/metric_handle.c
-    ${PERSIST_SOURCES})
+    plugins/monitor/metric_handle.c)
 
 set(CMAKE_BUILD_RPATH ./)
 add_executable(neuron)
@@ -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)
 

cmake/aarch64-linux-gnu.cmake

@@ -1,7 +1,7 @@
 set(CMAKE_SYSTEM_NAME Linux)
 set(COMPILER_PREFIX aarch64-linux-gnu)
 set(CMAKE_SYSTEM_PROCESSOR aarch64)
-set(LIBRARY_DIR /opt/externs/libs)
+set(LIBRARY_DIR /home/neuron/libs)
 
 set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
 set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)
@@ -20,3 +20,5 @@ set(CMAKE_FIND_ROOT_PATH_MODE_PROGRAM NEVER)
 set(CMAKE_FIND_ROOT_PATH_MODE_LIBRARY ONLY)
 set(CMAKE_FIND_ROOT_PATH_MODE_INCLUDE ONLY)
 link_directories(${CMAKE_STAGING_PREFIX})
+
+file(COPY ${CMAKE_STAGING_PREFIX}/lib/libzlog.so.1.2 DESTINATION /usr/local/lib)

cmake/arm-linux-gnueabi.cmake

@@ -1,7 +1,7 @@
 set(CMAKE_SYSTEM_NAME Linux)
 set(COMPILER_PREFIX arm-linux-gnueabi)
 set(CMAKE_SYSTEM_PROCESSOR armv5tejl)
-set(LIBRARY_DIR /opt/externs/libs)
+set(LIBRARY_DIR /home/neuron/libs)
 
 set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
 set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)
@@ -20,3 +20,5 @@ set(CMAKE_FIND_ROOT_PATH_MODE_PROGRAM NEVER)
 set(CMAKE_FIND_ROOT_PATH_MODE_LIBRARY ONLY)
 set(CMAKE_FIND_ROOT_PATH_MODE_INCLUDE ONLY)
 link_directories(${CMAKE_STAGING_PREFIX})
+
+file(COPY ${CMAKE_STAGING_PREFIX}/lib/libzlog.so.1.2 DESTINATION /usr/local/lib)

cmake/arm-linux-gnueabihf.cmake

@@ -1,7 +1,7 @@
 set(CMAKE_SYSTEM_NAME Linux)
 set(COMPILER_PREFIX arm-linux-gnueabihf)
 set(CMAKE_SYSTEM_PROCESSOR armv7l)
-set(LIBRARY_DIR /opt/externs/libs)
+set(LIBRARY_DIR /home/neuron/libs)
 
 set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
 set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)
@@ -20,3 +20,5 @@ set(CMAKE_FIND_ROOT_PATH_MODE_PROGRAM NEVER)
 set(CMAKE_FIND_ROOT_PATH_MODE_LIBRARY ONLY)
 set(CMAKE_FIND_ROOT_PATH_MODE_INCLUDE ONLY)
 link_directories(${CMAKE_STAGING_PREFIX})
+
+file(COPY ${CMAKE_STAGING_PREFIX}/lib/libzlog.so.1.2 DESTINATION /usr/local/lib)

cmake/x86_64-linux-gnu.cmake

@@ -1,7 +1,24 @@
 set(CMAKE_SYSTEM_NAME Linux)
 set(COMPILER_PREFIX x86_64-linux-gnu)
 set(CMAKE_SYSTEM_PROCESSOR x86_64)
-set(LIBRARY_DIR /usr/local)
+set(LIBRARY_DIR /home/neuron/libs)
 
+set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
+set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)
+set(CMAKE_AR ${COMPILER_PREFIX}-ar)
+set(CMAKE_LINKER ${COMPILER_PREFIX}-ld)
+set(CMAKE_NM ${COMPILER_PREFIX}-nm)
+set(CMAKE_OBJDUMP ${COMPILER_PREFIX}-objdump)
+set(CMAKE_RANLIB ${COMPILER_PREFIX}-ranlib)
 set(CMAKE_STAGING_PREFIX ${LIBRARY_DIR}/${COMPILER_PREFIX})
-set(CMAKE_PREFIX_PATH ${CMAKE_STAGING_PREFIX})
+set(CMAKE_PREFIX_PATH ${CMAKE_STAGING_PREFIX})
+
+include_directories(SYSTEM ${CMAKE_STAGING_PREFIX}/include)
+include_directories(SYSTEM ${CMAKE_STAGING_PREFIX}/openssl/include)
+set(CMAKE_FIND_ROOT_PATH ${CMAKE_STAGING_PREFIX})
+set(CMAKE_FIND_ROOT_PATH_MODE_PROGRAM NEVER)
+set(CMAKE_FIND_ROOT_PATH_MODE_LIBRARY ONLY)
+set(CMAKE_FIND_ROOT_PATH_MODE_INCLUDE ONLY)
+link_directories(${CMAKE_STAGING_PREFIX})
+
+file(COPY ${CMAKE_STAGING_PREFIX}/lib/libzlog.so.1.2 DESTINATION /usr/local/lib)

include/neuron/adapter.h

@@ -414,6 +414,7 @@ typedef struct {
 typedef struct {
     char                 driver[NEU_NODE_NAME_LEN];
     char                 group[NEU_GROUP_NAME_LEN];
+    int64_t              timestamp;
     uint16_t             n_tag;
     neu_resp_tag_value_t tags[];
 } neu_reqresp_trans_data_t;

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/neuron.h

@@ -27,6 +27,7 @@ extern "C" {
 #include <stdbool.h>
 #include <stdint.h>
 #include <stdlib.h>
+#include <string.h>
 
 #include "errcodes.h"
 

include/neuron/persist/persist.h

@@ -24,7 +24,9 @@
 extern "C" {
 #endif
 
-#include "adapter/adapter_info.h"
+#include <sqlite3.h>
+
+#include "adapter_info.h"
 #include "persist/json/persist_json_plugin.h"
 
 typedef neu_json_plugin_req_plugin_t neu_persist_plugin_info_t;
@@ -84,12 +86,14 @@ static inline void neu_persist_user_info_fini(neu_persist_user_info_t *info)
  * Create persister.
  * @return 0 on success, -1 otherwise.
  */
-int neu_persister_create();
+int neu_persister_create(const char *schema_dir);
 /**
  * Destroy perister.
  */
 void neu_persister_destroy();
 
+sqlite3 *neu_persister_get_db();
+
 /**
  * Persist nodes.
  * @param node_info                 neu_persist_node_info_t.
@@ -306,4 +310,4 @@ int neu_persister_delete_user(const char *user_name);
 }
 #endif
 
-#endif
+#endif

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;

package-sdk.sh

@@ -44,7 +44,7 @@ cp -r build/dist ${package_name}/
 
 cp build/neuron ${package_name}
 cp build/libneuron-base.so ${package_name}/lib
-cp /library/zlog/src/libzlog.so.1.2 ${package_name}/lib
+cp /usr/local/lib/libzlog.so.1.2 ${package_name}/lib
 
 cp persistence/*.sql ${package_name}/config/
 

persistence/0001_2.3.0_users.sql

@@ -32,15 +32,36 @@ VALUES
     "$5$PwFeXpBBIBZuZdZl$fP8fFPWCLoaWcnVXVSR.3Xi8TEqCvX92gjhowNNn6S4"
   );
 
---- Modify tags table ---
 BEGIN TRANSACTION;
+
+ALTER TABLE nodes RENAME TO old_nodes;
+CREATE TABLE
+  nodes (
+    name TEXT PRIMARY KEY check(length(name) <= 128),
+    type integer(1) NOT NULL check(type IN (1, 2)),
+    state integer(1) NOT NULL check(state BETWEEN 1 AND 4),
+    plugin_name TEXT NOT NULL check(length(plugin_name) <= 32)
+  );
+INSERT INTO nodes SELECT * FROM old_nodes;
+
+ALTER TABLE groups RENAME TO old_groups;
+CREATE TABLE IF NOT EXISTS
+  groups (
+    driver_name TEXT NOT NULL,
+    name TEXT NULL check(length(name) <= 128),
+    interval INTEGER NOT NULL check(interval >= 100),
+    UNIQUE (driver_name, name),
+    FOREIGN KEY (driver_name) REFERENCES nodes (name) ON UPDATE CASCADE ON DELETE CASCADE
+  );
+INSERT INTO groups SELECT * FROM old_groups;
+
 ALTER TABLE tags RENAME TO old_tags;
 CREATE TABLE IF NOT EXISTS
   tags (
     driver_name TEXT NOT NULL,
     group_name TEXT NOT NULL,
-    name TEXT NULL check(length(name) <= 32),
-    address TEXT NULL check(length(address) <= 64),
+    name TEXT NULL check(length(name) <= 64),
+    address TEXT NULL check(length(address) <= 128),
     attribute INTEGER NOT NULL check(attribute BETWEEN 0 AND 7),
     precision INTEGER NOT NULL check(precision BETWEEN 0 AND 17),
     decimal REAL NOT NULL,
@@ -50,5 +71,33 @@ CREATE TABLE IF NOT EXISTS
     FOREIGN KEY (driver_name, group_name) REFERENCES groups (driver_name, name) ON UPDATE CASCADE ON DELETE CASCADE
   );
 INSERT INTO tags SELECT * FROM old_tags;
-DROP TABLE old_tags;
+
+ALTER TABLE subscriptions RENAME TO old_subscriptions;
+CREATE TABLE IF NOT EXISTS
+  subscriptions (
+    app_name TEXT NOT NULL,
+    driver_name TEXT NOT NULL,
+    group_name TEXT NOT NULL,
+    CHECK (app_name != driver_name),
+    UNIQUE (app_name, driver_name, group_name),
+    FOREIGN KEY (app_name) REFERENCES nodes (name) ON UPDATE CASCADE ON DELETE CASCADE,
+    FOREIGN KEY (driver_name, group_name) REFERENCES groups (driver_name, name) ON UPDATE CASCADE ON DELETE CASCADE
+  );
+INSERT INTO subscriptions SELECT * FROM old_subscriptions;
+
+ALTER TABLE settings RENAME TO old_settings;
+CREATE TABLE IF NOT EXISTS
+  settings (
+    node_name TEXT NOT NULL,
+    setting TEXT NOT NULL,
+    UNIQUE (node_name),
+    FOREIGN KEY (node_name) REFERENCES nodes (name) ON UPDATE CASCADE ON DELETE CASCADE
+  );
+INSERT INTO settings SELECT * FROM old_settings;
+
+CREATE TABLE IF NOT EXISTS
+  node_cache (
+    node_name TEXT NOT NULL,
+    cache TEXT NOT NULL
+  );
 COMMIT;

plugins/ekuiper/json_rw.c

@@ -201,7 +201,7 @@ int json_encode_read_resp(void *json_object, void *param)
 
     json_read_resp_header_t header = { .group_name = trans_data->group,
                                        .node_name  = trans_data->driver,
-                                       .timestamp  = global_timestamp };
+                                       .timestamp  = trans_data->timestamp };
 
     ret = json_encode_read_resp_header(json_object, &header);
     if (0 != ret) {

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,136 @@
+/**
+ * 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)
+{
+    neu_adapter_update_metric_cb_t update_metric =
+        plugin->common.adapter_callbacks->update_metric;
+    int64_t rtt = NEU_METRIC_LAST_RTT_MS_MAX;
+
+    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);
+
+        rtt = 1;
+
+    dvalue_result:
+        plugin->common.adapter_callbacks->driver.update(
+            plugin->common.adapter, group->group_name, tag->name, dvalue);
+
+        free(buf);
+    }
+
+    update_metric(plugin->common.adapter, NEU_METRIC_LAST_RTT_MS, rtt, NULL);
+
+    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/monitor/metric_handle.c

@@ -242,7 +242,6 @@ void handle_get_metric(nng_aio *aio)
     FILE *        stream = NULL;
     neu_plugin_t *plugin = neu_monitor_get_plugin();
 
-    VALIDATE_JWT(aio);
     neu_metrics_category_e cat           = NEU_METRICS_CATEGORY_ALL;
     size_t                 cat_param_len = 0;
     const char *cat_param = http_get_param(aio, "category", &cat_param_len);

plugins/mqtt/mqtt.json

@@ -19,16 +19,6 @@
       "length": 255
     }
   },
-  "heartbeat-topic": {
-    "name": "heartbeat topic",
-    "description": "Topic to which publish heartbeat messages.",
-    "type": "string",
-    "attribute": "required",
-    "default": "/neuron/${node-name}/heartbeat",
-    "valid": {
-      "length": 255
-    }
-  },
   "format": {
     "name": "upload format",
     "description": "JSON format of the data reported in the upload topic. In values mode, data are split into `values` and `errors` sub objects. In tags mode, tag data are put in a single array.",

plugins/mqtt/mqtt_config.c

@@ -122,13 +122,11 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
     char *      err_param   = NULL;
     const char *placeholder = "********";
 
-    neu_json_elem_t client_id    = { .name = "client-id", .t = NEU_JSON_STR };
-    neu_json_elem_t upload_topic = { .name = "upload-topic",
+    neu_json_elem_t client_id      = { .name = "client-id", .t = NEU_JSON_STR };
+    neu_json_elem_t upload_topic   = { .name = "upload-topic",
                                      .t    = NEU_JSON_STR };
-    neu_json_elem_t heartbeat_topic = { .name = "heartbeat-topic",
-                                        .t    = NEU_JSON_STR };
-    neu_json_elem_t format          = { .name = "format", .t = NEU_JSON_INT };
-    neu_json_elem_t cache_mem_size  = { .name = "cache-mem-size",
+    neu_json_elem_t format         = { .name = "format", .t = NEU_JSON_INT };
+    neu_json_elem_t cache_mem_size = { .name = "cache-mem-size",
                                        .t    = NEU_JSON_INT };
     neu_json_elem_t cache_disk_size = { .name = "cache-disk-size",
                                         .t    = NEU_JSON_INT };
@@ -147,9 +145,9 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
         return -1;
     }
 
-    ret = neu_parse_param(setting, &err_param, 8, &client_id, &upload_topic,
-                          &heartbeat_topic, &format, &cache_mem_size,
-                          &cache_disk_size, &host, &port);
+    ret = neu_parse_param(setting, &err_param, 7, &client_id, &upload_topic,
+                          &format, &cache_mem_size, &cache_disk_size, &host,
+                          &port);
     if (0 != ret) {
         plog_error(plugin, "parsing setting fail, key: `%s`", err_param);
         goto error;
@@ -167,12 +165,6 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
         goto error;
     }
 
-    // heartbeat-topic, required
-    if (0 == strlen(heartbeat_topic.v.val_str)) {
-        plog_error(plugin, "setting empty heartbeat-topic");
-        goto error;
-    }
-
     // format, required
     if (MQTT_UPLOAD_FORMAT_VALUES != format.v.val_int &&
         MQTT_UPLOAD_FORMAT_TAGS != format.v.val_int) {
@@ -239,10 +231,9 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
         goto error;
     }
 
-    config->client_id       = client_id.v.val_str;
-    config->upload_topic    = upload_topic.v.val_str;
-    config->heartbeat_topic = heartbeat_topic.v.val_str;
-    config->format          = format.v.val_int;
+    config->client_id    = client_id.v.val_str;
+    config->upload_topic = upload_topic.v.val_str;
+    config->format       = format.v.val_int;
     config->cache =
         0 != cache_mem_size.v.val_int || 0 != cache_disk_size.v.val_int;
     config->cache_mem_size  = cache_mem_size.v.val_int * MB;
@@ -258,7 +249,6 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
 
     plog_info(plugin, "config client-id      : %s", config->client_id);
     plog_info(plugin, "config upload-topic   : %s", config->upload_topic);
-    plog_info(plugin, "config heartbeat-topic: %s", config->heartbeat_topic);
     plog_info(plugin, "config format         : %s",
               mqtt_upload_format_str(config->format));
     plog_info(plugin, "config cache          : %zu", config->cache);
@@ -292,7 +282,6 @@ int mqtt_config_parse(neu_plugin_t *plugin, const char *setting,
 error:
     free(err_param);
     free(upload_topic.v.val_str);
-    free(heartbeat_topic.v.val_str);
     free(host.v.val_str);
     free(username.v.val_str);
     free(password.v.val_str);
@@ -307,7 +296,6 @@ void mqtt_config_fini(mqtt_config_t *config)
 {
     free(config->client_id);
     free(config->upload_topic);
-    free(config->heartbeat_topic);
     free(config->host);
     free(config->username);
     free(config->password);

plugins/mqtt/mqtt_config.h

@@ -49,7 +49,6 @@ static inline const char *mqtt_upload_format_str(mqtt_upload_format_e f)
 typedef struct {
     char *               client_id;       // client id
     char *               upload_topic;    // upload topic
-    char *               heartbeat_topic; // heartbeat topic
     mqtt_upload_format_e format;          // upload format
     size_t               cache;           // cache enable flag
     size_t               cache_mem_size;  // cache memory size in bytes

plugins/mqtt/mqtt_handle.c

@@ -26,36 +26,6 @@
 #include "mqtt_handle.h"
 #include "mqtt_plugin.h"
 
-static char *generate_heartbeat_json(neu_plugin_t *plugin, UT_array *states)
-{
-    (void) plugin;
-    char *                 version  = NEURON_VERSION;
-    neu_json_states_head_t header   = { .version   = version,
-                                      .timpstamp = global_timestamp };
-    neu_json_states_t      json     = { 0 };
-    char *                 json_str = NULL;
-
-    json.n_state = utarray_len(states);
-    json.states  = calloc(json.n_state, sizeof(neu_json_node_state_t));
-    if (NULL == json.states) {
-        return NULL;
-    }
-
-    utarray_foreach(states, neu_nodes_state_t *, state)
-    {
-        int index                  = utarray_eltidx(states, state);
-        json.states[index].node    = state->node;
-        json.states[index].link    = state->state.link;
-        json.states[index].running = state->state.running;
-    }
-
-    neu_json_encode_with_mqtt(&json, neu_json_encode_states_resp, &header,
-                              neu_json_encode_state_header_resp, &json_str);
-
-    free(json.states);
-    return json_str;
-}
-
 static int tag_values_to_json(neu_resp_tag_value_t *tags, uint16_t len,
                               neu_json_read_resp_t *json)
 {
@@ -524,38 +494,3 @@ int handle_trans_data(neu_plugin_t *            plugin,
 
     return rv;
 }
-
-int handle_nodes_state(neu_plugin_t *plugin, neu_reqresp_nodes_state_t *states)
-{
-    int   rv       = 0;
-    char *json_str = NULL;
-
-    if (NULL == plugin->client) {
-        rv = NEU_ERR_MQTT_IS_NULL;
-        goto end;
-    }
-
-    if (0 == plugin->config.cache &&
-        !neu_mqtt_client_is_connected(plugin->client)) {
-        // cache disable and we are disconnected
-        rv = NEU_ERR_MQTT_FAILURE;
-        goto end;
-    }
-
-    json_str = generate_heartbeat_json(plugin, states->states);
-    if (NULL == json_str) {
-        plog_error(plugin, "generate heartbeat json fail");
-        rv = NEU_ERR_EINTERNAL;
-        goto end;
-    }
-
-    char *         topic = plugin->config.heartbeat_topic;
-    neu_mqtt_qos_e qos   = NEU_MQTT_QOS0;
-    rv       = publish(plugin, qos, topic, json_str, strlen(json_str));
-    json_str = NULL;
-
-end:
-    utarray_free(states->states);
-
-    return rv;
-}

plugins/mqtt/mqtt_handle.h

@@ -42,8 +42,6 @@ int handle_read_response(neu_plugin_t *plugin, neu_json_mqtt_t *mqtt_json,
 int handle_trans_data(neu_plugin_t *            plugin,
                       neu_reqresp_trans_data_t *trans_data);
 
-int handle_nodes_state(neu_plugin_t *plugin, neu_reqresp_nodes_state_t *states);
-
 #ifdef __cplusplus
 }
 #endif

plugins/mqtt/mqtt_plugin.c

@@ -170,7 +170,14 @@ static int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
         return NEU_ERR_NODE_SETTING_INVALID;
     }
 
-    if (plugin->client && neu_mqtt_client_is_open(plugin->client)) {
+    if (NULL == plugin->client) {
+        plugin->client = neu_mqtt_client_new(NEU_MQTT_VERSION_V311);
+        if (NULL == plugin->client) {
+            plog_error(plugin, "neu_mqtt_client_new fail");
+            rv = NEU_ERR_EINTERNAL;
+            goto error;
+        }
+    } else if (neu_mqtt_client_is_open(plugin->client)) {
         started = true;
         rv      = neu_mqtt_client_close(plugin->client);
         if (0 != rv) {
@@ -178,13 +185,6 @@ static int mqtt_plugin_config(neu_plugin_t *plugin, const char *setting)
             rv = NEU_ERR_EINTERNAL;
             goto error;
         }
-    } else {
-        plugin->client = neu_mqtt_client_new(NEU_MQTT_VERSION_V311);
-        if (NULL == plugin->client) {
-            plog_error(plugin, "neu_mqtt_client_new fail");
-            rv = NEU_ERR_EINTERNAL;
-            goto error;
-        }
     }
 
     rv = config_mqtt_client(plugin, plugin->client, &config);
@@ -349,10 +349,6 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
     case NEU_REQRESP_TRANS_DATA:
         error = handle_trans_data(plugin, data);
         break;
-    case NEU_REQRESP_NODES_STATE: {
-        error = handle_nodes_state(plugin, data);
-        break;
-    }
     case NEU_REQRESP_NODE_DELETED:
         break;
     default:
@@ -385,7 +381,7 @@ const neu_plugin_module_t neu_plugin_module = {
     .intf_funs       = &plugin_intf_funs,
     .kind            = NEU_PLUGIN_KIND_SYSTEM,
     .type            = NEU_NA_TYPE_APP,
-    .sub_msg[0]      = NEU_SUBSCRIBE_NODES_STATE,
+    .sub_msg         = { 0 },
     .display         = true,
     .single          = false,
 };

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,32 +44,51 @@
 
 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);
 
 void handle_logs_files(nng_aio *aio)
 {
-    char   path[128] = { 0 };
-    char   disp[128] = { 0 };
-    void * data      = NULL;
-    size_t len       = 0;
-    int    rv        = 0;
+    void * data = NULL;
+    size_t len  = 0;
+    int    rv   = 0;
 
     VALIDATE_JWT(aio);
 
-    rv = mkdir("neuron_logs", S_IRWXU);
-    if (0 != rv && EEXIST == errno) {
-        rv = 0;
+    /* check whether the neuron directory exists */
+    rv = access("../neuron", F_OK);
+    if (0 != rv) {
+        nlog_error("The neuron directory does not exists");
+        NEU_JSON_RESPONSE_ERROR(NEU_ERR_FILE_NOT_EXIST, {
+            http_response(aio, error_code.error, result_error);
+        });
+        return;
     }
 
-    // check if coredump exists
-    rv = access("./core", F_OK);
-    if (0 == rv) {
-        rv = system("cp -r ./core neuron_logs");
-        if (0 != rv) {
-            nlog_error("failed to copy coredump file");
+    rv = system("rm -rf /tmp/neuron_debug.tar.gz");
+    nlog_warn("remove old neuron_debug.tar.gz, rv: %d", rv);
+    /* tar the neuron directory */
+    rv = system("tar -zcvf /tmp/neuron_debug.tar.gz ../neuron");
+    if (rv == -1) {
+        nlog_error("failed to create neuron_debug.tar.gz, rv: %d", rv);
+        NEU_JSON_RESPONSE_ERROR(NEU_ERR_COMMAND_EXECUTION_FAILED, {
+            http_response(aio, error_code.error, result_error);
+        });
+        return;
+    } else {
+        if (WIFEXITED(rv)) {
+            if (WEXITSTATUS(rv) < 0) {
+                nlog_error("failed to create neuron_debug.tar.gz, rv: %d",
+                           WEXITSTATUS(rv));
+                NEU_JSON_RESPONSE_ERROR(NEU_ERR_COMMAND_EXECUTION_FAILED, {
+                    http_response(aio, error_code.error, result_error);
+                });
+                return;
+            }
+        } else {
+            nlog_error("failed to create neuron_debug.tar.gz, rv: %d",
+                       WEXITSTATUS(rv));
             NEU_JSON_RESPONSE_ERROR(NEU_ERR_COMMAND_EXECUTION_FAILED, {
                 http_response(aio, error_code.error, result_error);
             });
@@ -77,48 +96,18 @@ void handle_logs_files(nng_aio *aio)
         }
     }
 
-    rv = system("cp -r ./logs neuron_logs");
-    if (0 != rv) {
-        nlog_error("failed to copy logs file");
-        NEU_JSON_RESPONSE_ERROR(NEU_ERR_COMMAND_EXECUTION_FAILED, {
-            http_response(aio, error_code.error, result_error);
-        });
-        return;
-    }
-
-    rv = system("tar -zcvf neuron_logs.tar.gz neuron_logs");
-    if (0 != rv) {
-        nlog_error("failed to create neuron_logs.tar.gz");
-        NEU_JSON_RESPONSE_ERROR(NEU_ERR_COMMAND_EXECUTION_FAILED, {
-            http_response(aio, error_code.error, result_error);
-        });
-        return;
-    }
-
-    strcpy(path, "neuron_logs.tar.gz");
-    rv = read_file(path, &data, &len);
+    rv = read_file("/tmp/neuron_debug.tar.gz", &data, &len);
     if (0 != rv) {
         NEU_JSON_RESPONSE_ERROR(
             rv, { http_response(aio, error_code.error, result_error); });
         return;
     }
 
-    // delete neuron_logs folder from build
-    rv = system("rm -rf neuron_logs.tar.gz neuron_logs");
-    if (0 != rv) {
-        nlog_error("failed to delete neuron_logs folder");
-        NEU_JSON_RESPONSE_ERROR(NEU_ERR_COMMAND_EXECUTION_FAILED, {
-            http_response(aio, error_code.error, result_error);
-        });
-        return;
-    }
+    /* handle http response */
+    rv = http_resp_files(aio, data, len,
+                         "attachment; filename=neuron_debug.tar.gz");
 
-    // handle http response
-    strcpy(disp, "attachment; filename=neuron_logs.tar.gz");
-    rv = http_resp_files(aio, data, len, disp);
-    if (0 != rv) {
-        return;
-    }
+    nlog_notice("download neuron_debug.tar.gz, ret: %d", rv);
 }
 
 void handle_log_level(nng_aio *aio)
@@ -185,106 +174,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 +254,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 +264,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/driver/driver.c

@@ -299,6 +299,8 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
                         (double) (cmd->value.value.d64 / tag->decimal);
                 } else if (cmd->value.type == NEU_TYPE_DOUBLE) {
                     cmd->value.value.d64 = cmd->value.value.d64 / tag->decimal;
+                } else if (cmd->value.type == NEU_TYPE_FLOAT) {
+                    cmd->value.value.f32 = cmd->value.value.f32 / tag->decimal;
                 }
                 break;
             default:
@@ -357,8 +359,11 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
             }
             break;
         case NEU_TYPE_FLOAT:
-            cmd->value.type      = NEU_TYPE_FLOAT;
-            cmd->value.value.f32 = (float) cmd->value.value.d64;
+            if (NEU_TYPE_FLOAT != cmd->value.type) {
+                cmd->value.value.f32 = (float) cmd->value.value.d64;
+                cmd->value.type      = NEU_TYPE_FLOAT;
+            }
+
             switch (tag->option.value32.endian) {
             case NEU_DATATAG_ENDIAN_LB32:
                 neu_ntohs_p((uint16_t *) cmd->value.value.bytes);
@@ -422,7 +427,7 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
             .second      = interval / 1000,
             .millisecond = interval % 1000,
             .usr_data    = (void *) find,
-            .type        = NEU_EVENT_TIMER_BLOCK,
+            .type        = NEU_EVENT_TIMER_NOBLOCK,
         };
 
         find->driver         = driver;
@@ -467,13 +472,14 @@ int neu_adapter_driver_update_group(neu_adapter_driver_t *driver,
             .second      = interval / 1000,
             .millisecond = interval % 1000,
             .usr_data    = (void *) find,
-            .type        = NEU_EVENT_TIMER_BLOCK,
+            .type        = NEU_EVENT_TIMER_NOBLOCK,
         };
 
         neu_group_set_interval(find->group, interval);
 
         param.cb     = report_callback;
         find->report = neu_adapter_add_timer((neu_adapter_t *) driver, param);
+        param.type   = driver->adapter.module->timer_type;
         param.cb     = read_callback;
         find->read   = neu_event_add_timer(driver->driver_events, param);
 
@@ -769,6 +775,7 @@ static int report_callback(void *usr_data)
                sizeof(neu_reqresp_trans_data_t) +
                    utarray_len(tags) * sizeof(neu_resp_tag_value_t));
 
+    data->timestamp = global_timestamp;
     strcpy(data->driver, group->driver->adapter.name);
     strcpy(data->group, group->name);
     data->n_tag = read_report_group(global_timestamp,

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/argparse.c

@@ -94,7 +94,7 @@ static inline int reset_password()
             "$5$PwFeXpBBIBZuZdZl$fP8fFPWCLoaWcnVXVSR.3Xi8TEqCvX92gjhowNNn6S4",
     };
 
-    int rv = neu_persister_create();
+    int rv = neu_persister_create(g_config_dir);
     if (0 != rv) {
         return rv;
     }

src/connection/connection.c

@@ -389,7 +389,7 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
                    conn->fd, len, ret, strerror(errno), errno);
     }
 
-    if (errno == EPIPE || ret == -1) {
+    if (errno == EPIPE || ret == -1 || ret == 0) {
         conn_disconnect(conn);
     }
 
@@ -589,7 +589,7 @@ static void conn_connect(neu_conn_t *conn)
                       sizeof(struct sockaddr_in));
         if (ret != 0 && errno != EINPROGRESS) {
             close(fd);
-            zlog_error(neuron, "connect %s:%d error: %s(%d)",
+            zlog_error(conn->param.log, "connect %s:%d error: %s(%d)",
                        conn->param.params.tcp_client.ip,
                        conn->param.params.tcp_client.port, strerror(errno),
                        errno);
@@ -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;

src/connection/mqtt_client.c

@@ -150,6 +150,7 @@ static void connect_cb(nng_pipe p, nng_pipe_ev ev, void *arg);
 
 static inline task_t *client_alloc_task(neu_mqtt_client_t *client);
 static inline void    client_free_task(neu_mqtt_client_t *client, task_t *task);
+static inline size_t  client_task_free_list_len(neu_mqtt_client_t *client);
 static inline void    client_add_subscription(neu_mqtt_client_t *client,
                                               subscription_t *   sub);
 static int            client_send_sub_msg(neu_mqtt_client_t *client,
@@ -545,6 +546,16 @@ static inline void client_free_task(neu_mqtt_client_t *client, task_t *task)
     DL_PREPEND(client->task_free_list, task);
 }
 
+static inline size_t client_task_free_list_len(neu_mqtt_client_t *client)
+{
+    size_t  count = 0;
+    task_t *elt   = NULL;
+    // count by iterating the list,
+    // this won't be a problem as a short free list is expected
+    DL_COUNT(client->task_free_list, elt, count);
+    return count;
+}
+
 static inline void client_add_subscription(neu_mqtt_client_t *client,
                                            subscription_t *   sub)
 {
@@ -1149,6 +1160,12 @@ int neu_mqtt_client_close(neu_mqtt_client_t *client)
     }
 
     nng_mtx_lock(client->mtx);
+    // wait for all tasks
+    while (client_task_free_list_len(client) != client->task_count) {
+        nng_mtx_unlock(client->mtx);
+        nng_msleep(100);
+        nng_mtx_lock(client->mtx);
+    }
     client->open = false;
     nng_mtx_unlock(client->mtx);
 

src/core/manager.c

@@ -58,7 +58,6 @@ inline static void forward_msg(neu_manager_t *manager, nng_msg *msg,
                                const char *ndoe);
 static void start_static_adapter(neu_manager_t *manager, const char *name);
 static int  update_timestamp(void *usr_data);
-static int  report_nodes_state(void *usr_data);
 static void start_single_adapter(neu_manager_t *manager, const char *name,
                                  const char *plugin_name, bool display);
 
@@ -75,13 +74,7 @@ neu_manager_t *neu_manager_create()
         .second      = 0,
         .millisecond = 10,
         .cb          = update_timestamp,
-        .type        = NEU_EVENT_TIMER_BLOCK,
-    };
-    neu_event_timer_param_t timer_param = {
-        .second      = 1,
-        .millisecond = 0,
-        .cb          = report_nodes_state,
-        .type        = NEU_EVENT_TIMER_BLOCK,
+        .type        = NEU_EVENT_TIMER_NOBLOCK,
     };
 
     neu_event_timer_param_t timer_level = {
@@ -138,9 +131,6 @@ neu_manager_t *neu_manager_create()
     timer_level.usr_data = (void *) manager;
     manager->timer_lev   = neu_event_add_timer(manager->events, timer_level);
 
-    timer_param.usr_data = (void *) manager;
-    manager->timer       = neu_event_add_timer(manager->events, timer_param);
-
     timestamp_timer_param.usr_data = (void *) manager;
     manager->timer_timestamp =
         neu_event_add_timer(manager->events, timestamp_timer_param);
@@ -155,7 +145,6 @@ void neu_manager_destroy(neu_manager_t *manager)
     nng_msg *           uninit_msg = NULL;
     UT_array *pipes = neu_node_manager_get_pipes_all(manager->node_manager);
 
-    neu_event_del_timer(manager->events, manager->timer);
     neu_event_del_timer(manager->events, manager->timer_lev);
     neu_event_del_timer(manager->events, manager->timer_timestamp);
     strcpy(header.sender, "manager");
@@ -733,46 +722,6 @@ inline static void reply(neu_manager_t *manager, neu_reqresp_head_t *header,
     }
 }
 
-static int report_nodes_state(void *usr_data)
-{
-    neu_manager_t *manager = (neu_manager_t *) usr_data;
-    UT_array *     apps    = neu_sub_msg_manager_get(manager->sub_msg_manager,
-                                             NEU_SUBSCRIBE_NODES_STATE);
-
-    neu_manager_count_tag(manager);
-    if (apps == NULL) {
-        return 0;
-    }
-
-    UT_array *states = neu_node_manager_get_state(manager->node_manager);
-
-    utarray_foreach(apps, char *, app)
-    {
-        nng_msg *                  msg         = NULL;
-        neu_reqresp_head_t         header      = { 0 };
-        neu_resp_get_nodes_state_t node_states = { 0 };
-        nng_pipe pipe = neu_node_manager_get_pipe(manager->node_manager, app);
-
-        header.type = NEU_REQRESP_NODES_STATE;
-        strcpy(header.sender, "manager");
-        strcpy(header.receiver, app);
-        node_states.states = utarray_clone(states);
-
-        msg = neu_msg_gen(&header, (void *) &node_states);
-        nng_msg_set_pipe(msg, pipe);
-
-        if (nng_sendmsg(manager->socket, msg, 0) != 0) {
-            nng_msg_free(msg);
-            nlog_warn("manager -> %s nodes state fail", app);
-        }
-    }
-
-    utarray_free(apps);
-    utarray_free(states);
-
-    return 0;
-}
-
 static int update_timestamp(void *usr_data)
 {
     (void) usr_data;

src/core/manager_internal.h

@@ -41,7 +41,6 @@ typedef struct neu_manager {
     neu_subscribe_mgr_t * subscribe_manager;
     neu_sub_msg_mgr_t *   sub_msg_manager;
 
-    neu_event_timer_t *timer;
     neu_event_timer_t *timer_timestamp;
 
     neu_event_timer_t *timer_lev;

src/event/event_linux.c

@@ -88,6 +88,11 @@ static void *event_loop(void *arg)
         if (ret == 0) {
             continue;
         }
+
+        if (ret == -1 && errno == EINTR) {
+            continue;
+        }
+
         if (ret == -1 || events->stop) {
             zlog_warn(neuron, "event loop exit, errno: %s(%d), stop: %d",
                       strerror(errno), errno, events->stop);

src/main.c

@@ -42,7 +42,7 @@ static void sig_handler(int sig)
 {
     nlog_warn("recv sig: %d", sig);
 
-    if (sig != SIGABRT) {
+    if (sig == SIGINT || sig == SIGTERM) {
         neu_manager_destroy(g_manager);
         neu_persister_destroy();
     }
@@ -58,6 +58,7 @@ static int neuron_run(const neu_cli_args_t *args)
     signal(SIGINT, sig_handler);
     signal(SIGTERM, sig_handler);
     signal(SIGABRT, sig_handler);
+    signal(SIGSEGV, sig_handler);
 
     // try to enable core dump
     rl.rlim_cur = rl.rlim_max = RLIM_INFINITY;
@@ -65,7 +66,7 @@ static int neuron_run(const neu_cli_args_t *args)
         nlog_warn("neuron process failed enable core dump, ignore");
     }
 
-    rv = neu_persister_create();
+    rv = neu_persister_create(args->config_dir);
     assert(rv == 0);
 
     zlog_notice(neuron, "neuron process, daemon: %d, version: %s (%s %s)",

src/persist/json/persist_json_plugin.c

@@ -27,7 +27,7 @@
 
 #include "json/json.h"
 
-#include "persist_json_plugin.h"
+#include "persist/json/persist_json_plugin.h"
 
 int neu_json_decode_plugin_req(char *buf, neu_json_plugin_req_t **result)
 {

src/persist/persist.c

@@ -333,7 +333,7 @@ static int apply_schema_file(sqlite3 *db, const char *dir, const char *file)
     char *err_msg = NULL;
     rv            = sqlite3_exec(db, sql, NULL, NULL, &err_msg);
     if (SQLITE_OK != rv) {
-        nlog_error("execute %s fail: %s", path, err_msg);
+        nlog_error("execute %s fail: (%d)%s", path, rv, err_msg);
         sqlite3_free(err_msg);
         rv = NEU_ERR_EINTERNAL;
         goto end;
@@ -430,7 +430,7 @@ static int apply_schemas(sqlite3 *db, const char *dir)
     return rv;
 }
 
-int neu_persister_create()
+int neu_persister_create(const char *schema_dir)
 {
     int rv = sqlite3_open(db_file, &global_db);
 
@@ -455,7 +455,7 @@ int neu_persister_create()
         return -1;
     }
 
-    rv = apply_schemas(global_db, g_config_dir);
+    rv = apply_schemas(global_db, schema_dir);
     if (rv != 0) {
         nlog_fatal("db apply schemas fail");
         sqlite3_close(global_db);
@@ -465,6 +465,11 @@ int neu_persister_create()
     return 0;
 }
 
+sqlite3 *neu_persister_get_db()
+{
+    return global_db;
+}
+
 void neu_persister_destroy()
 {
     sqlite3_close(global_db);
@@ -772,7 +777,8 @@ int neu_persister_load_tags(const char *driver_name, const char *group_name,
     sqlite3_stmt *stmt  = NULL;
     const char *  query = "SELECT name, address, attribute, precision, type, "
                         "decimal, description "
-                        "FROM tags WHERE driver_name=? AND group_name=?";
+                        "FROM tags WHERE driver_name=? AND group_name=? "
+                        "ORDER BY rowid ASC";
 
     utarray_new(*tags, neu_tag_get_icd());
 

tests/CMakeLists.txt

@@ -24,7 +24,7 @@ target_include_directories(jwt_test PRIVATE
     ${CMAKE_SOURCE_DIR}/src 
 	${CMAKE_SOURCE_DIR}/include
 	${CMAKE_SOURCE_DIR}/src/utils)
-target_link_libraries(jwt_test neuron-base gtest_main gtest jansson ssl crypto jwt)
+target_link_libraries(jwt_test neuron-base gtest_main gtest jansson ssl crypto jwt libzlog.so)
 
 add_executable(base64_test base64_test.cc ${CMAKE_SOURCE_DIR}/src/utils/base64.c)
 target_include_directories(base64_test PRIVATE 

version

@@ -1 +1 @@
-2.3.0
+2.3.13