diff --git a/CMakeLists.txt b/CMakeLists.txt index 08d04d6a..2b9ec372 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -162,7 +162,8 @@ set(NEURON_BASE_SOURCES src/utils/json.c src/utils/neu_jwt.c src/utils/file.c - src/utils/base64.c) + src/utils/base64.c + src/utils/async_queue.c) if(USE_NNG_MQTT) message("-- USE_NNG_MQTT is defined") @@ -189,7 +190,7 @@ if(USE_NNG_MQTT) target_link_libraries(neuron-base mbedtls mbedx509 mbedcrypto nng) endif() -target_link_libraries(neuron-base libzlog.so vector-static csptr jansson crypto ssl mqttc jwt +target_link_libraries(neuron-base nng libzlog.so vector-static csptr jansson crypto ssl mqttc jwt ${CMAKE_THREAD_LIBS_INIT}) #copy file for run diff --git a/include/neuron/type.h b/include/neuron/type.h index d5a4e304..536da061 100644 --- a/include/neuron/type.h +++ b/include/neuron/type.h @@ -348,10 +348,14 @@ static inline uint64_t neu_htonll(uint64_t u) uint64_t ret = 0; uint8_t *ret_bytes = (uint8_t *) &ret; - ret_bytes[0] = bytes[3]; - ret_bytes[1] = bytes[2]; - ret_bytes[2] = bytes[1]; - ret_bytes[3] = bytes[0]; + ret_bytes[0] = bytes[7]; + ret_bytes[1] = bytes[6]; + ret_bytes[2] = bytes[5]; + ret_bytes[3] = bytes[4]; + ret_bytes[4] = bytes[3]; + ret_bytes[5] = bytes[2]; + ret_bytes[6] = bytes[1]; + ret_bytes[7] = bytes[0]; return ret; } diff --git a/include/neuron/utils/async_queue.h b/include/neuron/utils/async_queue.h new file mode 100644 index 00000000..442bdb9e --- /dev/null +++ b/include/neuron/utils/async_queue.h @@ -0,0 +1,45 @@ +/** + * 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_ASYNC_QUEUE_H_ +#define _NEU_ASYNC_QUEUE_H_ + +#include +#include + +typedef struct neu_async_queue neu_async_queue_t; + +typedef uint64_t (*neu_async_queue_key)(void *elem); +typedef bool (*neu_async_queue_expire)(void *elem); +typedef void (*neu_async_queue_free)(void *elem); +typedef bool (*neu_async_queue_filter)(void *filter_elem, void *elem); + +neu_async_queue_t *neu_async_queue_new(neu_async_queue_key key_fn, + neu_async_queue_expire expire_fn, + neu_async_queue_free free_fn, + uint16_t max_size); + +void neu_async_queue_destroy(neu_async_queue_t *q); +void neu_async_queue_push(neu_async_queue_t *q, void *elem); +int neu_async_queue_pop(neu_async_queue_t *q, uint64_t key, void **elem); + +void neu_async_queue_remove(neu_async_queue_t *q, neu_async_queue_filter filter, + void *filter_elem); + +#endif \ No newline at end of file diff --git a/include/neuron/utils/utextend.h b/include/neuron/utils/utextend.h index b50f36ac..7373d66e 100644 --- a/include/neuron/utils/utextend.h +++ b/include/neuron/utils/utextend.h @@ -39,8 +39,19 @@ static inline UT_array *utarray_clone(UT_array *array) return result; } -#define utarray_foreach(array, type, elem) \ - for (type elem = utarray_front(array); elem != NULL; \ - elem = utarray_next(array, elem)) +#define utarray_foreach(array, type, elem) \ + for (type elem = (type) utarray_front(array); elem != NULL; \ + elem = (type) utarray_next(array, elem)) + +#define DL_LAST(head) ((head) != NULL) ? (head)->prev : NULL; +#define DL_PREV(el) ((el) != NULL) ? (el)->prev : NULL; + +#define DL_REVERSE_FOREACH(head, tail, el) \ + (tail) = DL_LAST(head); \ + (el) = (tail); \ + do + +#define DL_REVERSE_FOREACH_END(head, tail, el) \ + while (((el) = (el)->prev) != (head)) #endif \ No newline at end of file diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 1eda14d9..fd2d797e 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -118,6 +118,7 @@ static void update(neu_adapter_t *adapter, const char *name, neu_dvalue_t value) case NEU_TYPE_BYTES: break; default: + assert(value.type == NEU_TYPE_INT8); break; } diff --git a/src/utils/async_queue.c b/src/utils/async_queue.c new file mode 100644 index 00000000..89e79212 --- /dev/null +++ b/src/utils/async_queue.c @@ -0,0 +1,163 @@ +/** + * 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_ **/ + +#include + +#include +#include + +#include "utils/async_queue.h" +#include "utils/utextend.h" + +typedef struct element { + uint32_t key; + + void *data; + + struct element *prev; + struct element *next; +} element, element_list; + +struct neu_async_queue { + neu_async_queue_key key_fn; + neu_async_queue_expire expire_fn; + neu_async_queue_free free_fn; + + nng_mtx *mtx; + + uint16_t max; + + element_list *list; +}; + +neu_async_queue_t *neu_async_queue_new(neu_async_queue_key key_fn, + neu_async_queue_expire expire_fn, + neu_async_queue_free free_fn, + uint16_t max_size) +{ + neu_async_queue_t *q = calloc(1, sizeof(neu_async_queue_t)); + + q->key_fn = key_fn; + q->expire_fn = expire_fn; + q->free_fn = free_fn; + + q->max = max_size; + + nng_mtx_alloc(&q->mtx); + + return q; +} + +void neu_async_queue_destroy(neu_async_queue_t *q) +{ + element *elt = NULL, *tmp = NULL; + + nng_mtx_lock(q->mtx); + DL_FOREACH_SAFE(q->list, elt, tmp) + { + q->free_fn(elt->data); + DL_DELETE(q->list, elt); + free(elt); + } + nng_mtx_unlock(q->mtx); + + nng_mtx_free(q->mtx); + free(q); +} + +void neu_async_queue_push(neu_async_queue_t *q, void *elem) +{ + element *elt = NULL; + int count = 0; + + nng_mtx_lock(q->mtx); + DL_COUNT(q->list, elt, count); + + if (count == q->max && count > 0) { + elt = q->list; + DL_DELETE(q->list, elt); + q->free_fn(elt->data); + free(elt); + } + + elt = calloc(1, sizeof(element)); + elt->data = elem; + elt->key = q->key_fn(elem); + + DL_APPEND(q->list, elt); + + nng_mtx_unlock(q->mtx); +} + +int neu_async_queue_pop(neu_async_queue_t *q, uint64_t key, void **elem) +{ + element *elt = NULL, *first = NULL; + int ret = -1; + + nng_mtx_lock(q->mtx); + first = q->list; + elt = DL_LAST(q->list); + + while (elt != NULL) { + if (elt->key == key) { + DL_DELETE(q->list, elt); + *elem = elt->data; + free(elt); + ret = 0; + break; + } + + if (first == elt) { + break; + } + + if (q->expire_fn(elt->data)) { + element *tmp = DL_PREV(elt); + + DL_DELETE(q->list, elt); + q->free_fn(elt->data); + free(elt); + + elt = tmp; + } else { + elt = DL_PREV(elt); + } + } + + nng_mtx_unlock(q->mtx); + + return ret; +} + +void neu_async_queue_remove(neu_async_queue_t *q, neu_async_queue_filter filter, + void *filter_elem) +{ + element *el = NULL, *tmp = NULL; + + nng_mtx_lock(q->mtx); + DL_FOREACH_SAFE(q->list, el, tmp) + { + if (filter(filter_elem, el->data)) { + DL_DELETE(q->list, el); + q->free_fn(el->data); + free(el); + } + } + nng_mtx_unlock(q->mtx); +} \ No newline at end of file