Merge pull request #1928 from fengzeroz/v2.6

version: 2.6.7

.github/workflows/ci.yml

@@ -8,7 +8,7 @@ on:
       - published
 
 env:
-  VERSION: 2.6.5
+  VERSION: 2.6.7
   UI_VERSION: 2.6.3
   BRANCH: v2.6
 

CMakeLists.txt

@@ -130,7 +130,6 @@ set(NEURON_SOURCES
     src/adapter/driver/driver.c
     plugins/restful/handle.c
     plugins/restful/log_handle.c
-    plugins/restful/file_handle.c
     plugins/restful/metric_handle.c
     plugins/restful/normal_handle.c
     plugins/restful/rw_handle.c

plugins/ekuiper/json_rw.c

@@ -185,7 +185,8 @@ int json_decode_write_req(char *buf, size_t len, json_write_req_t **result)
 
     json_obj = neu_json_decode_newb(buf, len);
     if (NULL == json_obj) {
-        goto decode_fail;
+        free(req);
+        return -1;
     }
 
     neu_json_elem_t req_elems[] = {
@@ -208,27 +209,20 @@ int json_decode_write_req(char *buf, size_t len, json_write_req_t **result)
     };
     ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
                                   req_elems);
-    if (ret != 0) {
-        goto decode_fail;
-    }
-
     req->node_name  = req_elems[0].v.val_str;
     req->group_name = req_elems[1].v.val_str;
     req->tag_name   = req_elems[2].v.val_str;
     req->t          = req_elems[3].t;
     req->value      = req_elems[3].v;
 
-    *result = req;
-    goto decode_exit;
-
-decode_fail:
-    free(req);
-    ret = -1;
-
-decode_exit:
-    if (json_obj != NULL) {
-        neu_json_decode_free(json_obj);
+    if (0 == ret) {
+        *result = req;
+    } else {
+        json_decode_write_req_free(req);
+        ret = -1;
     }
+
+    neu_json_decode_free(json_obj);
     return ret;
 }
 
@@ -243,6 +237,8 @@ void json_decode_write_req_free(json_write_req_t *req)
     free(req->tag_name);
     if (req->t == NEU_JSON_STR) {
         free(req->value.val_str);
+    } else if (req->t == NEU_JSON_BYTES) {
+        free(req->value.val_bytes.bytes);
     }
 
     free(req);

plugins/ekuiper/plugin_ekuiper.c

@@ -114,10 +114,9 @@ static int ekuiper_plugin_uninit(neu_plugin_t *plugin)
     return rv;
 }
 
-static int ekuiper_plugin_start(neu_plugin_t *plugin)
+static inline int start(neu_plugin_t *plugin, const char *url)
 {
-    int   rv  = 0;
-    char *url = plugin->url ? plugin->url : EKUIPER_PLUGIN_URL; // default url
+    int rv = 0;
 
     rv = nng_pair0_open(&plugin->sock);
     if (rv != 0) {
@@ -143,16 +142,43 @@ static int ekuiper_plugin_start(neu_plugin_t *plugin)
         return rv;
     }
 
+    nng_mtx_lock(plugin->mtx);
+    while (plugin->receiving) {
+        nng_mtx_unlock(plugin->mtx);
+        nng_msleep(10);
+        nng_mtx_lock(plugin->mtx);
+    }
+    plugin->receiving = true;
     nng_recv_aio(plugin->sock, plugin->recv_aio);
+    nng_mtx_unlock(plugin->mtx);
+
+    return NEU_ERR_SUCCESS;
+}
+
+static int ekuiper_plugin_start(neu_plugin_t *plugin)
+{
+    int   rv  = 0;
+    char *url = plugin->url ? plugin->url : EKUIPER_PLUGIN_URL; // default url
+
+    rv = start(plugin, url);
+    if (rv != 0) {
+        return rv;
+    }
+
     plugin->started = true;
     plog_notice(plugin, "start successfully");
 
     return NEU_ERR_SUCCESS;
 }
 
