Merge pull request #920 from fengzeroz/v2.2

version: 2.2.10

include/neuron/define.h

@@ -20,7 +20,7 @@
 #ifndef _NEU_DEFINE_H_
 #define _NEU_DEFINE_H_
 
-#define NEU_TAG_NAME_LEN 32
+#define NEU_TAG_NAME_LEN 48
 #define NEU_TAG_ADDRESS_LEN 128
 #define NEU_TAG_DESCRIPTION_LEN 128
 #define NEU_GROUP_NAME_LEN 32

plugins/ekuiper/json_rw.c

@@ -155,9 +155,10 @@ int json_encode_read_resp_tags(void *json_object, void *param)
         }
 
         neu_json_elem_t tag_elem = {
-            .name = json_tag.name,
-            .t    = json_tag.t,
-            .v    = json_tag.value,
+            .name      = json_tag.name,
+            .t         = json_tag.t,
+            .v         = json_tag.value,
+            .precision = trans_data->tags[i].value.precision,
         };
 
         ret = neu_json_encode_field((0 != json_tag.error) ? errors_object

plugins/ekuiper/plugin_ekuiper.c

@@ -114,6 +114,8 @@ static int ekuiper_plugin_start(neu_plugin_t *plugin)
 
     nng_pipe_notify(plugin->sock, NNG_PIPE_EV_ADD_POST, pipe_add_cb, plugin);
     nng_pipe_notify(plugin->sock, NNG_PIPE_EV_REM_POST, pipe_rm_cb, plugin);
+    nng_socket_set_int(plugin->sock, NNG_OPT_SENDBUF, 2048);
+    nng_socket_set_int(plugin->sock, NNG_OPT_RECVBUF, 2048);
 
     if ((rv = nng_listen(plugin->sock, EKUIPER_PLUGIN_URL, NULL, 0)) != 0) {
         plog_error(plugin, "nng_listen: %s", nng_strerror(rv));

plugins/ekuiper/read_write.c

@@ -56,7 +56,7 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data)
     rv = nng_sendmsg(plugin->sock, msg,
                      NNG_FLAG_NONBLOCK); // TODO: use aio to send message
     if (0 != rv) {
-        plog_error(plugin, "nng cannot send msg");
+        plog_error(plugin, "nng cannot send msg: %s", nng_strerror(rv));
         nng_msg_free(msg);
     }
 }

plugins/restful/adapter_handle.c

