Merge pull request #2017 from fengzeroz/v2.4

V2.4

.github/workflows/ci.yml

@@ -8,8 +8,8 @@ on:
       - published
 
 env:
-  VERSION: 2.4.7
-  UI_VERSION: 2.4.4
+  VERSION: 2.4.13
+  UI_VERSION: 2.4.8
   BRANCH: v2.4
 
 jobs:

include/neuron/adapter.h

@@ -207,8 +207,8 @@ typedef struct neu_resp_error {
 } neu_resp_error_t;
 
 typedef struct {
-    char node[NEU_NODE_NAME_LEN];
-    bool auto_start;
+    char                     node[NEU_NODE_NAME_LEN];
+    neu_node_running_state_e state;
 } neu_req_node_init_t, neu_req_node_uninit_t, neu_resp_node_uninit_t;
 
 typedef struct neu_req_add_plugin {

include/neuron/connection/neu_connection.h

@@ -30,6 +30,7 @@ typedef enum neu_conn_type {
     NEU_CONN_TCP_CLIENT,
     NEU_CONN_UDP,
     NEU_CONN_TTY_CLIENT,
+    NEU_CONN_UDP_TO,
 } neu_conn_type_e;
 
 typedef enum neu_conn_tty_baud {
@@ -103,6 +104,12 @@ typedef struct neu_conn_param {
             uint16_t timeout; // millisecond
         } udp;
 
+        struct {
+            char *   src_ip;
+            uint16_t src_port;
+            uint16_t timeout; // millisecond
+        } udpto;
+
         struct {
             char *                device;
             neu_conn_tty_data_e   data;
@@ -231,6 +238,8 @@ int neu_conn_tcp_server_close_client(neu_conn_t *conn, int fd);
  * less than or equal to 0 fails.
  */
 ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len);
+ssize_t neu_conn_udp_sendto(neu_conn_t *conn, uint8_t *buf, ssize_t len,
+                            void *dst);
 
 /**
  * @brief Receive data via connection read.
@@ -242,6 +251,8 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len);
  * received, less than or equal to 0 fails.
  */
 ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len);
+ssize_t neu_conn_udp_recvfrom(neu_conn_t *conn, uint8_t *buf, ssize_t len,
+                              void *src);
 
 /**
  * @brief Specify the client to send data.

include/neuron/connection/neu_connection_eth.h

@@ -32,23 +32,19 @@ neu_conn_eth_t *neu_conn_eth_init(const char *interface, void *ctx);
 int             neu_conn_eth_uninit(neu_conn_eth_t *conn);
 int             neu_conn_eth_check_interface(const char *interface);
 void            neu_conn_eth_get_mac(neu_conn_eth_t *conn, uint8_t *mac);
+int             neu_conn_eth_size();
 
 typedef void (*neu_conn_eth_msg_callback)(neu_conn_eth_t *conn, void *ctx,
                                           uint16_t protocol, uint16_t n_byte,
                                           uint8_t *bytes, uint8_t src_mac[6]);
 
-int neu_conn_eth_register_lldp(neu_conn_eth_t *conn, const char *device,
-                               neu_conn_eth_msg_callback callback);
-int neu_conn_eth_unregister_lldp(neu_conn_eth_t *conn, const char *device);
-
 typedef struct neu_conn_eth_sub neu_conn_eth_sub_t;
 
-neu_conn_eth_sub_t *neu_conn_eth_register(neu_conn_eth_t *conn,
-                                          uint16_t protocol, uint8_t mac[6],
+neu_conn_eth_sub_t *neu_conn_eth_register(neu_conn_eth_t *conn, uint8_t mac[6],
                                           neu_conn_eth_msg_callback callback);
 int neu_conn_eth_unregister(neu_conn_eth_t *conn, neu_conn_eth_sub_t *sub);
 
-int neu_conn_eth_send(neu_conn_eth_t *conn, uint8_t dst_mac[6],
-                      uint16_t protocol, uint16_t n_byte, uint8_t *bytes);
+int neu_conn_eth_send(neu_conn_eth_t *conn, uint16_t protocol, uint16_t n_byte,
+                      uint8_t *bytes);
 
 #endif

include/neuron/connection/neu_smart_link.h

@@ -0,0 +1,25 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2023 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 NEURON_CONNECTION_SMART_LINK_H
+#define NEURON_CONNECTION_SMART_LINK_H
+
+int neu_conn_smart_link_auto_set(const char *dev_path);
+
+#endif

include/neuron/errcodes.h

@@ -133,10 +133,11 @@ typedef enum {
     NEU_ERR_MQTT_SUSPENDED                      = 4015,
     NEU_ERR_MQTT_PUBLISH_OVER_LENGTH            = 4016,
 
-    NEU_ERR_FILE_TOO_LONG      = 4100,
+    NEU_ERR_STRING_TOO_LONG    = 4100,
     NEU_ERR_FILE_OPEN_FAILURE  = 4101,
     NEU_ERR_FILE_READ_FAILURE  = 4102,
     NEU_ERR_FILE_WRITE_FAILURE = 4103,
+    NEU_ERR_FILE_READ_NULL     = 4104,
 
     NEU_ERR_PLUGIN_ERROR_START = 10000,
     NEU_ERR_PLUGIN_ERROR_END   = 20000,

include/neuron/persist/persist.h

@@ -95,6 +95,8 @@ int neu_persister_create(const char *schema_dir);
 void neu_persister_destroy();
 
 sqlite3 *neu_persister_get_db();
+int      neu_persister_node_count();
+int      neu_persister_tag_count();
 
 /**
  * Persist nodes.

include/neuron/type.h

@@ -143,25 +143,19 @@ static inline char *neu_value_str(neu_type_e type, neu_value_u value)
         snprintf(str, sizeof(str), "type: %s, value: %u", neu_type_string(type),
                  value.u32);
         break;
-    case NEU_TYPE_INT64:
-#if __WORDSIZE == 64
-        snprintf(str, sizeof(str), "type: %s, value: %li",
-                 neu_type_string(type), value.i64);
-#else
-        snprintf(str, sizeof(str), "type: %s, value: %lli",
-                 neu_type_string(type), value.i64);
-#endif
+    case NEU_TYPE_INT64: {
+        long long int i = value.i64;
+        snprintf(str, sizeof(str), "type: %s, value: %lld",
+                 neu_type_string(type), i);
         break;
+    }
     case NEU_TYPE_LWORD:
-    case NEU_TYPE_UINT64:
-#if __WORDSIZE == 64
-        snprintf(str, sizeof(str), "type: %s, value: %lu",
-                 neu_type_string(type), value.u64);
-#else
+    case NEU_TYPE_UINT64: {
+        long long unsigned int i = value.u64;
         snprintf(str, sizeof(str), "type: %s, value: %llu",
-                 neu_type_string(type), value.u64);
-#endif
+                 neu_type_string(type), i);
         break;
+    }
     case NEU_TYPE_FLOAT:
         snprintf(str, sizeof(str), "type: %s, value: %f", neu_type_string(type),
                  value.f32);

plugins/file/file.json

@@ -15,7 +15,8 @@
         "type": "int",
         "default": 512,
         "valid": {
-            "min": 1
+            "min": 1,
+            "max": 512
         }
     }
 }

plugins/file/file_req.c

@@ -62,7 +62,7 @@ int file_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
         }
 
         if ((f = fopen(tag->address, "rb+")) == NULL) {
-            nlog_warn("open file:%s", tag->address);
+            nlog_warn("fail open to read and write file: %s", tag->address);
             dvalue.type      = NEU_TYPE_ERROR;
             dvalue.value.i32 = NEU_ERR_FILE_OPEN_FAILURE;
             goto dvalue_result;
@@ -71,8 +71,11 @@ int file_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
         fseek(f, 0, SEEK_END);
         int size = ftell(f);
         if (size >= length) {
+            nlog_warn(
+                "the file length exceeds the set value, file: %s, length: %d",
+                tag->address, size);
             dvalue.type      = NEU_TYPE_ERROR;
-            dvalue.value.i32 = NEU_ERR_FILE_TOO_LONG;
+            dvalue.value.i32 = NEU_ERR_STRING_TOO_LONG;
 
             fclose(f);
             goto dvalue_result;
@@ -81,9 +84,23 @@ int file_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
         fseek(f, 0, SEEK_SET);
         int ret = fread(buf, sizeof(char), length, f);
         if (ret == 0) {
-            nlog_warn("read file failed:%s", tag->address);
+            if (feof(f)) {
+                nlog_warn("file is empty:%s", tag->address);
+                dvalue.type      = NEU_TYPE_ERROR;
+                dvalue.value.i32 = NEU_ERR_FILE_READ_NULL;
+            } else if (ferror(f)) {
+                nlog_warn("read file failed: %s", tag->address);
+                clearerr(f);
+                dvalue.type      = NEU_TYPE_ERROR;
+                dvalue.value.i32 = NEU_ERR_FILE_READ_FAILURE;
+            }
+            fclose(f);
+            goto dvalue_result;
+        } else if (ret >= NEU_VALUE_SIZE) {
+            nlog_warn("the file length exceeds the maximum, file: %s, ret: %d",
+                      tag->address, ret);
             dvalue.type      = NEU_TYPE_ERROR;
-            dvalue.value.i32 = NEU_ERR_FILE_READ_FAILURE;
+            dvalue.value.i32 = NEU_ERR_STRING_TOO_LONG;
 
             fclose(f);
             goto dvalue_result;
@@ -121,19 +138,32 @@ static void plugin_group_free(neu_plugin_group_t *pgp)
 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;
+    FILE *fp     = NULL;
+    int   ret    = 0;
+    int   length = strlen(value.str);
+
+    if (length >= NEU_VALUE_SIZE) {
+        nlog_warn("write file exceeds the maximum value, file: %s, length: %d",
+                  tag->address, length);
+        ret = NEU_ERR_STRING_TOO_LONG;
 
-    if (0 == fwrite(value.str, sizeof(char), strlen(value.str), fp)) {
-        nlog_warn("write file failed:%s", tag->address);
-        ret = NEU_ERR_FILE_WRITE_FAILURE;
         goto dvalue_result;
     }
 
+    fp = fopen(tag->address, "w");
+
+    if (0 == fwrite(value.str, sizeof(char), strlen(value.str), fp)) {
+        nlog_warn("write file failed: %s", tag->address);
+        ret = NEU_ERR_FILE_WRITE_FAILURE;
+
+        fclose(fp);
+        goto dvalue_result;
+    }
+
+    fclose(fp);
     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);
 

plugins/restful/log_handle.c

@@ -63,7 +63,8 @@ void handle_logs_files(nng_aio *aio)
     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");
+    rv = system("tar -zcvf /tmp/neuron_debug.tar.gz "
+                "--exclude='persistence/neuron.lic' ../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, {

plugins/restful/normal_handle.c

@@ -54,12 +54,20 @@ void handle_login(nng_aio *aio)
         aio, neu_json_login_req_t, neu_json_decode_login_req, {
             neu_json_login_resp_t login_resp = { 0 };
             neu_user_t *          user       = neu_load_user(req->name);
+            int                   pass_len   = strlen(req->pass);
 
             if (NULL == user) {
                 nlog_error("could not find user `%s`", req->name);
                 NEU_JSON_RESPONSE_ERROR(NEU_ERR_INVALID_USER_OR_PASSWORD, {
                     neu_http_response(aio, error_code.error, result_error);
                 });
+            } else if (pass_len < NEU_USER_PASSWORD_MIN_LEN ||
+                       pass_len > NEU_USER_PASSWORD_MAX_LEN) {
+                nlog_error("user `%s` password too short or too long",
+                           req->name);
+                NEU_JSON_RESPONSE_ERROR(NEU_ERR_INVALID_PASSWORD_LEN, {
+                    neu_http_response(aio, error_code.error, result_error);
+                });
             } else if (neu_user_check_password(user, req->pass)) {
 
                 char *token  = NULL;

plugins/restful/rw_handle.c

@@ -58,7 +58,6 @@ void handle_write(nng_aio *aio)
 
     NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
         aio, neu_json_write_req_t, neu_json_decode_write_req, {
-            int                 ret    = 0;
             neu_reqresp_head_t  header = { 0 };
             neu_req_write_tag_t cmd    = { 0 };
 
@@ -75,8 +74,13 @@ void handle_write(nng_aio *aio)
                 cmd.value.value.u64 = req->value.val_int;
                 break;
             case NEU_JSON_STR:
-                cmd.value.type = NEU_TYPE_STRING;
-                strcpy(cmd.value.value.str, req->value.val_str);
+                if (strlen(req->value.val_str) >= NEU_VALUE_SIZE) {
+                    cmd.value.type      = NEU_TYPE_ERROR;
+                    cmd.value.value.i32 = NEU_ERR_STRING_TOO_LONG;
+                } else {
+                    cmd.value.type = NEU_TYPE_STRING;
+                    strcpy(cmd.value.value.str, req->value.val_str);
+                }
                 break;
             case NEU_JSON_DOUBLE:
                 cmd.value.type      = NEU_TYPE_DOUBLE;
@@ -91,11 +95,18 @@ void handle_write(nng_aio *aio)
                 break;
             }
 
-            ret = neu_plugin_op(plugin, header, &cmd);
-            if (ret != 0) {
-                NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
-                    neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
+            if (cmd.value.type == NEU_TYPE_ERROR) {
+                NEU_JSON_RESPONSE_ERROR(NEU_ERR_STRING_TOO_LONG, {
+                    neu_http_response(aio, NEU_ERR_STRING_TOO_LONG,
+                                      result_error);
                 });
+            } else {
+                int ret = neu_plugin_op(plugin, header, &cmd);
+                if (ret != 0) {
+                    NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
+                        neu_http_response(aio, NEU_ERR_IS_BUSY, result_error);
+                    });
+                }
             }
         })
 }

src/adapter/adapter.c

@@ -187,7 +187,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
     return adapter;
 }
 
-void neu_adapter_init(neu_adapter_t *adapter, bool auto_start)
+void neu_adapter_init(neu_adapter_t *adapter, neu_node_running_state_e state)
 {
     neu_reqresp_head_t  header   = { .type = NEU_REQ_NODE_INIT };
     neu_req_node_init_t init     = { 0 };
@@ -197,8 +197,8 @@ void neu_adapter_init(neu_adapter_t *adapter, bool auto_start)
     strcpy(header.receiver, "manager");
 
     strcpy(init.node, adapter->name);
-    init.auto_start = auto_start;
-    init_msg        = neu_msg_gen(&header, &init);
+    init.state = state;
+    init_msg   = neu_msg_gen(&header, &init);
 
     nng_sendmsg(adapter->sock, init_msg, 0);
 }

src/adapter/adapter_internal.h

@@ -64,7 +64,7 @@ typedef struct adapter_msg_handler {
 } adapter_msg_handler_t;
 
 neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info);
-void           neu_adapter_init(neu_adapter_t *adapter, bool auto_start);
+void neu_adapter_init(neu_adapter_t *adapter, neu_node_running_state_e state);
 
 int neu_adapter_start(neu_adapter_t *adapter);
 int neu_adapter_start_single(neu_adapter_t *adapter);

src/adapter/driver/driver.c

@@ -55,6 +55,7 @@ typedef struct group {
 
     neu_event_timer_t *report;
     neu_event_timer_t *read;
+    neu_event_timer_t *write;
 
     neu_plugin_group_t    grp;
     neu_adapter_driver_t *driver;
@@ -73,6 +74,7 @@ struct neu_adapter_driver {
 
 static int  report_callback(void *usr_data);
 static int  read_callback(void *usr_data);
+static int  write_callback(void *usr_data);
 static void read_group(int64_t timestamp, int64_t timeout,
                        neu_driver_cache_t *cache, const char *group,
                        UT_array *tags, neu_resp_tag_value_t *datas);
@@ -185,6 +187,7 @@ int neu_adapter_driver_uninit(neu_adapter_driver_t *driver)
 
         neu_adapter_del_timer((neu_adapter_t *) driver, el->report);
         neu_event_del_timer(driver->driver_events, el->read);
+        neu_event_del_timer(driver->driver_events, el->write);
         if (el->grp.group_free != NULL) {
             el->grp.group_free(&el->grp);
         }
@@ -481,6 +484,11 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name,
         param.cb     = read_callback;
         find->read   = neu_event_add_timer(driver->driver_events, param);
 
+        param.second      = 0;
+        param.millisecond = 100;
+        param.cb          = write_callback;
+        find->write       = neu_event_add_timer(driver->driver_events, param);
+
         REGISTER_GROUP_METRIC(&driver->adapter, find->name,
                               NEU_METRIC_GROUP_TAGS_TOTAL,
                               neu_group_tag_size(find->group));
@@ -539,6 +547,7 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
 
         neu_adapter_del_timer((neu_adapter_t *) driver, find->report);
         neu_event_del_timer(driver->driver_events, find->read);
+        neu_event_del_timer(driver->driver_events, find->write);
         if (find->grp.group_free != NULL) {
             find->grp.group_free(&find->grp);
         }
@@ -917,7 +926,7 @@ static void group_change(void *arg, int64_t timestamp, UT_array *static_tags,
                 timestamp);
 }
 
-static int read_callback(void *usr_data)
+static int write_callback(void *usr_data)
 {
     group_t *                group = (group_t *) usr_data;
     neu_node_running_state_e state = group->driver->adapter.state;
@@ -936,6 +945,17 @@ static int read_callback(void *usr_data)
     utarray_clear(group->wt_tags);
     pthread_mutex_unlock(&group->wt_mtx);
 
+    return 0;
+}
+
+static int read_callback(void *usr_data)
+{
+    group_t *                group = (group_t *) usr_data;
+    neu_node_running_state_e state = group->driver->adapter.state;
+    if (state != NEU_NODE_RUNNING_STATE_RUNNING) {
+        return 0;
+    }
+
     if (neu_group_is_change(group->group, group->timestamp)) {
         neu_group_change_test(group->group, group->timestamp, (void *) group,
                               group_change);

src/base/group.c

@@ -286,7 +286,8 @@ static void split_static_array(tag_elem_t *tags, UT_array **static_tags,
     {
         if (neu_tag_attribute_test(el->tag, NEU_ATTRIBUTE_STATIC)) {
             utarray_push_back(*static_tags, el->tag);
-        } else {
+        } else if (neu_tag_attribute_test(el->tag, NEU_ATTRIBUTE_SUBSCRIBE) ||
+                   neu_tag_attribute_test(el->tag, NEU_ATTRIBUTE_READ)) {
             utarray_push_back(*other_tags, el->tag);
         }
     }

src/connection/connection.c

@@ -47,6 +47,7 @@ struct neu_conn {
     void *           data;
     bool             is_connected;
     bool             stop;
+    bool             connection_ok;
 
     neu_conn_callback connected;
     neu_conn_callback disconnected;
@@ -237,6 +238,8 @@ int neu_conn_tcp_server_close_client(neu_conn_t *conn, int fd)
 
     conn->disconnected(conn->data, fd);
     conn_tcp_server_del_client(conn, fd);
+    conn->offset = 0;
+    memset(conn->buf, 0, conn->buf_size);
 
     pthread_mutex_unlock(&conn->mtx);
     return 0;
@@ -316,6 +319,7 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len)
 
     if (conn->is_connected) {
         switch (conn->param.type) {
+        case NEU_CONN_UDP_TO:
         case NEU_CONN_TCP_SERVER:
             assert(false);
             break;
@@ -338,8 +342,14 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len)
                        conn->fd, len, ret, strerror(errno), errno);
         }
 
-        if (ret == -1 && errno != EAGAIN) {
-            conn_disconnect(conn);
+        if (ret == -1) {
+            if (errno != EAGAIN) {
+                conn_disconnect(conn);
+            } else {
+                if (conn->connection_ok == true) {
+                    conn_disconnect(conn);
+                }
+            }
         }
 
         if (ret > 0 && conn->callback_trigger == false) {
@@ -350,6 +360,7 @@ ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len)
 
     if (ret > 0) {
         conn->state.send_bytes += ret;
+        conn->connection_ok = true;
     }
 
     pthread_mutex_unlock(&conn->mtx);
@@ -368,6 +379,7 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
     }
 
     switch (conn->param.type) {
+    case NEU_CONN_UDP_TO:
     case NEU_CONN_TCP_SERVER:
         zlog_fatal(conn->param.log, "neu_conn_recv cann't recv tcp server msg");
         assert(1 == 0);
@@ -395,19 +407,126 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
         }
         break;
     }
+    if (conn->param.type == NEU_CONN_TTY_CLIENT) {
+        if (ret == -1) {
+            zlog_error(
+                conn->param.log,
+                "tty conn fd: %d, recv buf len %zd, ret: %zd, errno: %s(%d)",
+                conn->fd, len, ret, strerror(errno), errno);
+            conn_disconnect(conn);
+        }
+    } else {
+        if (ret <= 0) {
+            zlog_error(conn->param.log,
+                       "conn fd: %d, recv buf len %zd, ret: %zd, errno: %s(%d)",
+                       conn->fd, len, ret, strerror(errno), errno);
+            if (ret == 0 || (ret == -1 && errno != EAGAIN)) {
+                conn_disconnect(conn);
+            }
+        }
+    }
+
+    pthread_mutex_unlock(&conn->mtx);
+
+    if (ret > 0) {
+        conn->state.recv_bytes += ret;
+    }
+
+    return ret;
+}
+ssize_t neu_conn_udp_sendto(neu_conn_t *conn, uint8_t *buf, ssize_t len,
+                            void *dst)
+{
+    ssize_t ret = 0;
+
+    pthread_mutex_lock(&conn->mtx);
+    if (conn->stop) {
+        pthread_mutex_unlock(&conn->mtx);
+        return ret;
+    }
+
+    if (!conn->is_connected) {
+        conn_connect(conn);
+    }
+
+    if (conn->is_connected) {
+        switch (conn->param.type) {
+        case NEU_CONN_TCP_CLIENT:
+        case NEU_CONN_TCP_SERVER:
+        case NEU_CONN_TTY_CLIENT:
+        case NEU_CONN_UDP:
+            assert(false);
+            break;
+        case NEU_CONN_UDP_TO:
+            if (conn->block) {
+                ret = sendto(conn->fd, buf, len, MSG_NOSIGNAL,
+                             (struct sockaddr *) dst, sizeof(struct sockaddr));
+            } else {
+                ret = sendto(conn->fd, buf, len, MSG_NOSIGNAL | MSG_DONTWAIT,
+                             (struct sockaddr *) dst, sizeof(struct sockaddr));
+            }
+            break;
+        }
+        if (ret != len) {
+            zlog_error(
+                conn->param.log,
+                "conn udp fd: %d, sendto (%s:%d) buf len: %zd, ret: %zd, "
+                "errno: %s(%d)",
+                conn->fd, inet_ntoa(((struct sockaddr_in *) dst)->sin_addr),
+                htons(((struct sockaddr_in *) dst)->sin_port), len, ret,
+                strerror(errno), errno);
+        }
+
+        if (ret == -1 && errno != EAGAIN) {
+            conn_disconnect(conn);
+        }
+
+        if (ret > 0 && conn->callback_trigger == false) {
+            conn->connected(conn->data, conn->fd);
+            conn->callback_trigger = true;
+        }
+    }
+
+    if (ret > 0) {
+        conn->state.send_bytes += ret;
+    }
+
+    pthread_mutex_unlock(&conn->mtx);
+
+    return ret;
+}
+
+ssize_t neu_conn_udp_recvfrom(neu_conn_t *conn, uint8_t *buf, ssize_t len,
+                              void *src)
+{
+    ssize_t   ret      = 0;
+    socklen_t addr_len = sizeof(struct sockaddr_in);
+
+    pthread_mutex_lock(&conn->mtx);
+    if (conn->stop) {
+        pthread_mutex_unlock(&conn->mtx);
+        return ret;
+    }
+
+    switch (conn->param.type) {
+    case NEU_CONN_TCP_SERVER:
+    case NEU_CONN_TCP_CLIENT:
+    case NEU_CONN_TTY_CLIENT:
+    case NEU_CONN_UDP:
+        assert(1 == 0);
+        break;
+    case NEU_CONN_UDP_TO:
+        ret =
+            recvfrom(conn->fd, buf, len, 0, (struct sockaddr *) src, &addr_len);
+        break;
+    }
     if (ret <= 0) {
         zlog_error(conn->param.log,
-                   "conn fd: %d, recv buf len %zd, ret: %zd, errno: %s(%d)",
+                   "conn udp fd: %d, recv buf len %zd, ret: %zd, errno: %s(%d)",
                    conn->fd, len, ret, strerror(errno), errno);
-    }
-
-    if (errno == EPIPE || ret == -1 || (ret == 0 && errno != 0)) {
-        conn_disconnect(conn);
-    }
-
-    if (ret > 0 && conn->callback_trigger == false) {
-        conn->connected(conn->data, conn->fd);
-        conn->callback_trigger = true;
+        if (ret == 0 || (ret == -1 && errno != EAGAIN)) {
+            conn_disconnect(conn);
+        }
     }
 
     pthread_mutex_unlock(&conn->mtx);
@@ -457,6 +576,9 @@ static void conn_free_param(neu_conn_t *conn)
         free(conn->param.params.udp.src_ip);
         free(conn->param.params.udp.dst_ip);
         break;
+    case NEU_CONN_UDP_TO:
+        free(conn->param.params.udpto.src_ip);
+        break;
     case NEU_CONN_TTY_CLIENT:
         free(conn->param.params.tty_client.device);
         break;
@@ -503,6 +625,12 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param)
         conn->param.params.udp.timeout  = param->params.udp.timeout;
         conn->block                     = conn->param.params.udp.timeout > 0;
         break;
+    case NEU_CONN_UDP_TO:
+        conn->param.params.udpto.src_ip   = strdup(param->params.udpto.src_ip);
+        conn->param.params.udpto.src_port = param->params.udpto.src_port;
+        conn->param.params.udpto.timeout  = param->params.udpto.timeout;
+        conn->block = conn->param.params.udpto.timeout > 0;
+        break;
     case NEU_CONN_TTY_CLIENT:
         conn->param.params.tty_client.device =
             strdup(param->params.tty_client.device);
@@ -687,10 +815,57 @@ static void conn_connect(neu_conn_t *conn)
         }
         break;
     }
+    case NEU_CONN_UDP_TO: {
+        if (conn->block) {
+            struct timeval tv = {
+                .tv_sec  = conn->param.params.udpto.timeout / 1000,
+                .tv_usec = (conn->param.params.udpto.timeout % 1000) * 1000,
+            };
+
+            fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
+            setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
+            setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv));
+        } else {
+            fd = socket(AF_INET, SOCK_DGRAM | SOCK_NONBLOCK, IPPROTO_UDP);
+        }
+        int so_broadcast = 1;
+        setsockopt(fd, SOL_SOCKET, SO_BROADCAST, &so_broadcast,
+                   sizeof(so_broadcast));
+        struct sockaddr_in local = {
+            .sin_family      = AF_INET,
+            .sin_port        = htons(conn->param.params.udpto.src_port),
+            .sin_addr.s_addr = inet_addr(conn->param.params.udpto.src_ip),
+        };
+
+        ret = bind(fd, (struct sockaddr *) &local, sizeof(struct sockaddr_in));
+        if (ret != 0) {
+            close(fd);
+            zlog_error(conn->param.log, "bind %s:%d error: %s(%d)",
+                       conn->param.params.udpto.src_ip,
+                       conn->param.params.udpto.src_port, strerror(errno),
+                       errno);
+            conn->is_connected = false;
+            return;
+        }
+
+        conn->is_connected = true;
+        conn->fd           = fd;
+
+        break;
+    }
     case NEU_CONN_TTY_CLIENT: {
         struct termios tty_opt = { 0 };
-
+#ifdef NEU_SMART_LINK
+#include "connection/neu_smart_link.h"
+        ret =
+            neu_conn_smart_link_auto_set(conn->param.params.tty_client.device);
+        zlog_notice(conn->param.log, "smart link ret: %d", ret);
+        if (ret > 0) {
+            fd = ret;
+        }
+#else
         fd = open(conn->param.params.tty_client.device, O_RDWR | O_NOCTTY, 0);
+#endif
         if (fd <= 0) {
             zlog_error(conn->param.log, "open %s error: %s(%d)",
                        conn->param.params.tty_client.device, strerror(errno),
@@ -846,6 +1021,15 @@ static void conn_connect(neu_conn_t *conn)
 
 static void conn_disconnect(neu_conn_t *conn)
 {
+    conn->is_connected  = false;
+    conn->connection_ok = false;
+    if (conn->callback_trigger == true) {
+        conn->disconnected(conn->data, conn->fd);
+        conn->callback_trigger = false;
+    }
+    conn->offset = 0;
+    memset(conn->buf, 0, conn->buf_size);
+
     switch (conn->param.type) {
     case NEU_CONN_TCP_SERVER:
         for (int i = 0; i < conn->param.params.tcp_server.max_link; i++) {
@@ -858,6 +1042,7 @@ static void conn_disconnect(neu_conn_t *conn)
         break;
     case NEU_CONN_TCP_CLIENT:
     case NEU_CONN_UDP:
+    case NEU_CONN_UDP_TO:
     case NEU_CONN_TTY_CLIENT:
         if (conn->fd > 0) {
             close(conn->fd);
@@ -865,14 +1050,6 @@ static void conn_disconnect(neu_conn_t *conn)
         }
         break;
     }
-
-    conn->is_connected = false;
-    if (conn->callback_trigger == true) {
-        conn->disconnected(conn->data, conn->fd);
-        conn->callback_trigger = false;
-    }
-    conn->offset = 0;
-    memset(conn->buf, 0, conn->buf_size);
 }
 
 static void conn_tcp_server_add_client(neu_conn_t *conn, int fd,
@@ -940,18 +1117,15 @@ int neu_conn_stream_consume(neu_conn_t *conn, void *context,
             if (used == 0) {
                 break;
             } else if (used == -1) {
-                memset(conn->buf, 0, conn->buf_size);
-                conn->offset = 0;
+                neu_conn_disconnect(conn);
                 break;
+            } else {
+                conn->offset -= used;
+                memmove(conn->buf, conn->buf + used, conn->offset);
+                neu_protocol_unpack_buf_init(&protocol_buf, conn->buf,
+                                             conn->offset);
             }
         }
-        if (conn->offset != 0) {
-            conn->offset -= neu_protocol_unpack_buf_used_size(&protocol_buf);
-            memmove(conn->buf,
-                    conn->buf +
-                        neu_protocol_unpack_buf_used_size(&protocol_buf),
-                    conn->offset);
-        }
     }
 
     return ret;
@@ -972,18 +1146,15 @@ int neu_conn_stream_tcp_server_consume(neu_conn_t *conn, int fd, void *context,
             if (used == 0) {
                 break;
             } else if (used == -1) {
-                memset(conn->buf, 0, conn->buf_size);
-                conn->offset = 0;
+                neu_conn_tcp_server_close_client(conn, fd);
                 break;
+            } else {
+                conn->offset -= used;
+                memmove(conn->buf, conn->buf + used, conn->offset);
+                neu_protocol_unpack_buf_init(&protocol_buf, conn->buf,
+                                             conn->offset);
             }
         }
-        if (conn->offset != 0) {
-            conn->offset -= neu_protocol_unpack_buf_used_size(&protocol_buf);
-            memmove(conn->buf,
-                    conn->buf +
-                        neu_protocol_unpack_buf_used_size(&protocol_buf),
-                    conn->offset);
-        }
     }
     return ret;
 }
@@ -1011,7 +1182,10 @@ int neu_conn_wait_msg(neu_conn_t *conn, void *context, uint16_t n_byte,
                            result.need, conn->buf_size, conn->offset);
                 return -1;
             }
+        } else if (result.need == 0) {
+            return result.used;
         } else {
+            neu_conn_disconnect(conn);
             return result.used;
         }
     }
@@ -1043,7 +1217,10 @@ int neu_conn_tcp_server_wait_msg(neu_conn_t *conn, int fd, void *context,
                            result.need, conn->buf_size, conn->offset);
                 return -1;
             }
+        } else if (result.need == 0) {
+            return result.used;
         } else {
+            neu_conn_tcp_server_close_client(conn, fd);
             return result.used;
         }
     }

src/connection/connection_eth.c

@@ -17,6 +17,7 @@
  * Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
  **/
 #include <assert.h>
+#include <errno.h>
 #include <pthread.h>
 #include <stdbool.h>
 #include <stdio.h>
@@ -32,6 +33,7 @@
 
 #include "connection/neu_connection_eth.h"
 #include "event/event.h"
+#include "type.h"
 #include "utils/log.h"
 #include "utils/utextend.h"
 
@@ -39,13 +41,12 @@
 #define ETH_P_LLDP 0x88CC
 #endif
 
-typedef struct neu_conn_eth_sub {
-    uint16_t protocol;
-    char     mac[ETH_ALEN];
-} callback_elem_key_t;
+struct neu_conn_eth_sub {
+    uint8_t mac[6];
+};
 
 typedef struct {
-    callback_elem_key_t key;
+    char mac[20];
 
     neu_conn_eth_msg_callback callback;
 
@@ -53,8 +54,6 @@ typedef struct {
 } callback_elem_t;
 
 typedef struct {
-    char device[NEU_NODE_NAME_LEN];
-
     neu_conn_eth_msg_callback callback;
 
     UT_hash_handle hh;
@@ -64,20 +63,16 @@ typedef struct interface_conn {
     char *        interface;
     neu_events_t *events;
     uint8_t       count;
-    uint8_t       buf[1500];
 
-    int     lldp_fd;
     int     profinet_fd;
     int     vlan_fd;
     uint8_t mac[ETH_ALEN];
 
     neu_event_io_t *profinet_event;
-    neu_event_io_t *lldp_event;
     neu_event_io_t *vlan_event;
 
-    lldp_callback_elem_t *lldp_callbacks;
-    callback_elem_t *     callbacks;
-    pthread_mutex_t       mtx;
+    callback_elem_t *callbacks;
+    pthread_mutex_t  mtx;
 } interface_conn_t;
 
 static pthread_mutex_t mtx      = { 0 };
@@ -91,13 +86,25 @@ struct neu_conn_eth {
 };
 
 static int get_mac(neu_conn_eth_t *conn, const char *interface);
-static int init_socket(const char *interface, uint16_t protocol);
+static int init_socket(const char *interface, uint16_t protocol,
+                       uint8_t mac[6]);
 static int uninit_socket(int fd);
 
-static int lldp_msg_cb(enum neu_event_io_type type, int fd, void *usr_data);
-static int eth_msg_cb(enum neu_event_io_type type, int fd, void *usr_data);
+static int     eth_msg_cb(enum neu_event_io_type type, int fd, void *usr_data);
+static uint8_t pf_dcp_broadcast[ETH_ALEN] = {
+    0x01, 0x0e, 0xcf, 0x00, 0x00, 0x01
+};
 
-static char *lldp_parse_port_id(uint16_t n_byte, uint8_t *bytes);
+int neu_conn_eth_size()
+{
+    int ret = 0;
+
+    for (int i = 0; i < interface_max; i++) {
+        ret += in_conns[i].count;
+    }
+
+    return ret;
+}
 
 neu_conn_eth_t *neu_conn_eth_init(const char *interface, void *ctx)
 {
@@ -126,17 +133,19 @@ neu_conn_eth_t *neu_conn_eth_init(const char *interface, void *ctx)
             if (in_conns[i].interface == NULL) {
                 neu_event_io_param_t param = { 0 };
 
+                pthread_mutex_init(&in_conns[i].mtx, NULL);
+                conn_eth->ic = &in_conns[i];
+
                 in_conns[i].interface = strdup(interface);
                 in_conns[i].events    = neu_event_new();
                 in_conns[i].count += 1;
 
-                in_conns[i].profinet_fd = init_socket(interface, 0x8892);
-                in_conns[i].lldp_fd     = init_socket(interface, ETH_P_LLDP);
-                in_conns[i].vlan_fd     = init_socket(interface, 0x8100);
-
                 get_mac(conn_eth, interface);
-                pthread_mutex_init(&in_conns[i].mtx, NULL);
-                conn_eth->ic = &in_conns[i];
+
+                in_conns[i].profinet_fd =
+                    init_socket(interface, 0x8892, in_conns[i].mac);
+                // in_conns[i].vlan_fd =
+                // init_socket(interface, 0x8100, in_conns[i].mac);
 
                 param.fd       = in_conns[i].profinet_fd;
                 param.usr_data = (void *) conn_eth;
@@ -145,17 +154,11 @@ neu_conn_eth_t *neu_conn_eth_init(const char *interface, void *ctx)
                 in_conns[i].profinet_event =
                     neu_event_add_io(in_conns[i].events, param);
 
-                param.fd = in_conns[i].lldp_fd;
-                param.cb = lldp_msg_cb;
+                // param.fd = in_conns[i].vlan_fd;
+                // param.cb = eth_msg_cb;
 
-                in_conns[i].lldp_event =
-                    neu_event_add_io(in_conns[i].events, param);
-
-                param.fd = in_conns[i].vlan_fd;
-                param.cb = eth_msg_cb;
-
-                in_conns[i].vlan_event =
-                    neu_event_add_io(in_conns[i].events, param);
+                // in_conns[i].vlan_event =
+                // neu_event_add_io(in_conns[i].events, param);
 
                 break;
             }
@@ -174,16 +177,15 @@ int neu_conn_eth_uninit(neu_conn_eth_t *conn)
     conn->ic->count -= 1;
     if (conn->ic->count == 0) {
         neu_event_del_io(conn->ic->events, conn->ic->profinet_event);
-        neu_event_del_io(conn->ic->events, conn->ic->lldp_event);
-        neu_event_del_io(conn->ic->events, conn->ic->vlan_event);
+        // neu_event_del_io(conn->ic->events, conn->ic->vlan_event);
 
         neu_event_close(conn->ic->events);
 
         uninit_socket(conn->ic->profinet_fd);
-        uninit_socket(conn->ic->lldp_fd);
-        uninit_socket(conn->ic->vlan_fd);
+        // uninit_socket(conn->ic->vlan_fd);
 
         free(conn->ic->interface);
+        conn->ic->interface = NULL;
         pthread_mutex_destroy(&conn->ic->mtx);
     }
     pthread_mutex_unlock(&mtx);
@@ -201,43 +203,46 @@ void neu_conn_eth_get_mac(neu_conn_eth_t *conn, uint8_t *mac)
 int neu_conn_eth_check_interface(const char *interface)
 {
     struct ifreq ifr = { 0 };
-    int          fd  = socket(AF_INET, SOCK_DGRAM, IPPROTO_IP);
+    int          fd  = socket(PF_PACKET, SOCK_RAW, ETH_P_IPV6);
     int          ret = -1;
 
+    if (fd <= 0) {
+        return -1;
+    }
+
     strcpy(ifr.ifr_name, interface);
     ret = ioctl(fd, SIOCGIFHWADDR, &ifr);
+    if (ret != 0) {
+        return -1;
+    }
 
     close(fd);
     return ret;
 }
 
-neu_conn_eth_sub_t *neu_conn_eth_register(neu_conn_eth_t *conn,
-                                          uint16_t protocol, uint8_t mac[6],
+neu_conn_eth_sub_t *neu_conn_eth_register(neu_conn_eth_t *conn, uint8_t xmac[6],
                                           neu_conn_eth_msg_callback callback)
 {
-    callback_elem_key_t key  = { 0 };
-    neu_conn_eth_sub_t *sub  = NULL;
-    callback_elem_t *   elem = NULL;
+    char                tmp_mac[20] = { 0 };
+    neu_conn_eth_sub_t *sub         = NULL;
+    callback_elem_t *   elem        = NULL;
 
-    key.protocol = protocol;
-    memcpy(key.mac, mac, ETH_ALEN);
+    snprintf(tmp_mac, sizeof(tmp_mac), "%X:%X:%X-%X:%X:%X", xmac[0], xmac[1],
+             xmac[2], xmac[3], xmac[4], xmac[5]);
 
     pthread_mutex_lock(&conn->ic->mtx);
 
-    HASH_FIND(hh, conn->ic->callbacks, &key, sizeof(callback_elem_key_t), elem);
+    HASH_FIND_STR(conn->ic->callbacks, tmp_mac, elem);
     if (elem == NULL) {
         elem = calloc(1, sizeof(callback_elem_t));
         sub  = calloc(1, sizeof(neu_conn_eth_sub_t));
 
-        elem->key.protocol = protocol;
-        memcpy(key.mac, mac, ETH_ALEN);
         elem->callback = callback;
+        memcpy(elem->mac, tmp_mac, sizeof(elem->mac));
 
-        memcpy(sub->mac, mac, ETH_ALEN);
-        sub->protocol = protocol;
+        memcpy(sub->mac, xmac, ETH_ALEN);
 
-        HASH_ADD(hh, conn->ic->callbacks, key, sizeof(callback_elem_key_t),
-                 elem);
+        HASH_ADD_STR(conn->ic->callbacks, mac, elem);
     }
 
     pthread_mutex_unlock(&conn->ic->mtx);
@@ -247,15 +252,15 @@ neu_conn_eth_sub_t *neu_conn_eth_register(neu_conn_eth_t *conn,
 
 int neu_conn_eth_unregister(neu_conn_eth_t *conn, neu_conn_eth_sub_t *sub)
 {
-    callback_elem_key_t key  = { 0 };
-    callback_elem_t *   elem = NULL;
+    char             tmp_mac[20] = { 0 };
+    callback_elem_t *elem        = NULL;
 
-    key.protocol = sub->protocol;
-    memcpy(key.mac, sub->mac, ETH_ALEN);
+    snprintf(tmp_mac, sizeof(tmp_mac), "%X:%X:%X-%X:%X:%X", sub->mac[0],
+             sub->mac[1], sub->mac[2], sub->mac[3], sub->mac[4], sub->mac[5]);
 
     pthread_mutex_lock(&conn->ic->mtx);
 
-    HASH_FIND(hh, conn->ic->callbacks, &key, sizeof(callback_elem_key_t), elem);
+    HASH_FIND_STR(conn->ic->callbacks, tmp_mac, elem);
     if (elem != NULL) {
         HASH_DEL(conn->ic->callbacks, elem);
         free(elem);
@@ -268,61 +273,17 @@ int neu_conn_eth_unregister(neu_conn_eth_t *conn, neu_conn_eth_sub_t *sub)
     return 0;
 }
 
-int neu_conn_eth_register_lldp(neu_conn_eth_t *conn, const char *device,
-                               neu_conn_eth_msg_callback callback)
+int neu_conn_eth_send(neu_conn_eth_t *conn, uint16_t protocol, uint16_t n_byte,
+                      uint8_t *bytes)
 {
-    lldp_callback_elem_t *elem = NULL;
-    int                   ret  = 0;
-
-    pthread_mutex_lock(&conn->ic->mtx);
-
-    HASH_FIND_STR(conn->ic->lldp_callbacks, device, elem);
-    if (elem != NULL) {
-        ret = -1;
-    } else {
-        elem = calloc(1, sizeof(lldp_callback_elem_t));
-
-        elem->callback = callback;
-        strcpy(elem->device, device);
-
-        HASH_ADD_STR(conn->ic->lldp_callbacks, device, elem);
-    }
-
-    pthread_mutex_unlock(&conn->ic->mtx);
-
-    return ret;
-}
-
-int neu_conn_eth_unregister_lldp(neu_conn_eth_t *conn, const char *device)
-{
-    lldp_callback_elem_t *elem = NULL;
-
-    pthread_mutex_lock(&conn->ic->mtx);
-    HASH_FIND_STR(conn->ic->lldp_callbacks, device, elem);
-    if (elem != NULL) {
-        HASH_DEL(conn->ic->lldp_callbacks, elem);
-        free(elem);
-    }
-    pthread_mutex_unlock(&conn->ic->mtx);
-
-    return 0;
-}
-
-int neu_conn_eth_send(neu_conn_eth_t *conn, uint8_t dst_mac[6],
-                      uint16_t protocol, uint16_t n_byte, uint8_t *bytes)
-{
-    int            fd   = 0;
-    int            ret  = 0;
-    struct ethhdr *ehdr = (struct ethhdr *) bytes;
-    (void) dst_mac;
+    int fd  = 0;
+    int ret = 0;
 
     switch (protocol) {
+    case ETH_P_LLDP:
     case 0x8892:
         fd = conn->ic->profinet_fd;
         break;
-    case ETH_P_LLDP:
-        fd = conn->ic->lldp_fd;
-        break;
     case 0x8100:
         fd = conn->ic->vlan_fd;
         break;
@@ -331,8 +292,6 @@ int neu_conn_eth_send(neu_conn_eth_t *conn, uint8_t dst_mac[6],
         break;
     }
 
-    memcpy(ehdr->h_source, conn->ic->mac, ETH_ALEN);
-
     ret = send(fd, bytes, n_byte, 0);
     return ret;
 }
@@ -353,20 +312,22 @@ static int get_mac(neu_conn_eth_t *conn, const char *interface)
     return ret;
 }
 
-static int init_socket(const char *interface, uint16_t protocol)
+static int init_socket(const char *interface, uint16_t protocol, uint8_t mac[6])
 {
     struct ifreq       ifr                     = { 0 };
     struct sockaddr_ll sll                     = { 0 };
-    struct packet_mreq mreq                    = { 0 };
     const uint8_t      pn_mcast_addr[ETH_ALEN] = {
         0x01, 0x0e, 0xcf, 0x00, 0x00, 0x00
     };
     int fd  = socket(PF_PACKET, SOCK_RAW, htons(protocol));
-    int ret = 0;
+    int ret = 1;
 
     assert(fd > 0);
     setsockopt(fd, SOL_SOCKET, SO_DONTROUTE, &ret, sizeof(ret));
 
+    (void) mac;
+    (void) pn_mcast_addr;
+
     strcpy(ifr.ifr_name, interface);
     if (ioctl(fd, SIOCGIFINDEX, &ifr) != 0) {
         close(fd);
@@ -376,25 +337,12 @@ static int init_socket(const char *interface, uint16_t protocol)
     sll.sll_family   = AF_PACKET;
     sll.sll_ifindex  = ifr.ifr_ifindex;
     sll.sll_protocol = htons(protocol);
-
-    mreq.mr_ifindex = ifr.ifr_ifindex;
-    mreq.mr_type    = PACKET_HOST | PACKET_MR_MULTICAST;
-    mreq.mr_alen    = ETH_ALEN;
-    memcpy(mreq.mr_address, pn_mcast_addr, ETH_ALEN);
-
-    ifr.ifr_flags = 0;
-    ioctl(fd, SIOCGIFFLAGS, &ifr);
-
-    bind(fd, (struct sockaddr *) &sll, sizeof(sll));
-
-    if (protocol == 0x8892) {
-        ret = setsockopt(fd, SOL_PACKET, PACKET_ADD_MEMBERSHIP, &mreq,
-                         sizeof(mreq));
-        assert(ret == 0);
-    }
+    ret              = bind(fd, (struct sockaddr *) &sll, sizeof(sll));
+    assert(ret == 0);
 
     return fd;
 }
+
 static int uninit_socket(int fd)
 {
     close(fd);
@@ -408,27 +356,53 @@ static int eth_msg_cb(enum neu_event_io_type type, int fd, void *usr_data)
 
     switch (type) {
     case NEU_EVENT_IO_READ: {
-        callback_elem_key_t key  = { 0 };
-        callback_elem_t *   elem = NULL;
-
-        memset(conn->ic->buf, 0, sizeof(conn->ic->buf));
-        int ret = recv(fd, conn->ic->buf, sizeof(conn->ic->buf), 0);
+        callback_elem_t *elem      = NULL;
+        uint8_t          buf[1500] = { 0 };
+        int              ret       = recv(fd, buf, 1500, 0);
 
         if (ret < ETH_HLEN) {
-            nlog_warn("eth recv length < %d msg", ETH_HLEN);
             return 0;
         }
 
-        struct ethhdr *ehdr = (struct ethhdr *) conn->ic->buf;
+        struct ethhdr *ehdr = (struct ethhdr *) buf;
+        ehdr->h_proto       = ntohs(ehdr->h_proto);
 
-        key.protocol = ntohs(ehdr->h_proto);
-        memcpy(key.mac, ehdr->h_source, ETH_ALEN);
+        if (ehdr->h_proto != 0x8100 && ehdr->h_proto != 0x8892) {
+            return 0;
+        }
 
-        HASH_FIND(hh, conn->ic->callbacks, &key, sizeof(key), elem);
+        if (memcmp(ehdr->h_dest, pf_dcp_broadcast, ETH_ALEN) == 0) {
+            callback_elem_t *tmp = NULL;
 
-        if (elem != NULL) {
-            elem->callback(conn, conn->ctx, key.protocol, ret - ETH_HLEN,
-                           conn->ic->buf + ETH_HLEN, ehdr->h_source);
+            HASH_ITER(hh, conn->ic->callbacks, elem, tmp)
+            {
+                elem->callback(conn, conn->ctx, ehdr->h_proto, ret - ETH_HLEN,
+                               buf + ETH_HLEN, ehdr->h_source);
+            }
+        } else {
+            if (memcmp(ehdr->h_dest, conn->ic->mac, ETH_ALEN) != 0) {
+                return 0;
+            }
+            char tmp_mac[20] = { 0 };
+
+            snprintf(tmp_mac, sizeof(tmp_mac), "%X:%X:%X-%X:%X:%X",
+                     ehdr->h_source[0], ehdr->h_source[1], ehdr->h_source[2],
+                     ehdr->h_source[3], ehdr->h_source[4], ehdr->h_source[5]);
+
+            HASH_FIND_STR(conn->ic->callbacks, tmp_mac, elem);
+            if (elem != NULL) {
+                elem->callback(conn, conn->ctx, ehdr->h_proto, ret - ETH_HLEN,
+                               buf + ETH_HLEN, ehdr->h_source);
+            } else {
+                memset(tmp_mac, 0, sizeof(tmp_mac));
+                snprintf(tmp_mac, sizeof(tmp_mac), "0:0:0-0:0:0");
+                HASH_FIND_STR(conn->ic->callbacks, tmp_mac, elem);
+                if (elem != NULL) {
+                    elem->callback(conn, conn->ctx, ehdr->h_proto,
+                                   ret - ETH_HLEN, buf + ETH_HLEN,
+                                   ehdr->h_source);
+                }
+            }
         }
 
         break;
@@ -442,70 +416,3 @@ static int eth_msg_cb(enum neu_event_io_type type, int fd, void *usr_data)
 
     return 0;
 }
-
-static int lldp_msg_cb(enum neu_event_io_type type, int fd, void *usr_data)
-{
-    neu_conn_eth_t *conn    = (neu_conn_eth_t *) usr_data;
-    int             ret     = 0;
-    char *          port_id = NULL;
-
-    switch (type) {
-    case NEU_EVENT_IO_READ:
-        memset(conn->ic->buf, 0, sizeof(conn->ic->buf));
-        ret     = recv(fd, conn->ic->buf, sizeof(conn->ic->buf), 0);
-        port_id = lldp_parse_port_id(ret - sizeof(struct ethhdr),
-                                     conn->ic->buf + sizeof(struct ethhdr));
-
-        if (port_id != NULL) {
-            lldp_callback_elem_t *elem = NULL;
-            struct ethhdr *       ehdr = (struct ethhdr *) conn->ic->buf;
-
-            HASH_FIND_STR(conn->ic->callbacks, port_id, elem);
-            if (elem != NULL) {
-                elem->callback(
-                    conn, conn->ctx, ETH_P_LLDP, ret - sizeof(struct ethhdr),
-                    conn->ic->buf + sizeof(struct ethhdr), ehdr->h_source);
-            }
-        }
-
-        break;
-    case NEU_EVENT_IO_CLOSED:
-    case NEU_EVENT_IO_HUP:
-        nlog_warn("eth conn lldp error type: %d, fd: %d(%s)", type, fd,
-                  conn->ic->interface);
-        break;
-    }
-
-    return 0;
-}
-
-struct lldp_tvl {
-    uint16_t type : 7;
-    uint16_t length : 9;
-} __attribute__((packed));
-
-#define END_OF_LLDPDU_TLV_TYPE 0x00
-#define CHASSIS_ID_TLV_TYPE 0x01
-#define PORT_ID_TLV_TYPE 0x02
-#define TIME_TO_LIVE_TLV_TYPE 0x03
-
-static char *lldp_parse_port_id(uint16_t n_byte, uint8_t *bytes)
-{
-    struct lldp_tvl *tvl = (struct lldp_tvl *) bytes;
-
-    if (n_byte < sizeof(struct lldp_tvl) + tvl->length) {
-        return NULL;
-    }
-
-    if (tvl->type == PORT_ID_TLV_TYPE) {
-        static __thread char port_id[NEU_NODE_NAME_LEN] = { 0 };
-        uint8_t *            data                       = (uint8_t *) &tvl[1];
-
-        strncpy(port_id, (char *) data + 1, tvl->length - 1);
-        return port_id;
-    } else {
-        return lldp_parse_port_id(
-            n_byte - sizeof(struct lldp_tvl) - tvl->length,
-            bytes + sizeof(struct lldp_tvl) + tvl->length);
-    }
-}

src/connection/smart_link.c

@@ -0,0 +1,286 @@
+#include <errno.h>
+#include <fcntl.h>
+#include <linux/serial.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/ioctl.h>
+#include <sys/stat.h>
+#include <sys/types.h>
+#include <termios.h>
+#include <unistd.h>
+
+#include <jansson.h>
+
+#include "json/json.h"
+
+#include "connection/neu_smart_link.h"
+
+#define DEVICE_CONF_PATH "/etc/config/device.conf"
+#define SERIAL_CONF_PATH "/etc/config/serial.conf"
+#define SMART_LINK_PATH "/dev/smartlink"
+
+#define TYPE 'S'
+#define UART_RS232 _IOWR(TYPE, 1, int)
+#define UART_RS485 _IOWR(TYPE, 0, int)
+
+enum SERIAL_MODE { SM_RS232, SM_RS422, SM_RS485 };
+
+static int get_port(const char *path);
+static int get_port_property(int port, int *mode, int *baudrate);
+static int read_file(const char *path, char **buf);
+static int set_serial_mode(const char *dev_path, int baudrate, int mode);
+
+int neu_conn_smart_link_auto_set(const char *dev_path)
+{
+    int mode     = 0;
+    int baudrate = 0;
+    int port     = get_port(dev_path);
+    if (port <= 0) {
+        return -1;
+    }
+    printf("get port %d\n", port);
+
+    int ret = get_port_property(port, &mode, &baudrate);
+    if (ret != 0) {
+        return -1;
+    }
+
+    printf("get port property %d %d\n", mode, baudrate);
+
+    ret = set_serial_mode(dev_path, baudrate, mode);
+
+    return ret;
+}
+
+static int read_file(const char *path, char **buf)
+{
+    FILE *fp = NULL;
+
+    fp = fopen(path, "r");
+    if (fp == NULL) {
+        return -1;
+    }
+
+    fseek(fp, 0, SEEK_END);
+    int size = ftell(fp);
+    rewind(fp);
+
+    *buf = calloc(1, size + 1);
+    if (*buf == NULL) {
+        fclose(fp);
+        return -1;
+    }
+
+    int ret = fread(*buf, 1, size, fp);
+    if (ret != size) {
+        fclose(fp);
+        free(*buf);
+        return -1;
+    }
+    fclose(fp);
+
+    return ret;
+}
+
+static int get_port(const char *dev_path)
+{
+    json_t *     root = NULL;
+    json_error_t error;
+    char *       buf = NULL;
+    int          ret = read_file(DEVICE_CONF_PATH, &buf);
+    if (ret <= 0) {
+        return -1;
+    }
+
+    root = json_loads(buf, JSON_DECODE_ANY, &error);
+    if (root == NULL) {
+        free(buf);
+        return -1;
+    }
+
+    neu_json_elem_t e_tty = {
+        .name = "tty_name",
+        .t    = NEU_JSON_OBJECT,
+    };
+
+    ret = neu_json_decode_value(root, &e_tty);
+    if (ret != 0 || json_is_array((json_t *) e_tty.v.val_object) == false) {
+        free(buf);
+        json_decref(root);
+        return -1;
+    }
+
+    int     index = 0;
+    json_t *e_dev = NULL;
+    json_array_foreach(e_tty.v.val_object, index, e_dev)
+    {
+        neu_json_elem_t e_dd = {
+            .name = "device",
+            .t    = NEU_JSON_STR,
+        };
+
+        ret = neu_json_decode_value(e_dev, &e_dd);
+        if (ret != 0) {
+            free(buf);
+            json_decref(root);
+            return -1;
+        }
+
+        if (strcmp(e_dd.v.val_str, dev_path) == 0) {
+            neu_json_elem_t e_port = {
+                .name = "port",
+                .t    = NEU_JSON_INT,
+            };
+
+            ret = neu_json_decode_value(e_dev, &e_port);
+            if (ret != 0) {
+                free(buf);
+                json_decref(root);
+                return -1;
+            }
+
+            free(buf);
+            json_decref(root);
+            free(e_dd.v.val_str);
+            return e_port.v.val_int;
+        }
+        free(e_dd.v.val_str);
+    }
+
+    free(buf);
+    json_decref(root);
+    return -1;
+}
+
+static int get_port_property(int port, int *mode, int *baudrate)
+{
+    json_t *     root = NULL;
+    json_error_t error;
+    char *       buf = NULL;
+    int          ret = read_file(SERIAL_CONF_PATH, &buf);
+    if (ret <= 0) {
+        return -1;
+    }
+
+    root = json_loads(buf, JSON_DECODE_ANY, &error);
+    if (root == NULL) {
+        free(buf);
+        return -1;
+    }
+
+    json_t *e_port = json_array_get(root, port - 1);
+    if (e_port == NULL) {
+        json_decref(root);
+        free(buf);
+        return -1;
+    }
+
+    neu_json_elem_t elems[] = {
+        {
+            .name = "mode",
+            .t    = NEU_JSON_INT,
+        },
+        {
+            .name = "baudrate",
+            .t    = NEU_JSON_INT,
+        },
+    };
+
+    ret = neu_json_decode_by_json(e_port, NEU_JSON_ELEM_SIZE(elems), elems);
+    if (ret != 0) {
+        free(buf);
+        json_decref(root);
+        return -1;
+    }
+
+    *mode     = elems[0].v.val_int;
+    *baudrate = elems[1].v.val_int;
+
+    json_decref(root);
+    free(buf);
+    return 0;
+}
+
+static int set_serial_mode(const char *dev_path, int baudrate, int mode)
+{
+    static pthread_mutex_t mode_switch_mtx = PTHREAD_MUTEX_INITIALIZER;
+    pthread_mutex_lock(&mode_switch_mtx);
+
+    int serial_fd, switch_fd;
+    int tty_num, para;
+    int ret = -1;
+
+    struct serial_rs485 conf;
+    conf.flags &= !SER_RS485_ENABLED;
+
+    switch (baudrate) {
+    case 38400:
+        conf.delay_rts_after_send = 1;
+        break;
+    case 57600:
+        conf.delay_rts_after_send = 4;
+        break;
+    case 115200:
+        conf.delay_rts_after_send = 6;
+        break;
+    default:
+        conf.delay_rts_after_send = 0;
+        break;
+    }
+
+    if ((serial_fd = open(dev_path, O_RDWR | O_NOCTTY)) < 0) {
+        fprintf(stderr, "open %s failed, fd: %d, strerr: %s\n", dev_path,
+                serial_fd, strerror(errno));
+        goto out;
+    }
+
+    if (mode != SM_RS232) {
+        if (ioctl(serial_fd, TIOCMGET, &para) != 0) {
+            fprintf(stderr, "ioctl failed, strerr: %s\n", strerror(errno));
+            goto out2;
+        }
+        if (mode == SM_RS422) {
+            para |= TIOCM_DTR;
+            conf.flags |= SER_RS485_ENABLED | SER_RS485_RTS_ON_SEND;
+        } else if (mode == SM_RS485) {
+            para &= ~TIOCM_DTR;
+            conf.flags |= SER_RS485_ENABLED | SER_RS485_RTS_ON_SEND;
+        }
+        if (ioctl(serial_fd, TIOCMSET, &para) != 0) {
+            fprintf(stderr, "ioctl failed, strerr: %s\n", strerror(errno));
+            goto out2;
+        }
+    }
+    ioctl(serial_fd, TIOCSRS485, &conf);
+
+    if ((switch_fd = open(SMART_LINK_PATH, O_RDWR)) < 0) {
+        fprintf(stderr, "open %s failed, ret: %d, strerr: %s\n",
+                SMART_LINK_PATH, switch_fd, strerror(errno));
+        goto out2;
+    }
+
+    sscanf(dev_path, "%*[a-zA-Z/]%d", &tty_num);
+    printf("tty_num: %d\n", tty_num);
+    if (mode == SM_RS232) {
+        ret = ioctl(switch_fd, UART_RS232, tty_num);
+    } else {
+        ret = ioctl(switch_fd, UART_RS485, tty_num);
+    }
+    if (ret < 0) {
+        fprintf(stderr, "ioctl failed, ret: %d, strerr: %s\n", ret,
+                strerror(errno));
+    }
+
+    close(switch_fd);
+    ret = serial_fd;
+    pthread_mutex_unlock(&mode_switch_mtx);
+    return ret;
+out2:
+    close(serial_fd);
+    serial_fd = -1;
+out:
+    pthread_mutex_unlock(&mode_switch_mtx);
+    return ret;
+}

src/core/manager.c

@@ -267,10 +267,14 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
 
         neu_node_manager_update(manager->node_manager, init->node, pipe);
 
-        if (init->auto_start) {
+        if (init->state == NEU_NODE_RUNNING_STATE_STOPPED ||
+            init->state == NEU_NODE_RUNNING_STATE_RUNNING) {
             neu_adapter_t *adapter =
                 neu_node_manager_find(manager->node_manager, init->node);
             neu_adapter_start(adapter);
+            if (init->state == NEU_NODE_RUNNING_STATE_STOPPED) {
+                neu_adapter_stop(adapter);
+            }
         }
 
         nlog_notice("bind node %s to pipe(%d)", init->node, pipe.id);

src/core/manager_internal.c

@@ -43,7 +43,8 @@ UT_array *neu_manager_get_plugins(neu_manager_t *manager)
 }
 
 int neu_manager_add_node(neu_manager_t *manager, const char *node_name,
-                         const char *plugin_name, bool start)
+                         const char *             plugin_name,
+                         neu_node_running_state_e state)
 {
     neu_adapter_t *       adapter      = NULL;
     neu_plugin_instance_t instance     = { 0 };
@@ -77,7 +78,7 @@ int neu_manager_add_node(neu_manager_t *manager, const char *node_name,
 
     adapter = neu_adapter_create(&adapter_info);
     neu_node_manager_add(manager->node_manager, adapter);
-    neu_adapter_init(adapter, start);
+    neu_adapter_init(adapter, state);
 
     return NEU_ERR_SUCCESS;
 }

src/core/manager_internal.h

@@ -52,7 +52,8 @@ int       neu_manager_del_plugin(neu_manager_t *manager, const char *plugin);
 UT_array *neu_manager_get_plugins(neu_manager_t *manager);
 
 int       neu_manager_add_node(neu_manager_t *manager, const char *node_name,
-                               const char *plugin_name, bool start);
+                               const char *             plugin_name,
+                               neu_node_running_state_e state);
 int       neu_manager_del_node(neu_manager_t *manager, const char *node_name);
 UT_array *neu_manager_get_nodes(neu_manager_t *manager, neu_node_type_e type,
                                 const char *plugin, const char *node);

src/core/storage.c

@@ -118,9 +118,8 @@ int manager_load_node(neu_manager_t *manager)
 
     utarray_foreach(node_infos, neu_persist_node_info_t *, node_info)
     {
-        rv = neu_manager_add_node(
-            manager, node_info->name, node_info->plugin_name,
-            node_info->state == NEU_NODE_RUNNING_STATE_RUNNING);
+        rv                    = neu_manager_add_node(manager, node_info->name,
+                                  node_info->plugin_name, node_info->state);
         const char *ok_or_err = (0 == rv) ? "success" : "fail";
         nlog_notice("load adapter %s type:%d, name:%s plugin:%s state:%d",
                     ok_or_err, node_info->type, node_info->name,

src/parser/neu_json_tag.c

@@ -199,7 +199,8 @@ int neu_json_tag_check_type(neu_json_tag_t *tag)
             return NEU_JSON_BOOL == tag->t;
         case NEU_TYPE_BYTES:
         case NEU_TYPE_STRING:
-            return NEU_JSON_STR == tag->t;
+            return NEU_JSON_STR == tag->t &&
+                strlen(tag->value.val_str) <= NEU_VALUE_SIZE;
         }
     } else {
         // non-static tag should have no initial value

src/persist/persist.c

@@ -25,6 +25,7 @@
 #include <dirent.h>
 #include <errno.h>
 #include <fcntl.h>
+#include <pthread.h>
 #include <sys/stat.h>
 #include <sys/types.h>
 #include <unistd.h>
@@ -49,9 +50,12 @@
 
 #define PATH_MAX_SIZE 128
 
-static const char *plugin_file = "persistence/plugins.json";
-static const char *db_file     = "persistence/sqlite.db";
-static sqlite3 *   global_db   = NULL;
+static const char *plugin_file       = "persistence/plugins.json";
+static const char *db_file           = "persistence/sqlite.db";
+static sqlite3 *   global_db         = NULL;
+pthread_rwlock_t   global_rwlock     = PTHREAD_RWLOCK_INITIALIZER;
+static int         global_node_count = 0;
+static int         global_tag_count  = 0;
 
 static inline bool ends_with(const char *str, const char *suffix)
 {
@@ -474,6 +478,44 @@ static int apply_schemas(sqlite3 *db, const char *dir)
     return rv;
 }
 
+static inline int node_count_load()
+{
+    int           rv    = 0;
+    sqlite3_stmt *stmt  = NULL;
+    char *        query = "select count(*) from nodes";
+
+    if (SQLITE_OK == sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL) &&
+        SQLITE_ROW == sqlite3_step(stmt)) {
+        pthread_rwlock_wrlock(&global_rwlock);
+        global_node_count = sqlite3_column_int(stmt, 0);
+        pthread_rwlock_unlock(&global_rwlock);
+    } else {
+        rv = -1;
+    }
+
+    sqlite3_finalize(stmt);
+    return rv;
+}
+
+static inline int tag_count_load()
+{
+    int           rv    = 0;
+    sqlite3_stmt *stmt  = NULL;
+    char *        query = "select count(*) from tags";
+
+    if (SQLITE_OK == sqlite3_prepare_v2(global_db, query, -1, &stmt, NULL) &&
+        SQLITE_ROW == sqlite3_step(stmt)) {
+        pthread_rwlock_wrlock(&global_rwlock);
+        global_tag_count = sqlite3_column_int(stmt, 0);
+        pthread_rwlock_unlock(&global_rwlock);
+    } else {
+        rv = -1;
+    }
+
+    sqlite3_finalize(stmt);
+    return rv;
+}
+
 int neu_persister_create(const char *schema_dir)
 {
     int rv = sqlite3_open(db_file, &global_db);
@@ -506,6 +548,12 @@ int neu_persister_create(const char *schema_dir)
         return -1;
     }
 
+    if (0 != node_count_load() || 0 != tag_count_load()) {
+        nlog_fatal("load node/tag count fail");
+        sqlite3_close(global_db);
+        return -1;
+    }
+
     return 0;
 }
 
@@ -514,6 +562,36 @@ sqlite3 *neu_persister_get_db()
     return global_db;
 }
 
+static inline void node_count_add(int n)
+{
+    pthread_rwlock_wrlock(&global_rwlock);
+    global_node_count += n;
+    pthread_rwlock_unlock(&global_rwlock);
+}
+
+int neu_persister_node_count()
+{
+    pthread_rwlock_rdlock(&global_rwlock);
+    int cnt = global_node_count;
+    pthread_rwlock_unlock(&global_rwlock);
+    return cnt;
+}
+
+static inline void tag_count_add(int n)
+{
+    pthread_rwlock_wrlock(&global_rwlock);
+    global_tag_count += n;
+    pthread_rwlock_unlock(&global_rwlock);
+}
+
+int neu_persister_tag_count()
+{
+    pthread_rwlock_rdlock(&global_rwlock);
+    int cnt = global_tag_count;
+    pthread_rwlock_unlock(&global_rwlock);
+    return cnt;
+}
+
 void neu_persister_destroy()
 {
     sqlite3_close(global_db);
@@ -521,10 +599,15 @@ void neu_persister_destroy()
 
 int neu_persister_store_node(neu_persist_node_info_t *info)
 {
-    return execute_sql(global_db,
-                       "INSERT INTO nodes (name, type, state, plugin_name) "
-                       "VALUES (%Q, %i, %i, %Q)",
-                       info->name, info->type, info->state, info->plugin_name);
+    int rv =
+        execute_sql(global_db,
+                    "INSERT INTO nodes (name, type, state, plugin_name) "
+                    "VALUES (%Q, %i, %i, %Q)",
+                    info->name, info->type, info->state, info->plugin_name);
+    if (0 == rv) {
+        node_count_add(1);
+    }
+    return rv;
 }
 
 static UT_icd node_info_icd = {
@@ -585,8 +668,13 @@ int neu_persister_delete_node(const char *node_name)
 {
     // rely on foreign key constraints to remove settings, groups, tags and
     // subscriptions
-    return execute_sql(global_db, "DELETE FROM nodes WHERE name=%Q;",
-                       node_name);
+    int rv =
+        execute_sql(global_db, "DELETE FROM nodes WHERE name=%Q;", node_name);
+    if (0 == rv) {
+        node_count_add(-1);
+        tag_count_load();
+    }
+    return rv;
 }
 
 int neu_persister_update_node(const char *node_name, const char *new_name)
@@ -716,6 +804,9 @@ int neu_persister_store_tag(const char *driver_name, const char *group_name,
                          tag->attribute, tag->precision, tag->type,
                          tag->decimal, tag->description, val_str);
 
+    if (0 == rv) {
+        tag_count_add(1);
+    }
     free(val_str);
     return rv;
 }
@@ -821,6 +912,7 @@ int neu_persister_store_tags(const char *driver_name, const char *group_name,
         goto error;
     }
 
+    tag_count_add(n);
     sqlite3_finalize(stmt);
     return 0;
 
@@ -926,10 +1018,14 @@ int neu_persister_update_tag_value(const char *         driver_name,
 int neu_persister_delete_tag(const char *driver_name, const char *group_name,
                              const char *tag_name)
 {
-    return execute_sql(
+    int rv = execute_sql(
         global_db,
         "DELETE FROM tags WHERE driver_name=%Q AND group_name=%Q AND name=%Q",
         driver_name, group_name, tag_name);
+    if (0 == rv) {
+        tag_count_add(-1);
+    }
+    return rv;
 }
 
 int neu_persister_store_subscription(const char *app_name,
@@ -1101,9 +1197,13 @@ error:
 int neu_persister_delete_group(const char *driver_name, const char *group_name)
 {
     // rely on foreign key constraints to delete tags and subscriptions
-    return execute_sql(global_db,
-                       "DELETE FROM groups WHERE driver_name=%Q AND name=%Q",
-                       driver_name, group_name);
+    int rv = execute_sql(global_db,
+                         "DELETE FROM groups WHERE driver_name=%Q AND name=%Q",
+                         driver_name, group_name);
+    if (0 == rv) {
+        tag_count_load();
+    }
+    return rv;
 }
 
 int neu_persister_store_node_setting(const char *node_name, const char *setting)

src/utils/http.c

@@ -295,6 +295,8 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content)
     case NEU_ERR_MQTT_CONNECT_FAILURE:
     case NEU_ERR_MQTT_SUBSCRIBE_FAILURE:
     case NEU_ERR_MQTT_PUBLISH_FAILURE:
+    case NEU_ERR_STRING_TOO_LONG:
+    case NEU_ERR_FILE_WRITE_FAILURE:
         status = NNG_HTTP_STATUS_OK;
         break;
     case NEU_ERR_EINTERNAL:

version

@@ -1 +1 @@
-2.4.7
+2.4.13