-static int ekuiper_plugin_stop(neu_plugin_t *plugin)
+static inline void stop(neu_plugin_t *plugin)
 {
     nng_close(plugin->sock);
+}
+
+static int ekuiper_plugin_stop(neu_plugin_t *plugin)
+{
+    stop(plugin);
     plugin->started = false;
     plog_notice(plugin, "stop successfully");
     return NEU_ERR_SUCCESS;
@@ -201,41 +227,6 @@ error:
     return -1;
 }
 
-static inline int check_url_listenable(neu_plugin_t *plugin, const char *url,
-                                       const char *host, uint16_t port)
-{
-    if (NULL != plugin->host &&                  // already configured and
-        port == plugin->port &&                  // port is the same, then if
-        (0 == strcmp(plugin->host, host)         // 1. host is the same
-         || 0 == strcmp("0.0.0.0", host)         // 2. to bind to any address
-         || 0 == strcmp("0.0.0.0", plugin->host) // 3. bound to any address
-         )) {
-        // early return, no need check url is listenable
-        return 0;
-    }
-
-    nng_socket sock = NNG_SOCKET_INITIALIZER;
-    int        rv   = nng_pair0_open(&sock);
-    if (0 != rv) {
-        plog_error(plugin, "nng_pair0_open: %s", nng_strerror(rv));
-        return NEU_ERR_EINTERNAL;
-    }
-
-    if (0 != (rv = nng_listen(sock, url, NULL, 0))) {
-        plog_error(plugin, "nng_listen: %s", nng_strerror(rv));
-        if (NNG_EADDRINVAL == rv) {
-            rv = NEU_ERR_IP_ADDRESS_INVALID;
-        } else if (NNG_EADDRINUSE == rv) {
-            rv = NEU_ERR_IP_ADDRESS_IN_USE;
-        } else {
-            rv = NEU_ERR_EINTERNAL;
-        }
-    }
-
-    nng_close(sock);
-    return rv;
-}
-
 static int ekuiper_plugin_config(neu_plugin_t *plugin, const char *setting)
 {
     int      rv   = 0;
@@ -255,23 +246,31 @@ static int ekuiper_plugin_config(neu_plugin_t *plugin, const char *setting)
         goto error;
     }
 
-    if (0 != (rv = check_url_listenable(plugin, url, host, port))) {
+    if (plugin->started) {
+        stop(plugin);
+    }
+
+    // check we could start the plugin with the new setting
+    if (0 != (rv = start(plugin, url))) {
+        // recover with old setting
+        if (plugin->started && 0 != start(plugin, plugin->url)) {
+            plog_warn(plugin, "restart host:%s port:%" PRIu16 " fail",
+                      plugin->host, plugin->port);
+        }
         goto error;
     }
 
+    if (!plugin->started) {
+        stop(plugin);
+    }
+
     plog_notice(plugin, "config success");
 
     free(plugin->host);
+    free(plugin->url);
     plugin->host = host;
     plugin->port = port;
-    free(plugin->url);
-    plugin->url = url;
-
-    if (plugin->started) {
-        // restart service
-        ekuiper_plugin_stop(plugin);
-        ekuiper_plugin_start(plugin);
-    }
+    plugin->url  = url;
 
     return rv;
 
@@ -310,12 +309,17 @@ static int ekuiper_plugin_request(neu_plugin_t *      plugin,
         }
         break;
     }
+    case NEU_REQRESP_NODE_DELETED: {
+        break;
+    }
     case NEU_REQ_UPDATE_NODE: {
         break;
     }
     case NEU_REQ_UPDATE_GROUP: {
         break;
     }
