diff --git a/include/neuron/neuron.h b/include/neuron/neuron.h index d41cc7fd..2da5e3ac 100644 --- a/include/neuron/neuron.h +++ b/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 diff --git a/include/neuron/utils/protocol_buf.h b/include/neuron/utils/protocol_buf.h index 610ff846..3fd8e5de 100644 --- a/include/neuron/utils/protocol_buf.h +++ b/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; diff --git a/include/neuron/utils/utextend.h b/include/neuron/utils/utextend.h new file mode 100644 index 00000000..b50f36ac --- /dev/null +++ b/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 + +#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 \ No newline at end of file diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index ae7c9d06..1eda14d9 100644 --- a/src/adapter/driver/driver.c +++ b/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; diff --git a/src/base/tag_sort.c b/src/base/tag_sort.c index 8072f961..38ed9648 100644 --- a/src/base/tag_sort.c +++ b/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); } } \ No newline at end of file diff --git a/src/connection/connection.c b/src/connection/connection.c index 8d8e2c4e..e626de94 100644 --- a/src/connection/connection.c +++ b/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 };