GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-27 04:19:50 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1824 from hxy7yx/main-1

Browse Source 
test(ut):add async queue ut
This commit is contained in:
fengzeroz
2024-01-25 10:37:17 +08:00
committed by GitHub
parent a677f9331c e27ee0bd49
commit 0bf39bbe3e
9 changed files with 133 additions and 412 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CMakeLists.txt
-1
View File
@@ -85,7 +85,6 @@ set(NEURON_BASE_SOURCES
src/utils/json.c
src/utils/http.c
src/utils/http_handler.c
src/utils/http_proxy.c
src/utils/neu_jwt.c
src/utils/base64.c
src/utils/async_queue.c
create_cov_report.sh
+6 -7
View File
@@ -12,14 +12,13 @@ function gen_trace () {
case $1 in
(ut)
lcov -c -d build/tests/ut/CMakeFiles/tag_sort_test.dir -o cov_report/cov-tag_sort_test.info
lcov -c -d build/tests/ut/CMakeFiles/modbus_test.dir -o cov_report/cov-modbus_test.info
lcov -c -d build/tests/ut/CMakeFiles/jwt_test.dir -o cov_report/cov-jwt_test.info
lcov -c -d build/tests/ut/CMakeFiles/json_test.dir -o cov_report/cov-json_test.info
lcov -c -d build/tests/ut/CMakeFiles/http_test.dir -o cov_report/cov-http_test.info
lcov -c -d build/tests/ut/CMakeFiles/base64_test.dir -o cov_report/cov-base64_test.info
rm -f cov_report/cov-neuron.info cov_report/cov-restful.info
for dir in build/tests/ut/CMakeFiles/*_test.dir; do
test_name=$(basename ${dir} "_test.dir")
lcov -c -d ${dir} -o "cov_report/cov-${test_name}_test.info"
done
cd cov_report
lcov -a cov-tag_sort_test.info -a cov-modbus_test.info -a cov-jwt_test.info -a cov-json_test.info -a cov-http_test.info -a cov-base64_test.info -o cov-ut.info;;
lcov $(for file in cov-*.info; do echo "-a $file "; done) -a cov-neuron-base.info -o cov-ut.info;;
(ft)
lcov -c -d build/plugins/mqtt/CMakeFiles/plugin-mqtt.dir -o cov_report/cov-mqtt.info
lcov -c -d build/plugins/modbus/CMakeFiles/plugin-modbus-rtu.dir -o cov_report/cov-modbus-rtu.info
include/neuron/utils/http.h
-2
View File
@@ -66,8 +66,6 @@ ssize_t neu_http_get_param_str(nng_aio *aio, const char *name, char *buf,
// NEU_ERR_EINVAL if the query parameter value is not valid.
int neu_http_get_param_uintmax(nng_aio *aio, const char *name,
uintmax_t *param);
int neu_http_get_param_uint32(nng_aio *aio, const char *name, uint32_t *param);
int neu_http_get_param_uint64(nng_aio *aio, const char *name, uint64_t *param);
int neu_http_get_param_node_type(nng_aio *aio, const char *name,
neu_node_type_e *param);
include/neuron/utils/http_handler.h
-4
View File
@@ -89,7 +89,6 @@ enum neu_http_method {
enum neu_http_handler_type {
NEU_HTTP_HANDLER_FUNCTION = 0x0,
NEU_HTTP_HANDLER_DIRECTORY,
NEU_HTTP_HANDLER_PROXY,
NEU_HTTP_HANDLER_REDIRECT,
};
@@ -108,7 +107,4 @@ int neu_http_add_handler(nng_http_server * server,
const struct neu_http_handler *http_handler);
void neu_http_handle_cors(nng_aio *aio);
int neu_http_proxy_handler(const struct neu_http_handler *http_handler,
nng_http_handler ** handler);
#endif
src/utils/http.c
-71
View File
@@ -219,40 +219,6 @@ int neu_http_get_param_uintmax(nng_aio *aio, const char *name, uintmax_t *param)
return 0;
}
int neu_http_get_param_uint64(nng_aio *aio, const char *name, uint64_t *param)
{
uintmax_t val;
int rv = neu_http_get_param_uintmax(aio, name, &val);
if (0 != rv) {
return rv;
}
if (val > UINT64_MAX) {
return -1;
}
*param = (uint64_t) val;
return 0;
}
int neu_http_get_param_uint32(nng_aio *aio, const char *name, uint32_t *param)
{
uintmax_t val;
int rv = neu_http_get_param_uintmax(aio, name, &val);
if (0 != rv) {
return rv;
}
if (val > UINT32_MAX) {
return -1;
}
*param = (uint32_t) val;
return 0;
}
int neu_http_get_param_node_type(nng_aio *aio, const char *name,
neu_node_type_e *param)
{
@@ -408,43 +374,6 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content)
return response(aio, content, status);
}
int neu_http_response_file(nng_aio *aio, void *data, size_t len,
const char *disposition)
{
nng_http_res *res = NULL;
int rv = 0;
if (((rv = nng_http_res_alloc(&res)) != 0) ||
((rv = nng_http_res_set_status(res, NNG_HTTP_STATUS_OK)) != 0) ||
((rv = nng_http_res_set_header(res, "Content-Type",
"application/octet-stream")) != 0) ||
((rv = nng_http_res_set_header(res, "Content-Disposition",
disposition)) != 0) ||
((rv = nng_http_res_set_header(res, "Access-Control-Allow-Origin",
"*")) != 0) ||
((rv = nng_http_res_set_header(res, "Access-Control-Allow-Methods",
"POST,GET,PUT,DELETE,OPTIONS")) != 0) ||
((rv = nng_http_res_set_header(res, "Access-Control-Allow-Headers",
"*")) != 0) ||
((rv = nng_http_res_set_header(res, "Access-Control-Expose-Headers",
"Content-Disposition")) != 0) ||
((rv = nng_http_res_copy_data(res, data, len)) != 0)) {
nng_http_res_free(res);
nng_aio_finish(aio, rv);
return -1;
}
nng_http_req *nng_req = nng_aio_get_input(aio, 0);
nlog_notice("%s %s [%d]", nng_http_req_get_method(nng_req),
nng_http_req_get_uri(nng_req), NNG_HTTP_STATUS_OK);
nng_aio_set_output(aio, 0, res);
nng_aio_finish(aio, 0);
return 0;
}
int neu_http_ok(nng_aio *aio, char *content)
{
return response(aio, content, NNG_HTTP_STATUS_OK);
src/utils/http_handler.c
-3
View File
@@ -43,9 +43,6 @@ int neu_http_add_handler(nng_http_server * server,
ret = nng_http_handler_alloc_directory(&handler, http_handler->url,
http_handler->value.path);
break;
case NEU_HTTP_HANDLER_PROXY:
ret = neu_http_proxy_handler(http_handler, &handler);
break;
case NEU_HTTP_HANDLER_REDIRECT:
ret = nng_http_handler_alloc_redirect(&handler, http_handler->url, 0,
http_handler->value.path);
src/utils/http_proxy.c
-322
View File
@@ -1,322 +0,0 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2022 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.
**/
#include <pthread.h>
#include "errcodes.h"
#include "utils/http.h"
#include "utils/http_handler.h"
#include "utils/log.h"
#include "utils/utlist.h"
#include "json/neu_json_fn.h"
struct http_proxy_ctx;
static void handle_proxy(nng_aio *aio);
typedef struct {
pthread_mutex_t mtx;
size_t src_uri_len; // source uri length
size_t dst_uri_len; // destination uri length
size_t uri_buf_len; // request uri buffer length
char * uri_buf; // request uri buffer
nng_http_client * client; // client to destination
struct http_proxy_ctx *free_list; // context free list
} http_proxy_t;
struct http_proxy_ctx {
http_proxy_t * proxy;
nng_aio * aio;
nng_aio * user_aio; // aio to write back response
nng_http_res * res; // http response from the destination
struct http_proxy_ctx *prev;
struct http_proxy_ctx *next;
};
static void http_proxy_ctx_free(struct http_proxy_ctx *ctx)
{
if (NULL == ctx) {
return;
}
nng_aio_free(ctx->aio);
}
static int http_proxy_alloc(http_proxy_t **proxy, const char *src_url,
const char *dst_url)
{
int ret = 0;
nng_url *url = NULL;
http_proxy_t *p = calloc(1, sizeof(http_proxy_t));
if (NULL == p) {
nlog_error("alloc http proxy fail");
return -1;
}
if (0 != pthread_mutex_init(&p->mtx, NULL)) {
nlog_error("init http proxy mutex fail");
ret = -1;
goto err_mutex_init;
}
ret = nng_url_parse(&url, dst_url);
if (0 != ret) {
nlog_error("parse url: `%s` fail", dst_url);
goto err_url_parse;
}
p->dst_uri_len = strlen(url->u_requri);
p->uri_buf_len = p->dst_uri_len + 1;
p->uri_buf = calloc(p->uri_buf_len, sizeof(char));
if (NULL == p->uri_buf) {
nlog_error("alloc uri buffer for `%s` fail", dst_url);
ret = -1;
goto err_uri_buf;
}
// copy destination uri to head of buffer
strcpy(p->uri_buf, url->u_requri);
ret = nng_http_client_alloc(&p->client, url);
if (0 != ret) {
nlog_error("alloc http client for `%s` fail", dst_url);
goto err_http_client;
}
p->src_uri_len = strlen(src_url);
*proxy = p;
nng_url_free(url);
return ret;
err_http_client:
free(p->uri_buf);
err_uri_buf:
nng_url_free(url);
err_url_parse:
pthread_mutex_destroy(&p->mtx);
err_mutex_init:
free(p);
return ret;
}
static void http_proxy_free(http_proxy_t *proxy)
{
if (NULL == proxy) {
return;
}
if (proxy->client) {
nng_http_client_free(proxy->client);
}
free(proxy->uri_buf);
struct http_proxy_ctx *el = NULL, *tmp = NULL;
DL_FOREACH_SAFE(proxy->free_list, el, tmp)
{
DL_DELETE(proxy->free_list, el);
nng_aio_free(el->aio);
nng_aio_free(el->user_aio);
free(el);
}
pthread_mutex_destroy(&proxy->mtx);
free(proxy);
}
static inline int http_proxy_set_uri(http_proxy_t *proxy, nng_http_req *req)
{
const char *uri = nng_http_req_get_uri(req);
size_t len = strlen(uri);
size_t buf_len = len - proxy->src_uri_len + proxy->dst_uri_len + 1;
// TODO: use per context buffer if a single lock cause too much contention
pthread_mutex_lock(&proxy->mtx);
if (buf_len > proxy->uri_buf_len) {
proxy->uri_buf_len = buf_len;
proxy->uri_buf = realloc(proxy->uri_buf, buf_len);
if (NULL == proxy->uri_buf) {
nlog_error("realloc uri buffer fail");
pthread_mutex_unlock(&proxy->mtx);
return -1;
}
}
// buffer large enough guaranteed
sprintf(proxy->uri_buf + proxy->dst_uri_len, "%s",
uri + proxy->src_uri_len);
if (0 != nng_http_req_set_uri(req, proxy->uri_buf)) {
nlog_error("proxy fail set req uri: `%s`", proxy->uri_buf);
pthread_mutex_unlock(&proxy->mtx);
return -1;
}
pthread_mutex_unlock(&proxy->mtx);
return 0;
}
static void http_proxy_cb(void *arg)
{
int rv = 0;
struct http_proxy_ctx *ctx = arg;
nng_aio_set_output(ctx->user_aio, 0, ctx->res);
if (0 != (rv = nng_aio_result(ctx->aio))) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
if (0 !=
(rv = nng_http_res_set_status(
ctx->res, NNG_HTTP_STATUS_SERVICE_UNAVAILABLE)) ||
0 !=
(rv = nng_http_res_set_header(ctx->res, "Content-Type",
"application/json")) ||
0 !=
(rv = nng_http_res_copy_data(ctx->res, result_error,
strlen(result_error)))) {
nng_aio_set_output(ctx->user_aio, 0, NULL);
nng_http_res_free(ctx->res);
}
});
}
nng_aio_finish(ctx->user_aio, rv);
ctx->user_aio = NULL;
ctx->res = NULL;
pthread_mutex_lock(&ctx->proxy->mtx);
DL_APPEND(ctx->proxy->free_list, ctx); // put to free list
pthread_mutex_unlock(&ctx->proxy->mtx);
}
static struct http_proxy_ctx *http_proxy_get_ctx(http_proxy_t *proxy)
{
pthread_mutex_lock(&proxy->mtx);
struct http_proxy_ctx *ctx = proxy->free_list;
if (NULL != ctx) {
DL_DELETE(proxy->free_list, ctx);
pthread_mutex_unlock(&proxy->mtx);
return ctx;
}
pthread_mutex_unlock(&proxy->mtx);
ctx = calloc(1, sizeof(*ctx));
if (NULL == ctx) {
nlog_error("alloc proxy ctx fail");
return NULL;
}
if (0 != nng_aio_alloc(&ctx->aio, http_proxy_cb, ctx)) {
nlog_error("alloc proxy ctx aio fail");
free(ctx);
return NULL;
}
ctx->proxy = proxy;
return ctx;
}
int neu_http_proxy_handler(const struct neu_http_handler *http_handler,
nng_http_handler ** handler)
{
int ret = 0;
const char * dst = http_handler->value.dst_url;
nng_url * url = NULL;
http_proxy_t *proxy = NULL;
ret = http_proxy_alloc(&proxy, http_handler->url,
http_handler->value.dst_url);
if (0 != ret) {
return ret;
}
ret = nng_http_handler_alloc(handler, http_handler->url, handle_proxy);
if (0 != ret) {
nlog_error("alloc proxy handler for `%s` fail", dst);
goto error;
}
ret = nng_http_handler_set_tree(*handler);
if (0 != ret) {
nlog_error("proxy handler for `%s` set tree fail", dst);
goto error;
}
ret = nng_http_handler_set_method(*handler, NULL);
if (0 != ret) {
nlog_error("proxy handler for `%s` set all method fail", dst);
goto error;
}
ret = nng_http_handler_set_data(*handler, proxy,
(void (*)(void *)) http_proxy_free);
if (0 != ret) {
nlog_error("proxy handler for `%s` set data fail", dst);
goto error;
}
return ret;
error:
if (NULL != url) {
nng_url_free(url);
}
if (NULL != *handler) {
nng_http_handler_free(*handler);
}
if (NULL != proxy) {
http_proxy_free(proxy);
}
return ret;
}
static void handle_proxy(nng_aio *aio)
{
nng_http_res * res = NULL;
nng_http_req * req = nng_aio_get_input(aio, 0);
nng_http_handler *handler = nng_aio_get_input(aio, 1);
http_proxy_t * proxy = nng_http_handler_get_data(handler);
// fix the uri
if (0 != http_proxy_set_uri(proxy, req)) {
goto error;
}
struct http_proxy_ctx *ctx = http_proxy_get_ctx(proxy);
if (NULL == ctx) {
nlog_error("get proxy ctx fail");
goto error;
}
if (0 != nng_http_res_alloc(&res)) {
nlog_error("alloc proxy resp fail");
http_proxy_ctx_free(ctx);
goto error;
}
ctx->user_aio = aio;
ctx->res = res;
nng_http_client_transact(proxy->client, req, ctx->res, ctx->aio);
return;
error:
NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
neu_http_response(aio, error_code.error, result_error);
});
}
tests/ut/CMakeLists.txt
+9 -2
View File
@@ -42,7 +42,6 @@ target_include_directories(tag_sort_test PRIVATE
${CMAKE_SOURCE_DIR}/include
)
target_link_libraries(tag_sort_test neuron-base gtest_main gtest)
#target_link_directories(modbus_point_test PRIVATE /usr/local/lib)
add_executable(modbus_test modbus_test.cc
${CMAKE_SOURCE_DIR}/plugins/modbus/modbus.c
@@ -51,10 +50,18 @@ target_include_directories(modbus_test PRIVATE
${CMAKE_SOURCE_DIR}/plugins/modbus)
target_link_libraries(modbus_test neuron-base gtest_main gtest pthread zlog)
add_executable(async_queue_test async_queue_test.cc
${CMAKE_SOURCE_DIR}/src/utils/async_queue.c)
target_include_directories(async_queue_test PRIVATE
${CMAKE_SOURCE_DIR}/src/utils
${CMAKE_SOURCE_DIR}/include)
target_link_libraries(async_queue_test neuron-base gtest_main gtest)
include(GoogleTest)
gtest_discover_tests(json_test)
gtest_discover_tests(http_test)
gtest_discover_tests(jwt_test)
gtest_discover_tests(base64_test)
gtest_discover_tests(tag_sort_test)
gtest_discover_tests(modbus_test)
gtest_discover_tests(modbus_test)
gtest_discover_tests(async_queue_test)
tests/ut/async_queue_test.cc
+118
View File
@@ -0,0 +1,118 @@
#include <gtest/gtest.h>
#include <vector>
extern "C" {
#include "utils/async_queue.h"
#include "utils/log.h"
}
zlog_category_t *neuron = NULL;
uint64_t mock_key_fn(void *data)
{
return *reinterpret_cast<uint64_t *>(data);
}
bool mock_expire_fn(void *data)
{
return false;
}
void mock_free_fn(void *data)
{
delete reinterpret_cast<uint64_t *>(data);
}
bool mock_filter(void *filter_data, void *data)
{
return *reinterpret_cast<uint64_t *>(filter_data) ==
*reinterpret_cast<uint64_t *>(data);
}
void *create_test_data(uint64_t value)
{
uint64_t *data = new uint64_t(value);
return static_cast<void *>(data);
}
TEST(neu_async_queue_test, create_destroy)
{
auto q = neu_async_queue_new(mock_key_fn, mock_expire_fn, mock_free_fn, 10);
ASSERT_TRUE(q != nullptr);
neu_async_queue_destroy(q);
}
TEST(neu_async_queue_test, push)
{
const int max_size = 3;
auto q = neu_async_queue_new(mock_key_fn, mock_expire_fn, mock_free_fn,
max_size);
for (int i = 0; i < max_size; ++i) {
neu_async_queue_push(q, create_test_data(i));
}
neu_async_queue_push(q, create_test_data(max_size));
neu_async_queue_destroy(q);
}
TEST(neu_async_queue_test, pop)
{
auto q = neu_async_queue_new(mock_key_fn, mock_expire_fn, mock_free_fn, 10);
auto data = create_test_data(42);
neu_async_queue_push(q, data);
void *popped_data = nullptr;
auto result = neu_async_queue_pop(q, mock_key_fn(data), &popped_data);
ASSERT_EQ(result, 0);
ASSERT_EQ(popped_data, data);
mock_free_fn(data);
neu_async_queue_destroy(q);
}
TEST(neu_async_queue_test, remove)
{
auto q = neu_async_queue_new(mock_key_fn, mock_expire_fn, mock_free_fn, 10);
for (int i = 0; i < 5; ++i) {
neu_async_queue_push(q, create_test_data(i));
}
uint64_t remove_key = 2;
neu_async_queue_remove(q, mock_filter, &remove_key);
void *popped_data = nullptr;
auto pop_result = neu_async_queue_pop(q, remove_key, &popped_data);
ASSERT_EQ(pop_result, -1);
neu_async_queue_destroy(q);
}
TEST(neu_async_queue_test, clean)
{
auto q = neu_async_queue_new(mock_key_fn, mock_expire_fn, mock_free_fn, 10);
for (int i = 0; i < 5; ++i) {
neu_async_queue_push(q, create_test_data(i));
}
neu_async_queue_clean(q);
void *popped_data = nullptr;
auto pop_result = neu_async_queue_pop(q, 0, &popped_data);
ASSERT_EQ(pop_result, -1);
neu_async_queue_destroy(q);
}
int main(int argc, char **argv)
{
zlog_init("./config/dev.conf");
neuron = zlog_get_category("neuron");
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}