+    case NEU_REQ_DEL_GROUP:
+        break;
     case NEU_REQ_SUBSCRIBE_GROUP:
     case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: {
         neu_req_subscribe_t *sub_info = data;

plugins/ekuiper/plugin_ekuiper.h

@@ -37,6 +37,7 @@ struct neu_plugin {
     nng_socket          sock;
     nng_mtx *           mtx;
     bool                started;
+    bool                receiving;
     nng_aio *           recv_aio;
     char *              host;
     uint16_t            port;

plugins/ekuiper/read_write.c

@@ -86,6 +86,9 @@ void recv_data_callback(void *arg)
     rv = nng_aio_result(plugin->recv_aio);
     if (0 != rv) {
         plog_error(plugin, "nng_recv error: %s", nng_strerror(rv));
+        nng_mtx_lock(plugin->mtx);
+        plugin->receiving = false;
+        nng_mtx_unlock(plugin->mtx);
         return;
     }
 

plugins/modbus/modbus_req.c

@@ -698,7 +698,7 @@ static int process_protocol_buf(neu_plugin_t *plugin, uint8_t slave_id,
                 plog_recv_protocol(plugin, recv_buf, ret);
             }
             neu_protocol_unpack_buf_init(&pbuf, recv_buf, ret);
-            int ret_s = modbus_stack_recv(plugin->stack, 0, &pbuf);
+            int ret_s = modbus_stack_recv(plugin->stack, slave_id, &pbuf);
             if (ret_s == MODBUS_DEVICE_ERR) {
                 ret = ret_s;
             } else {

plugins/modbus/modbus_stack.c

@@ -84,10 +84,8 @@ int modbus_stack_recv(modbus_stack_t *stack, uint8_t slave_id,
         return -1;
     }
 
-    if (stack->protocol == MODBUS_PROTOCOL_TCP) {
-        if (code.slave_id != slave_id) {
-            return -1;
-        }
+    if (code.slave_id != slave_id) {
+        return -1;
     }
 
     switch (code.function) {
@@ -102,18 +100,36 @@ int modbus_stack_recv(modbus_stack_t *stack, uint8_t slave_id,
             return -1;
         }
 
-        if (stack->protocol == MODBUS_PROTOCOL_TCP && data.n_byte == 0xff) {
-            bytes = neu_protocol_unpack_buf(buf,
-                                            header.len -
-                                                sizeof(struct modbus_code) -
-                                                sizeof(struct modbus_data));
-            stack->value_fn(stack->ctx, code.slave_id,
-                            header.len - sizeof(struct modbus_code) -
-                                sizeof(struct modbus_data),
-                            bytes, 0);
-        } else {
+        switch (stack->protocol) {
+        case MODBUS_PROTOCOL_TCP:
+            if (data.n_byte == 0xff) {
+                bytes = neu_protocol_unpack_buf(buf,
+                                                header.len -
+                                                    sizeof(struct modbus_code) -
+                                                    sizeof(struct modbus_data));
+                if (bytes == NULL) {
+                    return -1;
+                }
+                stack->value_fn(stack->ctx, code.slave_id,
+                                header.len - sizeof(struct modbus_code) -
+                                    sizeof(struct modbus_data),
+                                bytes, 0);
+            } else {
+                bytes = neu_protocol_unpack_buf(buf, data.n_byte);
+                if (bytes == NULL) {
+                    return -1;
+                }
+                stack->value_fn(stack->ctx, code.slave_id, data.n_byte, bytes,
+                                0);
+            }
+            break;
+        case MODBUS_PROTOCOL_RTU:
             bytes = neu_protocol_unpack_buf(buf, data.n_byte);
+            if (bytes == NULL) {
+                return -1;
+            }
             stack->value_fn(stack->ctx, code.slave_id, data.n_byte, bytes, 0);
+            break;
         }
 
         break;

plugins/mqtt/mqtt_handle.c

@@ -601,6 +601,14 @@ int handle_unsubscribe_group(neu_plugin_t *         plugin,
     return 0;
 }
 
+int handle_del_group(neu_plugin_t *plugin, neu_req_del_group_t *req)
+{
+    route_tbl_del(&plugin->route_tbl, req->driver, req->group);
+    plog_notice(plugin, "del route driver:%s group:%s", req->driver,
+                req->group);
+    return 0;
+}
+
 int handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req)
 {
     route_tbl_update_group(&plugin->route_tbl, req->driver, req->group,

plugins/mqtt/mqtt_handle.h

@@ -49,6 +49,7 @@ int handle_unsubscribe_group(neu_plugin_t *         plugin,
                              neu_req_unsubscribe_t *unsub_info);
 
 int handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req);
+int handle_del_group(neu_plugin_t *plugin, neu_req_del_group_t *cmd);
 
 int handle_update_driver(neu_plugin_t *plugin, neu_req_update_node_t *req);
 int handle_del_driver(neu_plugin_t *plugin, neu_reqresp_node_deleted_t *req);

plugins/mqtt/mqtt_plugin.c

@@ -368,6 +368,9 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
     case NEU_REQ_UPDATE_GROUP:
         error = handle_update_group(plugin, data);
         break;
+    case NEU_REQ_DEL_GROUP:
+        error = handle_del_group(plugin, data);
+        break;
     case NEU_REQ_UPDATE_NODE:
         error = handle_update_driver(plugin, data);
         break;

plugins/restful/file_handle.c

@@ -1,191 +0,0 @@
-/**
- * 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 <dirent.h>
-#include <errno.h>
-#include <stdio.h>
-#include <stdlib.h>
-#include <string.h>
-#include <strings.h>
-#include <sys/stat.h>
-#include <sys/types.h>
-#include <time.h>
-#include <unistd.h>
-
-#include <nng/nng.h>
-#include <nng/supplemental/http/http.h>
-
-#include "define.h"
-#include "handle.h"
-#include "utils/http.h"
-
-#include "plugin.h"
-#include "utils/log.h"
-#include "json/neu_json_error.h"
-#include "json/neu_json_fn.h"
-
-#include "file_handle.h"
-#include "log_handle.h"
-#include "parser/neu_json_file.h"
-
-void handle_file_info(nng_aio *aio)
-{
-    char dir_path[128] = { 0 };
-
-    int            n_file      = 0;
-    struct dirent *files       = { 0 };
-    char *         f_name[256] = { 0 };
-    char           c_time[32]  = { 0 };
-    char           m_time[32]  = { 0 };
-
-    neu_json_get_file_list_resp_t file_resp = { 0 };
-    char *                        result    = NULL;
-
-    NEU_VALIDATE_JWT(aio);
-
-    if (neu_http_get_param_str(aio, "dir_path", dir_path, sizeof(dir_path)) <=
-        0) {
-        NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
-            neu_http_response(aio, error_code.error, result_error);
-        });
-
-        return;
-    }
-
-    /*check whether the directory exists*/
-    if (0 != access(dir_path, F_OK)) {
-        NEU_JSON_RESPONSE_ERROR(NEU_ERR_FILE_NOT_EXIST, {
-            neu_http_response(aio, error_code.error, result_error);
-        });
-
-        return;
-    }
-
-    /*open directory*/
-    DIR *dir = opendir(dir_path);
-    if (dir == NULL) {
-        nlog_error("open a directory failed");
-        NEU_JSON_RESPONSE_ERROR(NEU_ERR_FILE_OPEN_FAILURE, {
-            neu_http_response(aio, error_code.error, result_error);
-        });
-
-        return;
-    }
-
-    /*read directory*/
-    while ((files = readdir(dir)) != NULL) {
-        if (strcmp(files->d_name, ".") == 0 ||
-            strcmp(files->d_name, "..") == 0) {
-            continue;
-        }
-
-        f_name[n_file] = files->d_name;
-        ++n_file;
-    }
-
-    /*handle response*/
-    file_resp.n_file = n_file;
-    file_resp.file = calloc(file_resp.n_file, sizeof(neu_json_get_file_resp_t));
-
-    for (int i = 0; i < file_resp.n_file; i++) {
-        char        file_path[512] = { 0 };
-        struct stat statbuf;
-
-        sprintf(file_path, "%s/%s", dir_path, f_name[i]);
-        stat(file_path, &statbuf);
-
-        strncpy(c_time, ctime(&statbuf.st_ctim.tv_sec),
-                strlen(ctime(&statbuf.st_ctim.tv_sec)) - 1);
-        strncpy(m_time, ctime(&statbuf.st_mtim.tv_sec),
-                strlen(ctime(&statbuf.st_mtim.tv_sec)) - 1);
-
-        file_resp.file[i].name  = f_name[i];
-        file_resp.file[i].size  = statbuf.st_size;
-        file_resp.file[i].ctime = c_time;
-        file_resp.file[i].mtime = m_time;
-    }
-
-    neu_json_encode_by_fn(&file_resp, neu_json_encode_get_file_list_resp,
-                          &result);
-
-    closedir(dir);
-    neu_http_ok(aio, result);
-    free(result);
-    free(file_resp.file);
-
-    return;
-}
-
-void handle_download_file(nng_aio *aio)
-{
-    char file_name[NEU_FILE_PATH_LEN] = { 0 };
-    char file_path[256]               = { 0 };
-    char http_resp[256]               = { 0 };
-
-    int    rv        = 0;
-    void * file_data = NULL;
-    size_t file_len  = 0;
-    char * p         = NULL;
-
-    NEU_VALIDATE_JWT(aio);
-
-    if (neu_http_get_param_str(aio, "file_path", file_path,
-                               sizeof(file_path)) <= 0) {
-        NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
-            neu_http_response(aio, error_code.error, result_error);
-        });
-
-        return;
-    }
-
-    /*check whether the file exists*/
-    rv = access(file_path, F_OK);
-    if (0 != rv) {
-        NEU_JSON_RESPONSE_ERROR(NEU_ERR_FILE_NOT_EXIST, {
-            neu_http_response(aio, error_code.error, result_error);
-        });
-
-        return;
-    }
-
-    /*read file content*/
-    rv = read_file(file_path, &file_data, &file_len);
-    if (0 != rv) {
-        NEU_JSON_RESPONSE_ERROR(
-            rv, { neu_http_response(aio, error_code.error, result_error); });
-
-        return;
-    }
-
-    /*extract file name*/
-    p = strchr(file_path, '/');
-    while (strchr(p, '/') != NULL) {
-        p = strchr(p, '/') + 1;
-    }
-
-    if (p != NULL) {
-        strcpy(file_name, p);
-    }
-
-    /*handle http response*/
-    sprintf(http_resp, "%s%s", "attachment; filename=", file_name);
-    rv = neu_http_response_file(aio, file_data, file_len, http_resp);
-    free(file_data);
-    nlog_notice("download file %s, rv: %d", file_name, rv);
-}

plugins/restful/file_handle.h

@@ -1,28 +0,0 @@
-/**
- * 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_FILE_HANDLE_H_
-#define _NEU_FILE_HANDLE_H_
-
-#include <nng/nng.h>
-
-void handle_download_file(nng_aio *aio);
-void handle_file_info(nng_aio *aio);
-
-#endif

plugins/restful/handle.c

@@ -26,7 +26,6 @@
 
 #include "adapter_handle.h"
 #include "datatag_handle.h"
-#include "file_handle.h"
 #include "global_config_handle.h"
 #include "group_config_handle.h"
 #include "log_handle.h"
@@ -111,12 +110,6 @@ static struct neu_http_handler cors_handler[] = {
     {
         .url = "/api/v2/log/level",
     },
-    {
-        .url = "/api/v2/file",
-    },
-    {
-        .url = "/api/v2/file/info",
-    },
     {
         .url = "/api/v2/global/config",
     },
@@ -377,18 +370,6 @@ static struct neu_http_handler rest_handlers[] = {
         .url           = "/api/v2/version",
         .value.handler = handle_get_version,
     },
-    {
-        .method        = NEU_HTTP_METHOD_GET,
-        .type          = NEU_HTTP_HANDLER_FUNCTION,
-        .url           = "/api/v2/file",
-        .value.handler = handle_download_file,
-    },
-    {
-        .method        = NEU_HTTP_METHOD_GET,
-        .type          = NEU_HTTP_HANDLER_FUNCTION,
-        .url           = "/api/v2/file/info",
-        .value.handler = handle_file_info,
-    },
     {
         .method        = NEU_HTTP_METHOD_GET,
         .type          = NEU_HTTP_HANDLER_FUNCTION,

src/adapter/adapter.c

@@ -941,7 +941,10 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
             error.error = neu_adapter_driver_del_group(
                 (neu_adapter_driver_t *) adapter, cmd->group);
         } else {
-            error.error = NEU_ERR_GROUP_NOT_ALLOW;
+            adapter->module->intf_funs->request(
+                adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
+            neu_msg_free(msg);
+            break;
         }
 
         if (error.error == NEU_ERR_SUCCESS) {

src/adapter/driver/driver.c

@@ -91,6 +91,7 @@ static void read_group(int64_t timestamp, int64_t timeout,
                        neu_driver_cache_t *cache, const char *group,
                        UT_array *tags, UT_array *tag_values);
 static void read_report_group(int64_t timestamp, int64_t timeout,
+                              neu_tag_cache_type_e cache_type,
                               neu_driver_cache_t *cache, const char *group,
                               UT_array *tags, UT_array *tag_values);
 static void update(neu_adapter_t *adapter, const char *group, const char *tag,
@@ -225,8 +226,9 @@ static void update_im(neu_adapter_t *adapter, const char *group,
     data->group  = strdup(group);
     utarray_new(data->tags, neu_resp_tag_value_meta_icd());
 
-    read_report_group(global_timestamp, 0, driver->cache, group, tags,
-                      data->tags);
+    read_report_group(global_timestamp, 0,
+                      neu_adapter_get_tag_cache_type(&driver->adapter),
+                      driver->cache, group, tags, data->tags);
 
     if (utarray_len(data->tags) > 0) {
         group_t *find = NULL;
@@ -1474,6 +1476,7 @@ static int report_callback(void *usr_data)
     read_report_group(global_timestamp,
                       neu_group_get_interval(group->group) *
                           NEU_DRIVER_TAG_CACHE_EXPIRE_TIME,
+                      neu_adapter_get_tag_cache_type(&group->driver->adapter),
                       group->driver->cache, group->name, tags, data->tags);
 
     nlog_info("report group: %s, all tags: %d, report tags: %d", group->name,
@@ -1495,6 +1498,12 @@ static int report_callback(void *usr_data)
                 }
             }
         } else {
+            utarray_foreach(data->tags, neu_resp_tag_value_meta_t *, tag_value)
+            {
+                if (tag_value->value.type == NEU_TYPE_PTR) {
+                    free(tag_value->value.value.ptr.ptr);
+                }
+            }
             utarray_free(data->tags);
             free(data->group);
             free(data->driver);
@@ -1643,6 +1652,7 @@ static int read_callback(void *usr_data)
 }
 
 static void read_report_group(int64_t timestamp, int64_t timeout,
+                              neu_tag_cache_type_e cache_type,
                               neu_driver_cache_t *cache, const char *group,
                               UT_array *tags, UT_array *tag_values)
 {
@@ -1735,7 +1745,8 @@ static void read_report_group(int64_t timestamp, int64_t timeout,
             break;
         }
 
-        if (!neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) &&
+        if (cache_type != NEU_TAG_CACHE_TYPE_NEVER &&
+            !neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) &&
             (timestamp - value.timestamp) > timeout && timeout > 0) {
             tag_value.value.type      = NEU_TYPE_ERROR;
             tag_value.value.value.i32 = NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED;

src/connection/connection.c

@@ -1284,10 +1284,18 @@ int neu_conn_stream_consume(neu_conn_t *conn, void *context,
                 neu_conn_disconnect(conn);
                 break;
             } else {
+                pthread_mutex_lock(&conn->mtx);
+                if (conn->offset - used < 0) {
+                    pthread_mutex_unlock(&conn->mtx);
+                    zlog_warn(conn->param.log, "reset offset: %d, used: %d",
+                              conn->offset, used);
+                    return -1;
+                }
                 conn->offset -= used;
                 memmove(conn->buf, conn->buf + used, conn->offset);
                 neu_protocol_unpack_buf_init(&protocol_buf, conn->buf,
                                              conn->offset);
+                pthread_mutex_unlock(&conn->mtx);
             }
         }
     }
@@ -1313,10 +1321,18 @@ int neu_conn_stream_tcp_server_consume(neu_conn_t *conn, int fd, void *context,
                 neu_conn_tcp_server_close_client(conn, fd);
                 break;
             } else {
+                pthread_mutex_lock(&conn->mtx);
+                if (conn->offset - used < 0) {
+                    pthread_mutex_unlock(&conn->mtx);
+                    zlog_warn(conn->param.log, "reset offset: %d, used: %d",
+                              conn->offset, used);
+                    return -1;
+                }
                 conn->offset -= used;
                 memmove(conn->buf, conn->buf + used, conn->offset);
                 neu_protocol_unpack_buf_init(&protocol_buf, conn->buf,
                                              conn->offset);
+                pthread_mutex_unlock(&conn->mtx);
             }
         }
     }

src/core/manager.c

@@ -982,9 +982,24 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
             neu_msg_exchange(header);
             reply(manager, header, &e);
         } else {
+            UT_array *apps = neu_subscribe_manager_find(
+                manager->subscribe_manager, cmd->driver, cmd->group);
+
             forward_msg(manager, header, header->receiver);
             neu_subscribe_manager_remove(manager->subscribe_manager,
                                          cmd->driver, cmd->group);
+
+            if (NULL == apps) {
+                break;
+            }
+
+            // notify app node about group deletion
+            utarray_foreach(apps, neu_app_subscribe_t *, app)
+            {
+                forward_msg_copy(manager, header, app->app_name);
+            }
+
+            utarray_free(apps);
         }
         break;
     }

src/core/manager_internal.c

@@ -703,7 +703,6 @@ int neu_manager_subscribe(neu_manager_t *manager, const char *app,
     // this is not an elegant solution due to the current architecture
     if (params && 0 == strcmp(adapter->module->module_name, "MQTT")) {
         neu_json_elem_t elem = { .name = "topic", .t = NEU_JSON_STR };
-        neu_json_decode((char *) params, 1, &elem);
         neu_parse_param(params, NULL, 1, &elem);
         if (elem.v.val_str && 0 == strlen(elem.v.val_str)) {
             free(elem.v.val_str);

src/parser/neu_json_file.c

@@ -1,81 +0,0 @@
-/**
- * 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.
- **/
-
-/*
- * DO NOT EDIT THIS FILE MANUALLY!
- * It was automatically generated by `json-autotype`.
- */
-
-#include <stdlib.h>
-#include <string.h>
-
-#include "json/json.h"
-
-#include "neu_json_file.h"
-
-int neu_json_encode_get_file_list_resp(void *json_object, void *param)
-{
-    int                            ret = 0;
-    neu_json_get_file_list_resp_t *resp =
-        (neu_json_get_file_list_resp_t *) param;
-
-    void *                    file_list_array = neu_json_array();
-    neu_json_get_file_resp_t *p_file          = resp->file;
-
-    for (int i = 0; i < resp->n_file; i++) {
-        neu_json_elem_t file_elems[] = {
-            {
-                .name      = "name",
-                .t         = NEU_JSON_STR,
-                .v.val_str = p_file->name,
-            },
-            {
-                .name      = "size",
-                .t         = NEU_JSON_INT,
-                .v.val_int = p_file->size,
-            },
-            {
-                .name      = "ctime",
-                .t         = NEU_JSON_STR,
-                .v.val_str = p_file->ctime,
-            },
-            {
-                .name      = "mtime",
-                .t         = NEU_JSON_STR,
-                .v.val_str = p_file->mtime,
-            },
-        };
-
-        file_list_array = neu_json_encode_array(file_list_array, file_elems,
-                                                NEU_JSON_ELEM_SIZE(file_elems));
-
-        p_file++;
-    }
-
-    neu_json_elem_t resp_elems[] = { {
-        .name         = "files",
-        .t            = NEU_JSON_OBJECT,
-        .v.val_object = file_list_array,
-    } };
-
-    ret = neu_json_encode_field(json_object, resp_elems,
-                                NEU_JSON_ELEM_SIZE(resp_elems));
-
-    return ret;
-}

src/parser/neu_json_file.h

@@ -1,52 +0,0 @@
-/**
- * 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.
- **/
-
-/*
- * DO NOT EDIT THIS FILE MANUALLY!
- * It was automatically generated by `json-autotype`.
- */
-
-#ifndef _NEU_JOSN_API_NEU_JSON_FILE_H_
-#define _NEU_JSON_API_NEU_JSON_FILE_H_
-
-#include "json/json.h"
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-typedef struct {
-    char *name;
-    int   size;
-    char *ctime;
-    char *mtime;
-} neu_json_get_file_resp_t;
-
-typedef struct {
-    int                       n_file;
-    neu_json_get_file_resp_t *file;
-} neu_json_get_file_list_resp_t;
-
-int neu_json_encode_get_file_list_resp(void *json_object, void *param);
-
-#ifdef __cplusplus
-}
-#endif
-
-#endif

version

@@ -1 +1 @@
-2.6.5
+2.6.7