@@ -37,7 +37,7 @@ void handle_add_adapter(nng_aio *aio)
 
     REST_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
         aio, neu_json_add_node_req_t, neu_json_decode_add_node_req, {
-            if (strlen(req->name) > NEU_NODE_NAME_LEN) {
+            if (strlen(req->name) >= NEU_NODE_NAME_LEN) {
                 NEU_JSON_RESPONSE_ERROR(NEU_ERR_NODE_NAME_TOO_LONG, {
                     http_response(aio, NEU_ERR_NODE_NAME_TOO_LONG,
                                   result_error);

plugins/restful/group_config_handle.c

@@ -36,7 +36,7 @@ void handle_add_group_config(nng_aio *aio)
     REST_PROCESS_HTTP_REQUEST_VALIDATE_JWT(
         aio, neu_json_add_group_config_req_t,
         neu_json_decode_add_group_config_req, {
-            if (strlen(req->group) > NEU_GROUP_NAME_LEN) {
+            if (strlen(req->group) >= NEU_GROUP_NAME_LEN) {
                 NEU_JSON_RESPONSE_ERROR(NEU_ERR_GROUP_NAME_TOO_LONG, {
                     http_response(aio, NEU_ERR_GROUP_NAME_TOO_LONG,
                                   result_error);

plugins/restful/license.c

@@ -36,6 +36,7 @@
 #include <time.h>
 
 #include "errcodes.h"
+#include "utils/asprintf.h"
 #include "utils/log.h"
 
 #include "license.h"
@@ -81,7 +82,7 @@ static uint64_t mktimestamp(struct tm dt)
     return (ret);
 }
 
-static uint64_t get_asn1_ts(ASN1_TIME *time)
+static uint64_t get_asn1_ts(ASN1_TIME *time, char **ts_str)
 {
     struct tm   t;
     const char *str = (const char *) time->data;
@@ -97,7 +98,11 @@ static uint64_t get_asn1_ts(ASN1_TIME *time)
                &t.tm_hour, &t.tm_min, &t.tm_sec);
     }
 
+    neu_asprintf(ts_str, "%4d-%02d-%02d %02d:%02d:%02d", t.tm_year, t.tm_mon,
+                 t.tm_mday, t.tm_hour, t.tm_min, t.tm_sec);
+
     t.tm_mon -= 1;
+
     /* Note: we did not adjust the time based on time zone information */
     return mktimestamp(t);
 }
@@ -245,9 +250,14 @@ static int extract_license(const char *license_fname, const char *pub_key,
 
     not_before      = X509_get_notBefore(cert);
     not_after       = X509_get_notAfter(cert);
-    license->since_ = get_asn1_ts(not_before);
-    license->until_ = get_asn1_ts(not_after);
+    license->since_ = get_asn1_ts(not_before, &license->since_str_);
+    license->until_ = get_asn1_ts(not_after, &license->until_str_);
     nlog_info("certificate expire date: %s", not_after->data);
+    if (NULL == license->since_str_ || NULL == license->until_str_) {
+        nlog_warn("cannot strdup");
+        rv = NEU_ERR_EINTERNAL;
+        goto final;
+    }
 
     subj = X509_get_subject_name(cert);
     for (i = 0; i < X509_NAME_entry_count(subj); i++) {
@@ -273,6 +283,7 @@ static int extract_license(const char *license_fname, const char *pub_key,
         goto final;
     }
 
+    unsigned mark = 0;
     for (i = 0; i < sk_X509_EXTENSION_num(ext_list); i++) {
         ext = sk_X509_EXTENSION_value(ext_list, i);
         obj = X509_EXTENSION_get_object(ext);
@@ -286,25 +297,36 @@ static int extract_license(const char *license_fname, const char *pub_key,
         if (0 == strcmp(buf, LICENSE_KEY_TYPE)) {
             license->type_ = scan_licese_type(val);
             if (LICENSE_TYPE_MAX == license->type_) {
-                zlog_error(neuron, "invalid license type:`%s`", val);
+                nlog_error("invalid license type:`%s`", val);
                 rv = NEU_ERR_LICENSE_INVALID;
             }
+            mark |= 0x01;
         } else if (0 == strcmp(buf, LICENSE_KEY_MAX_NODES)) {
             license->max_nodes_ = strtoul(val, NULL, 0);
+            mark |= 0x02;
         } else if (0 == strcmp(buf, LICENSE_KEY_MAX_NODE_TAGS)) {
             license->max_node_tags_ = strtoul(val, NULL, 0);
+            mark |= 0x04;
         } else if (0 == strcmp(buf, LICENSE_KEY_PLUGIN_FLAG)) {
             if (0 !=
                 scan_bitvec(license->plugin_flag_,
                             sizeof(license->plugin_flag_), val)) {
-                zlog_error(neuron, "fail to scan license plugin flag:`%s`",
-                           val);
+                nlog_error("fail to scan license plugin flag:`%s`", val);
                 rv = NEU_ERR_LICENSE_INVALID;
             }
+            mark |= 0x08;
+        } else if (0 == strcmp(buf, LICENSE_KEY_TOKEN)) {
+            license->token_ = strdup(val);
+            mark |= 0x10;
         }
     }
 
-    nlog_info("certification verification success");
+    if (0x0F == (mark & 0x0F)) {
+        nlog_info("certification verification success");
+    } else {
+        rv = NEU_ERR_LICENSE_INVALID;
+        nlog_error("missing license key fields, mark:%u", mark);
+    }
 
 final:
     if (pkey)
@@ -330,26 +352,15 @@ const char *plugin_bit_str(plugin_bit_e b)
 
 void license_print(license_t *license)
 {
-    struct tm   tm        = {};
-    char        start[16] = {};
-    char        end[16]   = {};
-    time_t      since     = license->since_;
-    time_t      until     = license->until_;
-    const char *fmt       = "%Y%m%dT%H%M%S";
-
-    gmtime_r(&since, &tm);
-    strftime(start, sizeof(start), fmt, &tm);
-    gmtime_r(&until, &tm);
-    strftime(end, sizeof(end), fmt, &tm);
-
     char flag[sizeof(license->plugin_flag_) * 2 + 1] = {};
     print_bitvec(flag, sizeof(flag), license->plugin_flag_,
                  sizeof(license->plugin_flag_));
 
     nlog_info("license `%s` for %s, from %s to %s, max_nodes:%" PRIu32
               ", max_node_tags:%" PRIu32 ", flag:0x%s",
-              license->fname_, license_type_str(license->type_), start, end,
-              license->max_nodes_, license->max_node_tags_, flag);
+              license->fname_, license_type_str(license->type_),
+              license->since_str_, license->until_str_, license->max_nodes_,
+              license->max_node_tags_, flag);
 }
 
 void license_init(license_t *license)
@@ -363,7 +374,10 @@ void license_init(license_t *license)
 
 void license_fini(license_t *license)
 {
+    free(license->since_str_);
+    free(license->until_str_);
     free(license->fname_);
+    free(license->token_);
 }
 
 license_t *license_new()
@@ -410,7 +424,10 @@ int license_read(license_t *license, const char *fname)
 
     license_print(&lic);
 
+    free(license->since_str_);
+    free(license->until_str_);
     free(license->fname_);
+    free(license->token_);
     *license = lic;
 
     return rv;

plugins/restful/license.h

@@ -50,6 +50,8 @@
 //
 #define LICENSE_KEY_PLUGIN_FLAG "1.3.6.1.4.1.52509.13"
 
+#define LICENSE_KEY_TOKEN "1.3.6.1.4.1.52509.14"
+
 #ifdef __cplusplus
 extern "C" {
 #endif
@@ -91,10 +93,13 @@ typedef struct {
     license_type_e type_;                 // license type
     char *         fname_;                // license file name
     uint64_t       since_;                // not before
+    char *         since_str_;            //
     uint64_t       until_;                // not after
+    char *         until_str_;            //
     uint32_t       max_nodes_;            // max number of nodes
     uint32_t       max_node_tags_;        // max number of tags per node
     uint8_t        plugin_flag_[200 / 8]; // plugin enable/disable bits
+    char *         token_;                // hardware token
 } license_t;
 
 // Initialize to be expired with no file.

plugins/restful/license_handle.c

@@ -135,17 +135,12 @@ int copy_license_file_if_necessary()
 
 void handle_get_license(nng_aio *aio)
 {
-    int                         rv              = 0;
-    char *                      license_path    = NULL;
-    license_t                   lic             = {};
-    neu_json_get_license_resp_t resp            = {};
-    time_t                      since           = 0;
-    time_t                      until           = 0;
-    char                        valid_since[20] = {};
-    char                        valid_until[20] = {};
-    struct tm                   tm              = {};
-    UT_array *                  plugin_names    = NULL;
-    char *                      result          = NULL;
+    int                         rv           = 0;
+    char *                      license_path = NULL;
+    license_t                   lic          = {};
+    neu_json_get_license_resp_t resp         = {};
+    UT_array *                  plugin_names = NULL;
+    char *                      result       = NULL;
 
     VALIDATE_JWT(aio);
 
@@ -163,26 +158,10 @@ void handle_get_license(nng_aio *aio)
         goto final;
     }
 
-    since = license_not_before(&lic);
-    gmtime_r(&since, &tm);
-    if (0 ==
-        strftime(valid_since, sizeof(valid_since), "%Y-%m-%d %H:%M:%S", &tm)) {
-        rv = NEU_ERR_EINTERNAL;
-        goto final;
-    }
-
-    until = license_not_after(&lic);
-    gmtime_r(&until, &tm);
-    if (0 ==
-        strftime(valid_until, sizeof(valid_until), "%Y-%m-%d %H:%M:%S", &tm)) {
-        rv = NEU_ERR_EINTERNAL;
-        goto final;
-    }
-
     resp.license_type  = (char *) license_type_str(license_get_type(&lic));
     resp.valid         = !license_is_expired(&lic);
-    resp.valid_since   = valid_since;
-    resp.valid_until   = valid_until;
+    resp.valid_since   = lic.since_str_;
+    resp.valid_until   = lic.until_str_;
     resp.max_nodes     = license_get_max_nodes(&lic);
     resp.max_node_tags = license_get_max_node_tags(&lic);
 

src/adapter/adapter.c

@@ -61,7 +61,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
     neu_event_io_param_t    param       = { 0 };
     neu_event_timer_param_t timer_param = {
         .second      = 0,
-        .millisecond = 100,
+        .millisecond = 10,
         .cb          = update_timestamp,
     };
 

src/adapter/driver/driver.c

@@ -502,15 +502,15 @@ int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group,
     int      ret  = NEU_ERR_SUCCESS;
     group_t *find = NULL;
 
-    if (strlen(tag->name) > NEU_TAG_NAME_LEN) {
+    if (strlen(tag->name) >= NEU_TAG_NAME_LEN) {
         return NEU_ERR_TAG_NAME_TOO_LONG;
     }
 
-    if (strlen(tag->address) > NEU_TAG_ADDRESS_LEN) {
+    if (strlen(tag->address) >= NEU_TAG_ADDRESS_LEN) {
         return NEU_ERR_TAG_ADDRESS_TOO_LONG;
     }
 
-    if (strlen(tag->description) > NEU_TAG_DESCRIPTION_LEN) {
+    if (strlen(tag->description) >= NEU_TAG_DESCRIPTION_LEN) {
         return NEU_ERR_TAG_DESCRIPTION_TOO_LONG;
     }
 
@@ -568,7 +568,7 @@ int neu_adapter_driver_update_tag(neu_adapter_driver_t *driver,
     int      ret  = NEU_ERR_SUCCESS;
     group_t *find = NULL;
 
-    if (strlen(tag->name) > NEU_TAG_NAME_LEN) {
+    if (strlen(tag->name) >= NEU_TAG_NAME_LEN) {
         return NEU_ERR_TAG_NAME_TOO_LONG;
     }
 

src/connection/connection.c

@@ -235,6 +235,11 @@ ssize_t neu_conn_tcp_server_send(neu_conn_t *conn, int fd, uint8_t *buf,
     conn_tcp_server_listen(conn);
     ret = send(fd, buf, len, MSG_NOSIGNAL | MSG_DONTWAIT);
 
+    if (ret <= 0 && errno != EAGAIN) {
+        conn->disconnected(conn->data, fd);
+        conn_tcp_server_del_client(conn, fd);
+    }
+
     pthread_mutex_unlock(&conn->mtx);
 
     return ret;
@@ -257,6 +262,11 @@ ssize_t neu_conn_tcp_server_recv(neu_conn_t *conn, int fd, uint8_t *buf,
         ret = recv(fd, buf, len, 0);
     }
 
+    if (ret <= 0 && errno != EAGAIN) {
+        conn->disconnected(conn->data, fd);
+        conn_tcp_server_del_client(conn, fd);
+    }
+
     pthread_mutex_unlock(&conn->mtx);
 
     return ret;
@@ -350,7 +360,7 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
                    conn->fd, len, ret, strerror(errno), errno);
     }
 
-    if (errno == EPIPE || ret <= 0) {
+    if (errno == EPIPE || ret == -1) {
         conn_disconnect(conn);
     }
 
@@ -509,6 +519,7 @@ static void conn_tcp_server_stop(neu_conn_t *conn)
             conn->fd = 0;
         }
 
+        conn->tcp_server.n_client  = 0;
         conn->tcp_server.is_listen = false;
     }
 }
@@ -758,6 +769,7 @@ static void conn_disconnect(neu_conn_t *conn)
                 conn->tcp_server.clients[i].fd = 0;
             }
         }
+        conn->tcp_server.n_client = 0;
         break;
     case NEU_CONN_TCP_CLIENT:
     case NEU_CONN_UDP:
@@ -804,8 +816,6 @@ static void conn_tcp_server_del_client(neu_conn_t *conn, int fd)
             return;
         }
     }
-
-    assert(1 == 0);
 }
 
 void neu_conn_stream_consume(neu_conn_t *conn, void *context,

src/event/event_linux.c

@@ -103,7 +103,10 @@ static void *event_loop(void *arg)
                 assert(size != -1);
 
                 nng_mtx_lock(data->ctx.timer->mtx);
+                epoll_ctl(epoll_fd, EPOLL_CTL_DEL, data->fd, NULL);
                 ret = data->callback.timer(data->usr_data);
+                timerfd_settime(data->fd, 0, &data->ctx.timer->value, NULL);
+                epoll_ctl(epoll_fd, EPOLL_CTL_ADD, data->fd, &event);
                 nng_mtx_unlock(data->ctx.timer->mtx);
             }
             break;

version

@@ -1 +1 @@
-2.2.6
+2.2.10