fix(driver): write string

include/neuron/neuron.h

@@ -33,6 +33,15 @@ extern "C" {
 #include "types.h"
 #include "utils/base64.h"
 
+#include "connection/neu_connection.h"
+#include "event/event.h"
+
+#include "json/neu_json_param.h"
+
+#include "utils/utextend.h"
+
+#include "tag_sort.h"
+
 #ifdef __cplusplus
 }
 #endif

include/neuron/utils/protocol_buf.h

@@ -54,10 +54,10 @@ inline static uint16_t neu_protocol_buf_size(neu_protocol_buf_t *buf)
 inline static neu_protocol_unpack_buf_t *
 neu_protocol_unpack_buf_new(uint16_t size)
 {
-    neu_protocol_unpack_buf_t *buf =
-        calloc(1, sizeof(neu_protocol_unpack_buf_t));
+    neu_protocol_unpack_buf_t *buf = (neu_protocol_unpack_buf_t *) calloc(
+        1, sizeof(neu_protocol_unpack_buf_t));
 
-    buf->base   = calloc(size, sizeof(uint8_t));
+    buf->base   = (uint8_t *) calloc(size, sizeof(uint8_t));
     buf->size   = size;
     buf->offset = 0;
 
@@ -66,9 +66,10 @@ neu_protocol_unpack_buf_new(uint16_t size)
 
 inline static neu_protocol_pack_buf_t *neu_protocol_pack_buf_new(uint16_t size)
 {
-    neu_protocol_pack_buf_t *buf = calloc(1, sizeof(neu_protocol_pack_buf_t));
+    neu_protocol_pack_buf_t *buf =
+        (neu_protocol_pack_buf_t *) calloc(1, sizeof(neu_protocol_pack_buf_t));
 
-    buf->base   = calloc(size, sizeof(uint8_t));
+    buf->base   = (uint8_t *) calloc(size, sizeof(uint8_t));
     buf->size   = size;
     buf->offset = size;
 
@@ -127,6 +128,16 @@ inline static uint8_t *neu_protocol_pack_buf(neu_protocol_pack_buf_t *buf,
     return buf->base + buf->offset;
 }
 
+inline static uint8_t *neu_protocol_pack_buf_set(neu_protocol_pack_buf_t *buf,
+                                                 uint16_t offset, uint16_t size)
+{
+    if (buf->offset + offset + size - 1 > buf->size) {
+        return NULL;
+    }
+
+    return buf->base + buf->offset + offset;
+}
+
 inline static uint8_t *neu_protocol_pack_buf_get(neu_protocol_pack_buf_t *buf)
 {
     return buf->base + buf->offset;

include/neuron/utils/utextend.h

@@ -0,0 +1,46 @@
+/**
+ * NEURON IIoT System for Industry 4.0
+ * Copyright (C) 2020-2021 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.
+config_ **/
+
+#ifndef _NEU_UT_EXTEND_H_
+#define _NEU_UT_EXTEND_H_
+
+#include <stdlib.h>
+
+#include "utarray.h"
+#include "uthash.h"
+#include "utlist.h"
+
+static inline UT_array *utarray_clone(UT_array *array)
+{
+    UT_array *result = (UT_array *) calloc(1, sizeof(UT_array));
+
+    result->i   = array->i;
+    result->n   = array->n;
+    result->icd = array->icd;
+    result->d   = (char *) calloc(1, array->n * array->icd.sz);
+    memcpy(result->d, array->d, array->n * array->icd.sz);
+
+    return result;
+}
+
+#define utarray_foreach(array, type, elem)               \
+    for (type elem = utarray_front(array); elem != NULL; \
+         elem      = utarray_next(array, elem))
+
+#endif

src/adapter/driver/driver.c

@@ -284,9 +284,13 @@ void neu_adapter_driver_process_msg(neu_adapter_driver_t *driver,
             case NEU_DTYPE_BOOL:
                 neu_dvalue_get_bool(int_val->val, &value.boolean);
                 break;
-            case NEU_DTYPE_CSTR:
-                neu_dvalue_get_ref_cstr(int_val->val, (char **) &value.str);
+            case NEU_DTYPE_CSTR: {
+                char *str = NULL;
+                neu_dvalue_get_cstr(int_val->val, &str);
+                strncpy(value.str, str, sizeof(value.str) - 1);
+                free(str);
                 break;
+            }
             default:
                 assert(1 == 0);
                 break;

src/base/tag_sort.c

@@ -90,5 +90,8 @@ static void tag_sort(neu_tag_sort_result_t *result, void *tag,
         utarray_new(result->sorts[result->n_sort - 1].tags, &icd);
         utarray_push_back(result->sorts[result->n_sort - 1].tags, &tag);
         result->sorts[result->n_sort - 1].info.size = 1;
+        fn(&result->sorts[result->n_sort - 1],
+           *(void **) utarray_back(result->sorts[result->n_sort - 1].tags),
+           tag);
     }
 }

src/connection/connection.c

@@ -228,6 +228,22 @@ ssize_t neu_conn_tcp_server_recv(neu_conn_t *conn, int fd, uint8_t *buf,
     return ret;
 }
 
+static ssize_t neu_conn_tcp_server_recv_msg(neu_conn_t *conn, int fd,
+                                            uint8_t *buf, ssize_t len)
+{
+    ssize_t ret = 0;
+    assert(conn->param.type == NEU_CONN_TCP_SERVER);
+
+    ret = recv(fd, buf, len, 0);
+    if (ret == -1) {
+        log_error("tcp server fd: %d, recv buf len %d, ret: %d, errno: %s(%d)",
+                  fd, len, ret, strerror(errno), errno);
+        return -1;
+    }
+
+    return ret;
+}
+
 ssize_t neu_conn_send(neu_conn_t *conn, uint8_t *buf, ssize_t len)
 {
     ssize_t ret = 0;
@@ -327,6 +343,38 @@ ssize_t neu_conn_recv(neu_conn_t *conn, uint8_t *buf, ssize_t len)
     return ret;
 }
 
+static ssize_t neu_conn_tcp_client_recv_msg(neu_conn_t *conn, uint8_t *buf,
+                                            ssize_t len)
+{
+    ssize_t ret = 0;
+
+    pthread_mutex_lock(&conn->mtx);
+
+    assert(conn->param.type == NEU_CONN_TCP_CLIENT);
+    ret = recv(conn->fd, buf, len, 0);
+    if (ret == -1) {
+        log_error("tcp client fd: %d, recv buf len %d, ret: %d, errno: %s(%d)",
+                  conn->fd, len, ret, strerror(errno), errno);
+    }
+
+    if (errno == EPIPE || ret <= 0) {
+        conn_disconnect(conn);
+        if (conn->callback_trigger == true) {
+            conn->disconnected(conn->data, conn->fd);
+            conn->callback_trigger = false;
+        }
+    }
+
+    if (ret > 0 && conn->callback_trigger == false) {
+        conn->connected(conn->data, conn->fd);
+        conn->callback_trigger = true;
+    }
+
+    pthread_mutex_unlock(&conn->mtx);
+
+    return ret;
+}
+
 void neu_conn_tcp_server_flush(neu_conn_t *conn, int fd)
 {
     (void) conn;
@@ -653,8 +701,8 @@ void neu_conn_stream_consume(neu_conn_t *conn, void *context,
 {
     static __thread uint8_t  recv_buf[2048] = { 0 };
     static __thread uint16_t offset         = 0;
-    ssize_t                  ret =
-        neu_conn_recv(conn, recv_buf + offset, sizeof(recv_buf) - offset);
+    ssize_t ret = neu_conn_tcp_client_recv_msg(conn, recv_buf + offset,
+                                               sizeof(recv_buf) - offset);
     if (ret > 0) {
         offset += ret;
         neu_protocol_unpack_buf_t protocol_buf = { 0 };
@@ -682,8 +730,8 @@ void neu_conn_stream_tcp_server_consume(neu_conn_t *conn, int fd, void *context,
 {
     static __thread uint8_t  recv_buf[2048] = { 0 };
     static __thread uint16_t offset         = 0;
-    ssize_t ret = neu_conn_tcp_server_recv(conn, fd, recv_buf + offset,
-                                           sizeof(recv_buf) - offset);
+    ssize_t ret = neu_conn_tcp_server_recv_msg(conn, fd, recv_buf + offset,
+                                               sizeof(recv_buf) - offset);
     if (ret > 0) {
         offset += ret;
         neu_protocol_unpack_buf_t protocol_buf = { 0 };