Merge pull request #1364 from lixiumei123/v2.4

version:2.4.9

.github/workflows/ci.yml

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

include/neuron/errcodes.h

@@ -133,7 +133,7 @@ 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,

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;
@@ -85,6 +88,14 @@ int file_group_timer(neu_plugin_t *plugin, neu_plugin_group_t *group)
             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_STRING_TOO_LONG;
+
             fclose(f);
             goto dvalue_result;
         }
@@ -121,19 +132,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/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/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/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;
@@ -338,8 +341,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 +359,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);
@@ -395,19 +405,23 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
         }
         break;
     }
-    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 (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 (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);
@@ -487,6 +501,10 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param)
         } else {
             conn->block = false;
         }
+        zlog_notice(conn->param.log, "tcp server %s:%d, timeout: %d, block: %d",
+                    conn->param.params.tcp_server.ip,
+                    conn->param.params.tcp_server.port,
+                    conn->param.params.tcp_server.timeout, conn->block);
         break;
     case NEU_CONN_TCP_CLIENT:
         conn->param.params.tcp_client.ip = strdup(param->params.tcp_client.ip);
@@ -494,6 +512,10 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param)
         conn->param.params.tcp_client.timeout =
             param->params.tcp_client.timeout;
         conn->block = conn->param.params.tcp_client.timeout > 0;
+        zlog_notice(conn->param.log, "tcp client %s:%d, timeout: %d, block: %d",
+                    conn->param.params.tcp_client.ip,
+                    conn->param.params.tcp_client.port,
+                    conn->param.params.tcp_client.timeout, conn->block);
         break;
     case NEU_CONN_UDP:
         conn->param.params.udp.src_ip   = strdup(param->params.udp.src_ip);
@@ -502,6 +524,11 @@ static void conn_init_param(neu_conn_t *conn, neu_conn_param_t *param)
         conn->param.params.udp.dst_port = param->params.udp.dst_port;
         conn->param.params.udp.timeout  = param->params.udp.timeout;
         conn->block                     = conn->param.params.udp.timeout > 0;
+        zlog_notice(
+            conn->param.log, "udp %s:%d -> %s:%d, timeout: %d, block: %d",
+            conn->param.params.udp.src_ip, conn->param.params.udp.src_port,
+            conn->param.params.udp.dst_ip, conn->param.params.udp.dst_port,
+            conn->param.params.udp.timeout, conn->block);
         break;
     case NEU_CONN_TTY_CLIENT:
         conn->param.params.tty_client.device =
@@ -624,9 +651,9 @@ static void conn_connect(neu_conn_t *conn)
             conn->is_connected = false;
             return;
         } else {
-            zlog_notice(conn->param.log, "connect %s:%d success",
+            zlog_notice(conn->param.log, "connect %s:%d, block: %d success",
                         conn->param.params.tcp_client.ip,
-                        conn->param.params.tcp_client.port);
+                        conn->param.params.tcp_client.port, conn->block);
             conn->is_connected = true;
             conn->fd           = fd;
         }
@@ -846,6 +873,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++) {
@@ -865,14 +901,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,8 +968,7 @@ 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;
             }
         }
@@ -972,8 +999,7 @@ 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;
             }
         }
@@ -1011,7 +1037,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 +1072,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/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/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.9