diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5d9b9a4f..bbf8c2d6 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -18,8 +18,6 @@ jobs: strategy: matrix: arch: - - [aarch64-linux-gnu, aarch64, arm64] - - [arm-linux-gnueabihf, armv4, armhf] - [x86_64-linux-gnu, x86_64, amd64] container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-main @@ -106,12 +104,6 @@ jobs: with: name: neuron-sdk-debug - - uses: actions/download-artifact@v4 - with: - name: neuron-sdk-arm64 - - uses: actions/download-artifact@v4 - with: - name: neuron-sdk-armhf - uses: actions/download-artifact@v4 with: name: neuron-sdk-amd64 diff --git a/.gitignore b/.gitignore index 4b621579..1883a5f8 100644 --- a/.gitignore +++ b/.gitignore @@ -19,4 +19,5 @@ neuron-sdk .vscode log.html output.xml -report.html \ No newline at end of file +report.html +*.cache \ No newline at end of file diff --git a/CMakeLists.txt b/CMakeLists.txt index 382ec98e..93111e72 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -162,6 +162,7 @@ set(NEURON_SOURCES plugins/restful/otel_handle.c plugins/restful/cid_handle.c plugins/restful/system_handle.c + plugins/restful/datalayers_handle.c plugins/restful/rest.c plugins/restful/user.c) @@ -187,6 +188,7 @@ add_subdirectory(plugins/mqtt) add_subdirectory(plugins/ekuiper) add_subdirectory(plugins/file) add_subdirectory(plugins/monitor) +add_subdirectory(plugins/datalayers) add_subdirectory(simulator) diff --git a/Install-dependencies.md b/Install-dependencies.md index 3a750bb3..4f4d73dd 100644 --- a/Install-dependencies.md +++ b/Install-dependencies.md @@ -82,3 +82,28 @@ $ ./autogen.sh $ ./configure --enable-shared=no CFLAGS=-fPIC CXXFLAGS=-fPIC $ make && sudo make install ``` + +[arrow](https://github.com/apache/arrow) +```shell +$ wget https://github.com/apache/arrow/releases/download/apache-arrow-19.0.1/apache-arrow-19.0.1.tar.gz +$ tar -xzvf apache-arrow-19.0.1.tar.gz +$ cd apache-arrow-19.0.1/cpp +$ mkdir build && cd build +$ cmake .. \ + -DCMAKE_BUILD_TYPE=Release \ + -DARROW_BUILD_SHARED=ON \ + -DARROW_COMPUTE=ON \ + -DARROW_CSV=ON \ + -DARROW_JSON=ON \ + -DARROW_PARQUET=ON \ + -DARROW_DATASET=ON \ + -DARROW_FLIGHT=ON \ + -DARROW_FLIGHT_SQL=ON \ + -DARROW_WITH_GRPC=ON \ + -DARROW_PROTOBUF_USE_SHARED=ON \ + -DProtobuf_ROOT=/usr \ + -DgRPC_ROOT=/usr \ + -GNinja +$ ninja +$ sudo ninja install +``` \ No newline at end of file diff --git a/cmake/x86_64-linux-gnu.cmake b/cmake/x86_64-linux-gnu.cmake index 98866b69..5c97384e 100644 --- a/cmake/x86_64-linux-gnu.cmake +++ b/cmake/x86_64-linux-gnu.cmake @@ -13,6 +13,11 @@ set(CMAKE_RANLIB ${COMPILER_PREFIX}-ranlib) set(CMAKE_STAGING_PREFIX ${LIBRARY_DIR}/${COMPILER_PREFIX}) set(CMAKE_PREFIX_PATH ${CMAKE_STAGING_PREFIX}) +set(CMAKE_INTERPROCEDURAL_OPTIMIZATION OFF) + +set(CMAKE_C_FLAGS "${CMAKE_C_FLAGS} -fno-lto") +set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fno-lto") + include_directories(SYSTEM ${CMAKE_STAGING_PREFIX}/include) include_directories(SYSTEM ${CMAKE_STAGING_PREFIX}/openssl/include) set(CMAKE_FIND_ROOT_PATH ${CMAKE_STAGING_PREFIX}) diff --git a/src/persist/datalayer/flight_sql_client.h b/include/neuron/datalayers/flight_sql_client.h similarity index 56% rename from src/persist/datalayer/flight_sql_client.h rename to include/neuron/datalayers/flight_sql_client.h index 8f8d1539..ce4b6c7b 100644 --- a/src/persist/datalayer/flight_sql_client.h +++ b/include/neuron/datalayers/flight_sql_client.h @@ -25,12 +25,12 @@ typedef struct { } datatag; typedef struct { + ValueType value_type; + ValueUnion value; char time[64]; char node_name[128]; char group_name[128]; char tag[128]; - ValueUnion value; - ValueType value_type; } datarow; typedef struct { @@ -38,29 +38,30 @@ typedef struct { datarow *rows; } query_result; -typedef struct Client Client; +typedef struct Client neu_datalayers_client; -Client *client_create(const char *host, int port, const char *username, - const char *password); +neu_datalayers_client *client_create(const char *host, int port, + const char *username, + const char *password); -int client_execute(Client *client, const char *sql); +int client_execute(neu_datalayers_client *client, const char *sql); -void client_destroy(Client *client); +void client_destroy(neu_datalayers_client *client); -int client_insert(Client *client, ValueType type, datatag *tags, +int client_insert(neu_datalayers_client *client, ValueType type, datatag *tags, size_t tag_count); -query_result *client_query(Client *client, ValueType type, +query_result *client_query(neu_datalayers_client *client, ValueType type, const char *node_name, const char *group_name, const char *tag); void client_query_free(query_result *result); -query_result *client_query_nodes_groups(Client *client, ValueType type); +query_result *client_query_nodes_groups(neu_datalayers_client *client, + ValueType type); -query_result *client_query_all_data(Client *client, ValueType type); - -query_result *client_query_by_tag(Client *client, const char *tag); +query_result *client_query_all_data(neu_datalayers_client *client, + ValueType type); #ifdef __cplusplus } diff --git a/include/neuron/define.h b/include/neuron/define.h index 22b85407..65778333 100644 --- a/include/neuron/define.h +++ b/include/neuron/define.h @@ -125,7 +125,7 @@ }); \ } while (0) -#define CHECK_NODE_MONITOR_ERR \ +#define CHECK_NODE_SINGLE_ERR \ do { \ NEU_JSON_RESPONSE_ERROR(NEU_ERR_NODE_NOT_ALLOW_UPDATE, { \ neu_http_response(aio, NEU_ERR_NODE_NOT_ALLOW_UPDATE, \ diff --git a/include/neuron/errcodes.h b/include/neuron/errcodes.h index 0bf3d3ab..67db6789 100644 --- a/include/neuron/errcodes.h +++ b/include/neuron/errcodes.h @@ -176,6 +176,11 @@ typedef enum { NEU_ERR_FILE_READ_FAILURE = 4102, NEU_ERR_FILE_WRITE_FAILURE = 4103, + NEU_ERR_DATALAYERS_FAILURE = 4200, + NEU_ERR_DATALAYERS_INIT_FAILURE = 4201, + NEU_ERR_DATALAYERS_CONNECT_FAILURE = 4202, + NEU_ERR_DATALAYERS_IS_NULL = 4203, + NEU_ERR_PLUGIN_ERROR_START = 10000, NEU_ERR_PLUGIN_ERROR_END = 20000, } neu_err_code_e; diff --git a/include/neuron/metrics.h b/include/neuron/metrics.h index 38bda0ef..7c312860 100644 --- a/include/neuron/metrics.h +++ b/include/neuron/metrics.h @@ -236,6 +236,18 @@ typedef enum { #define NEU_METRIC_DISCONNECTION_1800S_HELP \ "Number of disconnection within the last 1800 seconds" +// current cache queue size +#define NEU_METRIC_CACHE_QUEUE_SIZE "cache_queue_size" +#define NEU_METRIC_CACHE_QUEUE_SIZE_TYPE NEU_METRIC_TYPE_GAUGE +#define NEU_METRIC_CACHE_QUEUE_SIZE_HELP \ + "Current number of messages in the cache queue" + +// total number of dropped messages due to full cache +#define NEU_METRIC_CACHE_DROPPED_MSGS_TOTAL "cache_dropped_msgs_total" +#define NEU_METRIC_CACHE_DROPPED_MSGS_TOTAL_TYPE NEU_METRIC_TYPE_COUNTER +#define NEU_METRIC_CACHE_DROPPED_MSGS_TOTAL_HELP \ + "Total number of messages dropped due to cache queue being full" + typedef enum { NEU_METRICS_CATEGORY_GLOBAL, NEU_METRICS_CATEGORY_DRIVER, diff --git a/include/neuron/msg.h b/include/neuron/msg.h index ec25df29..48a62107 100644 --- a/include/neuron/msg.h +++ b/include/neuron/msg.h @@ -64,6 +64,11 @@ typedef enum neu_reqresp_type { NEU_REQ_GET_SUB_DRIVER_TAGS, NEU_RESP_GET_SUB_DRIVER_TAGS, + NEU_REQ_GET_DATALAYERS_GROUPS, + NEU_RESP_GET_DATALAYERS_GROUPS, + NEU_REQ_GET_DATALAYERS_TAGS, + NEU_RESP_GET_DATALAYERS_TAGS, + NEU_REQ_NODE_INIT, NEU_REQ_NODE_UNINIT, NEU_RESP_NODE_UNINIT, @@ -190,6 +195,11 @@ static const char *neu_reqresp_type_string_t[] = { [NEU_REQ_GET_SUB_DRIVER_TAGS] = "NEU_REQ_GET_SUB_DRIVER_TAGS", [NEU_RESP_GET_SUB_DRIVER_TAGS] = "NEU_RESP_GET_SUB_DRIVER_TAGS", + [NEU_REQ_GET_DATALAYERS_GROUPS] = "NEU_REQ_GET_DATALAYERS_GROUPS", + [NEU_RESP_GET_DATALAYERS_GROUPS] = "NEU_RESP_GET_DATALAYERS_GROUPS", + [NEU_REQ_GET_DATALAYERS_TAGS] = "NEU_REQ_GET_DATALAYERS_TAGS", + [NEU_RESP_GET_DATALAYERS_TAGS] = "NEU_RESP_GET_DATALAYERS_TAGS", + [NEU_REQ_NODE_INIT] = "NEU_REQ_NODE_INIT", [NEU_REQ_NODE_UNINIT] = "NEU_REQ_NODE_UNINIT", [NEU_RESP_NODE_UNINIT] = "NEU_RESP_NODE_UNINIT", diff --git a/plugins/datalayers/CMakeLists.txt b/plugins/datalayers/CMakeLists.txt new file mode 100644 index 00000000..79bcbbd4 --- /dev/null +++ b/plugins/datalayers/CMakeLists.txt @@ -0,0 +1,48 @@ +cmake_minimum_required(VERSION 3.12) + +project(plugin-datalayers) + +set(CMAKE_CXX_STANDARD 17) +set(CMAKE_CXX_STANDARD_REQUIRED ON) +set(CMAKE_CXX_EXTENSIONS OFF) + +set(LIBRARY_OUTPUT_PATH "${CMAKE_BINARY_DIR}/plugins") + +file(COPY ${CMAKE_SOURCE_DIR}/plugins/datalayers/datalayers.json DESTINATION ${CMAKE_BINARY_DIR}/plugins/schema/) + +add_library(${PROJECT_NAME} SHARED + datalayers_config.c + datalayers_handle.c + datalayers_plugin.c + datalayers_plugin_intf.c + ${CMAKE_SOURCE_DIR}/src/persist/datalayers/flight_sql_client.cpp +) + +target_include_directories(${PROJECT_NAME} PRIVATE + ${CMAKE_SOURCE_DIR}/include/neuron + ${CMAKE_SOURCE_DIR}/plugins/datalayers + ${CMAKE_SOURCE_DIR}/src/persist/datalayers +) + +set(ARROW_USE_STATIC_LIBS ON) +set(ARROW_STATIC_LIB TRUE) + +find_package(Arrow REQUIRED) +find_package(ArrowFlight REQUIRED) +find_package(ArrowFlightSql REQUIRED) +find_package(gRPC REQUIRED) + +target_link_libraries(${PROJECT_NAME} + -Wl,--whole-archive + gRPC::grpc++ + -Wl,--no-whole-archive + Arrow::arrow_static + ArrowFlight::arrow_flight_static + ArrowFlightSql::arrow_flight_sql_static + stdc++ + gflags + ${CMAKE_THREAD_LIBS_INIT} +) + +target_link_libraries(${PROJECT_NAME} neuron-base) +target_link_libraries(${PROJECT_NAME} ${CMAKE_THREAD_LIBS_INIT}) \ No newline at end of file diff --git a/plugins/datalayers/datalayers.json b/plugins/datalayers/datalayers.json new file mode 100644 index 00000000..4634e1e0 --- /dev/null +++ b/plugins/datalayers/datalayers.json @@ -0,0 +1,51 @@ +{ + "host": { + "name": "Broker Host", + "name_zh": "服务器地址", + "description": "Datalayers broker host", + "description_zh": "Datalayers 服务器地址", + "attribute": "required", + "type": "string", + "default": "127.0.0.1", + "valid": { + "length": 255 + } + }, + "port": { + "name": "Broker Port", + "name_zh": "服务器端口", + "description": "Datalayers broker port", + "description_zh": "Datalayers 服务器端口", + "attribute": "required", + "type": "int", + "default": 8360, + "valid": { + "min": 1, + "max": 65535 + } + }, + "username": { + "name": "Username", + "name_zh": "用户名", + "description": "Datalayers user name", + "description_zh": "Datalayers 用户名", + "attribute": "required", + "type": "string", + "default": "admin", + "valid": { + "length": 255 + } + }, + "password": { + "name": "Password", + "name_zh": "密码", + "description": "Datalayers user password", + "description_zh": "Datalayers 用户密码", + "attribute": "required", + "type": "string", + "default": "public", + "valid": { + "length": 255 + } + } +} \ No newline at end of file diff --git a/plugins/datalayers/datalayers_config.c b/plugins/datalayers/datalayers_config.c new file mode 100644 index 00000000..8f7b3a9c --- /dev/null +++ b/plugins/datalayers/datalayers_config.c @@ -0,0 +1,106 @@ +/** + * 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 "json/json.h" +#include "json/neu_json_param.h" + +#include "datalayers_config.h" +#include "datalayers_plugin.h" + +int datalayers_config_parse(neu_plugin_t *plugin, const char *setting, + datalayers_config_t *config) +{ + int ret = 0; + char * err_param = NULL; + const char *placeholder = "********"; + + neu_json_elem_t host = { .name = "host", .t = NEU_JSON_STR }; + neu_json_elem_t port = { .name = "port", .t = NEU_JSON_INT }; + neu_json_elem_t username = { .name = "username", .t = NEU_JSON_STR }; + neu_json_elem_t password = { .name = "password", .t = NEU_JSON_STR }; + + if (NULL == setting || NULL == config) { + plog_error(plugin, "invalid argument, null pointer"); + return -1; + } + + ret = neu_parse_param(setting, &err_param, 2, &host, &port); + if (0 != ret) { + plog_error(plugin, "parsing setting fail, key: `%s`", err_param); + goto error; + } + + // host, required + if (0 == strlen(host.v.val_str)) { + plog_error(plugin, "setting invalid host: `%s`", host.v.val_str); + goto error; + } + + // port, required + if (0 == port.v.val_int || port.v.val_int > 65535) { + plog_error(plugin, "setting invalid port: %" PRIi64, port.v.val_int); + goto error; + } + + ret = neu_parse_param(setting, NULL, 1, &username); + if (0 != ret) { + plog_notice(plugin, "setting no username"); + goto error; + } + + ret = neu_parse_param(setting, NULL, 1, &password); + if (0 != ret) { + plog_notice(plugin, "setting no password"); + goto error; + } + + config->host = host.v.val_str; + config->port = port.v.val_int; + config->username = username.v.val_str; + config->password = password.v.val_str; + + plog_notice(plugin, "config host : %s", config->host); + plog_notice(plugin, "config port : %" PRIu16, config->port); + + if (config->username) { + plog_notice(plugin, "config username : %s", config->username); + } + if (config->password) { + plog_notice(plugin, "config password : %s", + 0 == strlen(config->password) ? "" : placeholder); + } + + return 0; + +error: + free(err_param); + free(host.v.val_str); + free(username.v.val_str); + free(password.v.val_str); + return -1; +} + +void datalayers_config_fini(datalayers_config_t *config) +{ + free(config->host); + free(config->username); + free(config->password); + + memset(config, 0, sizeof(*config)); +} \ No newline at end of file diff --git a/plugins/datalayers/datalayers_config.h b/plugins/datalayers/datalayers_config.h new file mode 100644 index 00000000..a5815299 --- /dev/null +++ b/plugins/datalayers/datalayers_config.h @@ -0,0 +1,46 @@ +/** + * 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. + **/ + +#ifndef NEURON_PLUGIN_DATALAYERS_CONFIG_H +#define NEURON_PLUGIN_DATALAYERS_CONFIG_H + +#ifdef __cplusplus +extern "C" { +#endif + +#include + +#include "plugin.h" + +typedef struct { + char * host; + uint16_t port; + char * username; + char * password; +} datalayers_config_t; + +int datalayers_config_parse(neu_plugin_t *plugin, const char *setting, + datalayers_config_t *config); +void datalayers_config_fini(datalayers_config_t *config); + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/plugins/datalayers/datalayers_handle.c b/plugins/datalayers/datalayers_handle.c new file mode 100644 index 00000000..b89ee172 --- /dev/null +++ b/plugins/datalayers/datalayers_handle.c @@ -0,0 +1,397 @@ +/** + * 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 "errcodes.h" +#include "plugin.h" +#include "type.h" + +#include "datalayers_handle.h" +#include "datalayers_plugin.h" +#include "datalayers_plugin_intf.h" + +void task_queue_init(task_queue_t *queue) +{ + queue->head = NULL; + queue->tail = NULL; + queue->size = 0; + queue->max_size = 0; +} + +static inline db_write_task_t *task_new() +{ + db_write_task_t *task = malloc(sizeof(db_write_task_t)); + if (!task) { + return NULL; + } + memset(task, 0, sizeof(db_write_task_t)); + + return task; +} + +static inline void task_free(db_write_task_t *task) +{ + if (task) { + utarray_free(task->int_tags); + utarray_free(task->float_tags); + utarray_free(task->bool_tags); + utarray_free(task->string_tags); + free(task); + } +} + +void tasks_free(task_queue_t *queue) +{ + db_write_task_t *task = queue->head; + while (task) { + db_write_task_t *next = task->next; + task_free(task); + task = next; + } + queue->head = NULL; + queue->tail = NULL; + queue->size = 0; +} + +void task_queue_push(neu_plugin_t *plugin, db_write_task_t *task) +{ + if (plugin->task_queue.size >= MAX_QUEUE_SIZE) { + db_write_task_t *old_task = plugin->task_queue.head; + plugin->task_queue.head = plugin->task_queue.head->next; + task_free(old_task); + plugin->task_queue.size--; + NEU_PLUGIN_UPDATE_DISCARDED_MSGS_METRIC(plugin, 1); + } + + if (plugin->task_queue.tail) { + plugin->task_queue.tail->next = task; + } + plugin->task_queue.tail = task; + if (plugin->task_queue.head == NULL) { + plugin->task_queue.head = task; + } + plugin->task_queue.size++; + + if (plugin->task_queue.size > plugin->task_queue.max_size) { + plugin->task_queue.max_size = plugin->task_queue.size; + NEU_PLUGIN_UPDATE_MAX_CACHED_QUEUE_SIZE_METRIC( + plugin, plugin->task_queue.max_size); + } + NEU_PLUGIN_UPDATE_CACHED_QUEUE_SIZE_METRIC(plugin, plugin->task_queue.size); +} + +db_write_task_t *task_queue_pop(neu_plugin_t *plugin, task_queue_t *queue) +{ + if (queue->size == 0) { + return NULL; + } + + db_write_task_t *task = queue->head; + queue->head = queue->head->next; + if (queue->head == NULL) { + queue->tail = NULL; + } + queue->size--; + + NEU_PLUGIN_UPDATE_CACHED_QUEUE_SIZE_METRIC(plugin, plugin->task_queue.size); + + return task; +} + +static void db_write_task_cb(db_write_task_t *task, neu_plugin_t *plugin) +{ + int ret = 0; + if (utarray_len(task->int_tags) > 0) { + ret = client_insert(plugin->client, INT_TYPE, + utarray_eltptr(task->int_tags, 0), + utarray_len(task->int_tags)); + } + + if (utarray_len(task->float_tags) > 0) { + ret = client_insert(plugin->client, FLOAT_TYPE, + utarray_eltptr(task->float_tags, 0), + utarray_len(task->float_tags)); + } + + if (utarray_len(task->bool_tags) > 0) { + ret = client_insert(plugin->client, BOOL_TYPE, + utarray_eltptr(task->bool_tags, 0), + utarray_len(task->bool_tags)); + } + + if (utarray_len(task->string_tags) > 0) { + ret = client_insert(plugin->client, STRING_TYPE, + utarray_eltptr(task->string_tags, 0), + utarray_len(task->string_tags)); + } + + if (ret != 0) { + plog_error(plugin, "insert data to datalayers failed, disconnected"); + if (plugin->client) { + client_destroy(plugin->client); + plugin->client = NULL; + } + } +} + +void db_write_task_consumer(neu_plugin_t *plugin) +{ + db_write_task_t *task = NULL; + + while (1) { + task = task_queue_pop(plugin, &plugin->task_queue); + if (task) { + db_write_task_cb(task, plugin); + + task_free(task); + } else { + usleep(1000); + } + } +} + +static void tag_array_copy(void *dst, const void *src) +{ + const datatag *src_tag = (const datatag *) src; + datatag * dst_tag = (datatag *) dst; + + dst_tag->node_name = src_tag->node_name ? strdup(src_tag->node_name) : NULL; + dst_tag->group_name = + src_tag->group_name ? strdup(src_tag->group_name) : NULL; + dst_tag->tag = src_tag->tag ? strdup(src_tag->tag) : NULL; + + dst_tag->value_type = src_tag->value_type; + + switch (src_tag->value_type) { + case STRING_TYPE: + dst_tag->value.string_value = src_tag->value.string_value + ? strdup(src_tag->value.string_value) + : NULL; + break; + case INT_TYPE: + dst_tag->value.int_value = src_tag->value.int_value; + break; + case FLOAT_TYPE: + dst_tag->value.float_value = src_tag->value.float_value; + break; + case BOOL_TYPE: + dst_tag->value.bool_value = src_tag->value.bool_value; + break; + default: + break; + } +} + +static void tag_array_free(void *_elt) +{ + datatag *elt = (datatag *) _elt; + + free((void *) elt->node_name); + free((void *) elt->group_name); + free((void *) elt->tag); + + if (elt->value_type == STRING_TYPE) { + free((void *) elt->value.string_value); + } +} + +static void tag_array_init(void *_elt) +{ + datatag *elt = (datatag *) _elt; + memset(elt, 0, sizeof(datatag)); +} + +static UT_icd ut_datatag_icd = { sizeof(datatag), tag_array_init, + tag_array_copy, tag_array_free }; + +int handle_trans_data(neu_plugin_t * plugin, + neu_reqresp_trans_data_t *trans_data) +{ + int rv = 0; + + if (NULL == plugin->client) { + return NEU_ERR_DATALAYERS_IS_NULL; + } + + if (plugin->common.link_state != NEU_NODE_LINK_STATE_CONNECTED) { + return NEU_ERR_PLUGIN_NOT_RUNNING; + } + + const route_entry_t *route = route_tbl_get( + &plugin->route_tbl, trans_data->driver, trans_data->group); + if (NULL == route) { + plog_error(plugin, "no route for driver:%s group:%s", + trans_data->driver, trans_data->group); + return NEU_ERR_GROUP_NOT_SUBSCRIBE; + } + + UT_array *int_tags = NULL, *float_tags = NULL, *bool_tags = NULL, + *string_tags = NULL; + utarray_new(int_tags, &ut_datatag_icd); + utarray_new(float_tags, &ut_datatag_icd); + utarray_new(bool_tags, &ut_datatag_icd); + utarray_new(string_tags, &ut_datatag_icd); + + utarray_foreach(trans_data->tags, neu_resp_tag_value_meta_t *, tag_meta) + { + if (tag_meta->value.type == NEU_TYPE_ERROR) { + continue; + } + + switch (tag_meta->value.type) { + case NEU_TYPE_BIT: + case NEU_TYPE_INT8: + case NEU_TYPE_UINT8: + case NEU_TYPE_INT16: + case NEU_TYPE_UINT16: + case NEU_TYPE_INT32: + case NEU_TYPE_UINT32: + case NEU_TYPE_INT64: + case NEU_TYPE_UINT64: { + datatag tag = { trans_data->driver, + trans_data->group, + tag_meta->tag, + { .int_value = tag_meta->value.value.i64 }, + INT_TYPE }; + utarray_push_back(int_tags, &tag); + } break; + case NEU_TYPE_FLOAT: + case NEU_TYPE_DOUBLE: { + datatag tag = { trans_data->driver, + trans_data->group, + tag_meta->tag, + { .float_value = tag_meta->value.value.d64 }, + FLOAT_TYPE }; + utarray_push_back(float_tags, &tag); + } break; + case NEU_TYPE_BOOL: { + datatag tag = { trans_data->driver, + trans_data->group, + tag_meta->tag, + { .bool_value = tag_meta->value.value.boolean }, + BOOL_TYPE }; + utarray_push_back(bool_tags, &tag); + } break; + case NEU_TYPE_STRING: + case NEU_TYPE_DATA_AND_TIME: + case NEU_TYPE_TIME: { + datatag tag = { trans_data->driver, + trans_data->group, + tag_meta->tag, + { .string_value = tag_meta->value.value.str }, + STRING_TYPE }; + utarray_push_back(string_tags, &tag); + } break; + case NEU_TYPE_BYTES: { + char formatted[NEU_VALUE_SIZE]; + int offset = 0; + + offset += + snprintf(formatted + offset, sizeof(formatted) - offset, "[ "); + + for (int i = 0; i < tag_meta->value.value.bytes.length; ++i) { + offset += snprintf( + formatted + offset, sizeof(formatted) - offset, "%u%s", + tag_meta->value.value.bytes.bytes[i], + (i == tag_meta->value.value.bytes.length - 1) ? " " : ", "); + } + + snprintf(formatted + offset, sizeof(formatted) - offset, "]"); + datatag tag = { trans_data->driver, + trans_data->group, + tag_meta->tag, + { .string_value = formatted }, + STRING_TYPE }; + utarray_push_back(string_tags, &tag); + } break; + default: + break; + } + } + + db_write_task_t *task = task_new(); + task->int_tags = int_tags; + task->float_tags = float_tags; + task->bool_tags = bool_tags; + task->string_tags = string_tags; + + task_queue_push(plugin, task); + + return rv; +} + +int handle_subscribe_group(neu_plugin_t *plugin, neu_req_subscribe_t *sub_info) +{ + int rv = 0; + + rv = route_tbl_add_new(&plugin->route_tbl, sub_info->driver, + sub_info->group); + if (0 != rv) { + plog_error(plugin, "route driver:%s group:%s fail, `%s`", + sub_info->driver, sub_info->group, sub_info->params); + goto end; + } + + plog_notice(plugin, "route driver:%s group:%s", sub_info->driver, + sub_info->group); + +end: + free(sub_info->params); + return rv; +} + +int handle_unsubscribe_group(neu_plugin_t * plugin, + neu_req_unsubscribe_t *unsub_info) +{ + route_tbl_del(&plugin->route_tbl, unsub_info->driver, unsub_info->group); + plog_notice(plugin, "del route driver:%s group:%s", unsub_info->driver, + unsub_info->group); + return 0; +} + +int handle_del_group(neu_plugin_t *plugin, neu_req_del_group_t *req) +{ + route_tbl_del(&plugin->route_tbl, req->driver, req->group); + plog_notice(plugin, "del route driver:%s group:%s", req->driver, + req->group); + return 0; +} + +int handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req) +{ + route_tbl_update_group(&plugin->route_tbl, req->driver, req->group, + req->new_name); + plog_notice(plugin, "update route driver:%s group:%s to %s", req->driver, + req->group, req->new_name); + return 0; +} + +int handle_update_driver(neu_plugin_t *plugin, neu_req_update_node_t *req) +{ + route_tbl_update_driver(&plugin->route_tbl, req->node, req->new_name); + plog_notice(plugin, "update route driver:%s to %s", req->node, + req->new_name); + return 0; +} + +int handle_del_driver(neu_plugin_t *plugin, neu_reqresp_node_deleted_t *req) +{ + route_tbl_del_driver(&plugin->route_tbl, req->node); + plog_notice(plugin, "delete route driver:%s", req->node); + return 0; +} \ No newline at end of file diff --git a/plugins/datalayers/datalayers_handle.h b/plugins/datalayers/datalayers_handle.h new file mode 100644 index 00000000..93f5c2d0 --- /dev/null +++ b/plugins/datalayers/datalayers_handle.h @@ -0,0 +1,49 @@ +/** + * 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. + **/ + +#ifndef NEURON_PLUGIN_MQTT_HANDLE_H +#define NEURON_PLUGIN_MQTT_HANDLE_H + +#ifdef __cplusplus +extern "C" { +#endif + +#include "datalayers_plugin.h" + +void db_write_task_consumer(neu_plugin_t *plugin); +void task_queue_init(task_queue_t *queue); +void tasks_free(task_queue_t *queue); + +int handle_trans_data(neu_plugin_t * plugin, + neu_reqresp_trans_data_t *trans_data); + +int handle_subscribe_group(neu_plugin_t *plugin, neu_req_subscribe_t *sub_info); +int handle_unsubscribe_group(neu_plugin_t * plugin, + neu_req_unsubscribe_t *unsub_info); + +int handle_update_group(neu_plugin_t *plugin, neu_req_update_group_t *req); +int handle_del_group(neu_plugin_t *plugin, neu_req_del_group_t *cmd); + +int handle_update_driver(neu_plugin_t *plugin, neu_req_update_node_t *req); +int handle_del_driver(neu_plugin_t *plugin, neu_reqresp_node_deleted_t *req); +#ifdef __cplusplus +} +#endif + +#endif diff --git a/plugins/datalayers/datalayers_plugin.c b/plugins/datalayers/datalayers_plugin.c new file mode 100644 index 00000000..d18741a9 --- /dev/null +++ b/plugins/datalayers/datalayers_plugin.c @@ -0,0 +1,37 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2024 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 "datalayers_plugin_intf.h" + +#define DESCRIPTION "Northbound Datalayers plugin." +#define DESCRIPTION_ZH "北向应用 Datalayers 插件" + +const neu_plugin_module_t neu_plugin_module = { + .version = NEURON_PLUGIN_VER_1_0, + .schema = "datalayers", + .module_name = "Datalayers", + .module_descr = DESCRIPTION, + .module_descr_zh = DESCRIPTION_ZH, + .intf_funs = &datalayers_plugin_intf_funs, + .kind = NEU_PLUGIN_KIND_SYSTEM, + .type = NEU_NA_TYPE_APP, + .display = true, + .single = true, + .single_name = "datalayers", +}; \ No newline at end of file diff --git a/plugins/datalayers/datalayers_plugin.h b/plugins/datalayers/datalayers_plugin.h new file mode 100644 index 00000000..c20294ad --- /dev/null +++ b/plugins/datalayers/datalayers_plugin.h @@ -0,0 +1,192 @@ +/** + * 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. + **/ + +#ifndef NEURON_PLUGIN_DATALAYERS_H +#define NEURON_PLUGIN_DATALAYERS_H + +#include +#ifdef __cplusplus +extern "C" { +#endif + +#include "neuron.h" + +#include "datalayers/flight_sql_client.h" +#include "datalayers_config.h" + +typedef struct { + char driver[NEU_NODE_NAME_LEN]; + char group[NEU_GROUP_NAME_LEN]; +} route_key_t; + +typedef struct { + route_key_t key; + + UT_hash_handle hh; +} route_entry_t; + +#define MAX_QUEUE_SIZE 1000 + +typedef struct db_write_task_s { + UT_array * int_tags; + UT_array * float_tags; + UT_array * bool_tags; + UT_array * string_tags; + struct db_write_task_s *next; +} db_write_task_t; + +typedef struct { + db_write_task_t *head; + db_write_task_t *tail; + int size; + int max_size; +} task_queue_t; + +struct neu_plugin { + neu_plugin_common_t common; + datalayers_config_t config; + route_entry_t * route_tbl; + neu_datalayers_client *client; + + task_queue_t task_queue; + pthread_t consumer_thread; + + int (*parse_config)(neu_plugin_t *plugin, const char *setting, + datalayers_config_t *config); +}; + +static inline void route_entry_free(route_entry_t *e) +{ + free(e); +} + +static inline void route_tbl_free(route_entry_t *tbl) +{ + route_entry_t *e = NULL, *tmp = NULL; + HASH_ITER(hh, tbl, e, tmp) + { + HASH_DEL(tbl, e); + route_entry_free(e); + } +} + +static inline route_entry_t * +route_tbl_get(route_entry_t **tbl, const char *driver, const char *group) +{ + route_entry_t *find = NULL; + route_key_t key = { 0 }; + + strncpy(key.driver, driver, sizeof(key.driver)); + strncpy(key.group, group, sizeof(key.group)); + + HASH_FIND(hh, *tbl, &key, sizeof(key), find); + return find; +} + +// NOTE: we take ownership of `topic` +static inline int route_tbl_add_new(route_entry_t **tbl, const char *driver, + const char *group) +{ + route_entry_t *find = NULL; + + find = route_tbl_get(tbl, driver, group); + if (find) { + return NEU_ERR_GROUP_ALREADY_SUBSCRIBED; + } + + find = calloc(1, sizeof(*find)); + if (NULL == find) { + return NEU_ERR_EINTERNAL; + } + + strncpy(find->key.driver, driver, sizeof(find->key.driver)); + strncpy(find->key.group, group, sizeof(find->key.group)); + HASH_ADD(hh, *tbl, key, sizeof(find->key), find); + + return 0; +} + +static inline int route_tbl_update(route_entry_t **tbl, const char *driver, + const char *group) +{ + route_entry_t *find = NULL; + + find = route_tbl_get(tbl, driver, group); + if (NULL == find) { + return NEU_ERR_GROUP_NOT_SUBSCRIBE; + } + + return 0; +} + +static inline void route_tbl_update_driver(route_entry_t **tbl, + const char * driver, + const char * new_name) +{ + route_entry_t *e = NULL, *tmp = NULL; + HASH_ITER(hh, *tbl, e, tmp) + { + if (0 == strcmp(e->key.driver, driver)) { + HASH_DEL(*tbl, e); + strncpy(e->key.driver, new_name, sizeof(e->key.driver)); + HASH_ADD(hh, *tbl, key, sizeof(e->key), e); + } + } +} + +static inline void route_tbl_update_group(route_entry_t **tbl, + const char *driver, const char *group, + const char *new_name) +{ + route_entry_t *e = route_tbl_get(tbl, driver, group); + if (e) { + HASH_DEL(*tbl, e); + strncpy(e->key.group, new_name, sizeof(e->key.group)); + HASH_ADD(hh, *tbl, key, sizeof(e->key), e); + } +} + +static inline void route_tbl_del_driver(route_entry_t **tbl, const char *driver) +{ + route_entry_t *e = NULL, *tmp = NULL; + HASH_ITER(hh, *tbl, e, tmp) + { + if (0 == strcmp(e->key.driver, driver)) { + HASH_DEL(*tbl, e); + route_entry_free(e); + } + } +} + +static inline void route_tbl_del(route_entry_t **tbl, const char *driver, + const char *group) +{ + route_entry_t *find = NULL; + find = route_tbl_get(tbl, driver, group); + if (find) { + HASH_DEL(*tbl, find); + route_entry_free(find); + } +} + +#ifdef __cplusplus +} +#endif + +#endif \ No newline at end of file diff --git a/plugins/datalayers/datalayers_plugin_intf.c b/plugins/datalayers/datalayers_plugin_intf.c new file mode 100644 index 00000000..ba94ba54 --- /dev/null +++ b/plugins/datalayers/datalayers_plugin_intf.c @@ -0,0 +1,218 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2024 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 "define.h" +#include "errcodes.h" +#include "flight_sql_client.h" + +#include "plugin.h" + +#include "datalayers_config.h" +#include "datalayers_handle.h" +#include "datalayers_plugin.h" +#include "datalayers_plugin_intf.h" +#include + +extern const neu_plugin_module_t neu_plugin_module; + +neu_plugin_t *datalayers_plugin_open(void) +{ + neu_plugin_t *plugin = (neu_plugin_t *) calloc(1, sizeof(neu_plugin_t)); + neu_plugin_common_init(&plugin->common); + plugin->parse_config = datalayers_config_parse; + return plugin; +} + +void stop_consumer_thread(pthread_t *consumer_thread) +{ + pthread_cancel(*consumer_thread); + pthread_join(*consumer_thread, NULL); +} + +int datalayers_plugin_close(neu_plugin_t *plugin) +{ + const char *name = neu_plugin_module.module_name; + plog_notice(plugin, "success to free plugin:%s", name); + + stop_consumer_thread(&plugin->consumer_thread); + + tasks_free(&plugin->task_queue); + + client_destroy(plugin->client); + free(plugin); + return NEU_ERR_SUCCESS; +} + +int datalayers_plugin_init(neu_plugin_t *plugin, bool load) +{ + (void) load; + + task_queue_init(&plugin->task_queue); + + pthread_create(&plugin->consumer_thread, NULL, + (void *) db_write_task_consumer, plugin); + pthread_detach(plugin->consumer_thread); + + NEU_PLUGIN_REGISTER_CACHED_QUEUE_SIZE_METRIC(plugin); + NEU_PLUGIN_REGISTER_MAX_CACHED_QUEUE_SIZE_METRIC(plugin); + NEU_PLUGIN_REGISTER_DISCARDED_MSGS_METRIC(plugin); + + plog_notice(plugin, "initialize plugin `%s` success", + neu_plugin_module.module_name); + return NEU_ERR_SUCCESS; +} + +int datalayers_plugin_uninit(neu_plugin_t *plugin) +{ + tasks_free(&plugin->task_queue); + datalayers_config_fini(&plugin->config); + + route_tbl_free(plugin->route_tbl); + + plog_notice(plugin, "uninitialize plugin `%s` success", + neu_plugin_module.module_name); + return NEU_ERR_SUCCESS; +} + +static int config_datalayers_client(neu_plugin_t * plugin, + const datalayers_config_t *config) +{ + plugin->client = client_create(config->host, config->port, config->username, + config->password); + + if (NULL == plugin->client) { + plog_error(plugin, "datalayers client_create failed"); + plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED; + return NEU_ERR_DATALAYERS_CONNECT_FAILURE; + } + + plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED; + return NEU_ERR_SUCCESS; +} + +int datalayers_plugin_config(neu_plugin_t *plugin, const char *setting) +{ + int rv = 0; + const char * plugin_name = neu_plugin_module.module_name; + datalayers_config_t config = { 0 }; + + if (plugin->client != NULL) { + client_destroy(plugin->client); + } + + rv = plugin->parse_config(plugin, setting, &config); + if (0 != rv) { + plog_error(plugin, "neu_datalayers_config_parse fail"); + return NEU_ERR_NODE_SETTING_INVALID; + } + + rv = config_datalayers_client(plugin, &config); + if (0 != rv) { + plog_error(plugin, "datalayers client configuration failed"); + datalayers_config_fini(&config); + return rv; + } + + datalayers_config_fini(&plugin->config); + + memmove(&plugin->config, &config, sizeof(config)); + + plog_notice(plugin, "config plugin `%s` success", plugin_name); + return NEU_ERR_SUCCESS; +} + +int datalayers_plugin_start(neu_plugin_t *plugin) +{ + const char *plugin_name = neu_plugin_module.module_name; + + if (NULL == plugin->client) { + plog_error(plugin, "datalayers client_create failed"); + return NEU_ERR_DATALAYERS_CONNECT_FAILURE; + } + + plog_notice(plugin, "start plugin `%s` success", plugin_name); + plugin->common.link_state = NEU_NODE_LINK_STATE_CONNECTED; + return NEU_ERR_SUCCESS; +} + +int datalayers_plugin_stop(neu_plugin_t *plugin) +{ + + plog_notice(plugin, "stop plugin `%s` success", + neu_plugin_module.module_name); + plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED; + return NEU_ERR_SUCCESS; +} + +int datalayers_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, + void *data) +{ + neu_err_code_e error = NEU_ERR_SUCCESS; + + switch (head->type) { + case NEU_REQRESP_TRANS_DATA: { + if (plugin->client == NULL) { + plog_notice(plugin, "datalayers client is NULL, reconnect"); + + config_datalayers_client(plugin, &plugin->config); + } + + error = handle_trans_data(plugin, data); + break; + } + case NEU_REQ_SUBSCRIBE_GROUP: + error = handle_subscribe_group(plugin, data); + break; + case NEU_REQ_UNSUBSCRIBE_GROUP: + error = handle_unsubscribe_group(plugin, data); + break; + case NEU_REQ_UPDATE_GROUP: + error = handle_update_group(plugin, data); + break; + case NEU_REQ_DEL_GROUP: + error = handle_del_group(plugin, data); + break; + case NEU_REQ_UPDATE_NODE: + error = handle_update_driver(plugin, data); + break; + case NEU_REQRESP_NODE_DELETED: { + neu_reqresp_node_deleted_t *req = data; + if (0 != strcmp(plugin->common.name, req->node)) { + error = handle_del_driver(plugin, data); + } + break; + } + default: + error = NEU_ERR_EINTERNAL; + break; + } + + return error; +} + +const neu_plugin_intf_funs_t datalayers_plugin_intf_funs = { + .open = datalayers_plugin_open, + .close = datalayers_plugin_close, + .init = datalayers_plugin_init, + .uninit = datalayers_plugin_uninit, + .start = datalayers_plugin_start, + .stop = datalayers_plugin_stop, + .setting = datalayers_plugin_config, + .request = datalayers_plugin_request, +}; \ No newline at end of file diff --git a/plugins/datalayers/datalayers_plugin_intf.h b/plugins/datalayers/datalayers_plugin_intf.h new file mode 100644 index 00000000..ba21b8b4 --- /dev/null +++ b/plugins/datalayers/datalayers_plugin_intf.h @@ -0,0 +1,79 @@ +/** + * NEURON IIoT System for Industry 4.0 + * Copyright (C) 2020-2024 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. + **/ + +#ifndef NEURON_PLUGIN_DATALAYERS_INTF_H +#define NEURON_PLUGIN_DATALAYERS_INTF_H + +#ifdef __cplusplus +extern "C" { +#endif + +#include "neuron.h" +#include "plugin.h" + +#define NEU_METRIC_CACHED_QUEUE_SIZE "cached_queue_size" +#define NEU_METRIC_CACHED_QUEUE_SIZE_TYPE NEU_METRIC_TYPE_GAUAGE +#define NEU_METRIC_CACHED_QUEUE_SIZE_HELP "Size of the cached message queue" + +#define NEU_METRIC_MAX_CACHED_QUEUE_SIZE "max_cached_queue_size" +#define NEU_METRIC_MAX_CACHED_QUEUE_SIZE_TYPE NEU_METRIC_TYPE_GAUAGE +#define NEU_METRIC_MAX_CACHED_QUEUE_SIZE_HELP \ + "Max size of the cached message queue" + +#define NEU_METRIC_DISCARDED_MSGS "discarded_msgs" +#define NEU_METRIC_DISCARDED_MSGS_TYPE NEU_METRIC_TYPE_COUNTER +#define NEU_METRIC_DISCARDED_MSGS_HELP \ + "Number of messages discarded when cache is full" + +#define NEU_PLUGIN_REGISTER_CACHED_QUEUE_SIZE_METRIC(plugin) \ + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_CACHED_QUEUE_SIZE, 0) + +#define NEU_PLUGIN_REGISTER_MAX_CACHED_QUEUE_SIZE_METRIC(plugin) \ + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_MAX_CACHED_QUEUE_SIZE, 0) + +#define NEU_PLUGIN_REGISTER_DISCARDED_MSGS_METRIC(plugin) \ + NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCARDED_MSGS, 0) + +#define NEU_PLUGIN_UPDATE_CACHED_QUEUE_SIZE_METRIC(plugin, val) \ + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_CACHED_QUEUE_SIZE, val, NULL) + +#define NEU_PLUGIN_UPDATE_MAX_CACHED_QUEUE_SIZE_METRIC(plugin, val) \ + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_MAX_CACHED_QUEUE_SIZE, val, \ + NULL) + +#define NEU_PLUGIN_UPDATE_DISCARDED_MSGS_METRIC(plugin, val) \ + NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCARDED_MSGS, val, NULL) + +extern const neu_plugin_intf_funs_t datalayers_plugin_intf_funs; + +neu_plugin_t *datalayers_plugin_open(void); +int datalayers_plugin_close(neu_plugin_t *plugin); +int datalayers_plugin_init(neu_plugin_t *plugin, bool load); +int datalayers_plugin_uninit(neu_plugin_t *plugin); +int datalayers_plugin_config(neu_plugin_t *plugin, const char *setting); +int datalayers_plugin_start(neu_plugin_t *plugin); +int datalayers_plugin_stop(neu_plugin_t *plugin); +int datalayers_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, + void *data); + +#ifdef __cplusplus +} +#endif + +#endif \ No newline at end of file diff --git a/plugins/restful/adapter_handle.c b/plugins/restful/adapter_handle.c index 7316f738..10d11808 100644 --- a/plugins/restful/adapter_handle.c +++ b/plugins/restful/adapter_handle.c @@ -44,8 +44,9 @@ void handle_add_adapter(nng_aio *aio) aio, neu_json_add_node_req_t, neu_json_decode_add_node_req, { if (strlen(req->name) >= NEU_NODE_NAME_LEN) { CHECK_NODE_NAME_LENGTH_ERR; - } else if (strcmp(req->name, "monitor") == 0) { - CHECK_NODE_MONITOR_ERR; + } else if (strcmp(req->name, "monitor") == 0 || + strcmp(req->name, "datalayers") == 0) { + CHECK_NODE_SINGLE_ERR; } else { int ret = 0; neu_reqresp_head_t header = { 0 }; @@ -87,7 +88,9 @@ static inline int send_node_update_req(nng_aio * aio, } if (strcmp(req->new_name, "monitor") == 0 || - strcmp(req->name, "monitor") == 0) { + strcmp(req->name, "monitor") == 0 || + strcmp(req->new_name, "datalayers") == 0 || + strcmp(req->name, "datalayers") == 0) { return NEU_ERR_NODE_NOT_ALLOW_UPDATE; } diff --git a/plugins/restful/datalayers_handle.c b/plugins/restful/datalayers_handle.c new file mode 100644 index 00000000..521a743e --- /dev/null +++ b/plugins/restful/datalayers_handle.c @@ -0,0 +1,149 @@ +/** + * 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 + +#include "parser/neu_json_datalayers.h" +#include "plugin.h" +#include "json/neu_json_error.h" +#include "json/neu_json_fn.h" + +#include "handle.h" +#include "utils/http.h" + +#include "datalayers_handle.h" + +void handle_datalayers_get_groups(nng_aio *aio) +{ + int ret = 0; + neu_plugin_t * plugin = neu_rest_get_plugin(); + neu_req_get_subscribe_group_t cmd = { 0 }; + neu_reqresp_head_t header = { + .ctx = aio, + .type = NEU_REQ_GET_DATALAYERS_GROUPS, + .otel_trace_type = NEU_OTEL_TRACE_TYPE_REST_COMM, + }; + + NEU_VALIDATE_JWT(aio); + + // required parameter + snprintf(cmd.app, sizeof(cmd.app), "%s", "datalayers"); + + // optional parameter + ret = neu_http_get_param_str(aio, "driver", cmd.driver, sizeof(cmd.driver)); + if (-1 == ret || (size_t) ret == sizeof(cmd.driver)) { + NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, { + neu_http_response(aio, error_code.error, result_error); + }) + return; + } + + // optional parameter + ret = neu_http_get_param_str(aio, "group", cmd.group, sizeof(cmd.group)); + if (-1 == ret || (size_t) ret == sizeof(cmd.driver)) { + NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, { + neu_http_response(aio, error_code.error, result_error); + }) + return; + } + + ret = neu_plugin_op(plugin, header, &cmd); + if (ret != 0) { + NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, { + neu_http_response(aio, NEU_ERR_IS_BUSY, result_error); + }); + } +} + +void handle_datalayers_get_groups_resp(nng_aio * aio, + neu_resp_get_subscribe_group_t *groups) +{ + char * result = NULL; + neu_json_get_subscribe_resp_t sub_grp_configs = { 0 }; + + sub_grp_configs.n_group = utarray_len(groups->groups); + sub_grp_configs.groups = calloc( + sub_grp_configs.n_group, sizeof(neu_json_get_subscribe_resp_group_t)); + + utarray_foreach(groups->groups, neu_resp_subscribe_info_t *, group) + { + int index = utarray_eltidx(groups->groups, group); + sub_grp_configs.groups[index].driver = group->driver; + sub_grp_configs.groups[index].group = group->group; + sub_grp_configs.groups[index].params = group->params; + sub_grp_configs.groups[index].static_tags = group->static_tags; + } + + neu_json_encode_by_fn(&sub_grp_configs, + neu_json_encode_datalayers_get_groups_resp, &result); + + neu_http_ok(aio, result); + free(result); + free(sub_grp_configs.groups); + utarray_free(groups->groups); +} + +void handle_datalayers_get_tags(nng_aio *aio) +{ + int ret = 0; + neu_plugin_t * plugin = neu_rest_get_plugin(); + neu_req_get_sub_driver_tags_t cmd = { 0 }; + neu_reqresp_head_t header = { + .ctx = aio, + .type = NEU_REQ_GET_DATALAYERS_TAGS, + .otel_trace_type = NEU_OTEL_TRACE_TYPE_REST_COMM, + }; + + NEU_VALIDATE_JWT(aio); + + // required parameter + snprintf(cmd.app, sizeof(cmd.app), "%s", "datalayers"); + + ret = neu_plugin_op(plugin, header, &cmd); + if (ret != 0) { + NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, { + neu_http_response(aio, NEU_ERR_IS_BUSY, result_error); + }); + } +} + +void handle_datalayers_get_tags_resp(nng_aio * aio, + neu_resp_get_sub_driver_tags_t *tags) +{ + char *result = NULL; + int ret = neu_json_encode_by_fn( + tags, neu_json_encode_datalayers_get_tags_resp, &result); + + if (ret != 0 || result == NULL) { + neu_http_response( + aio, 500, "{\"error\":4200,\"reason\":\"failed to encode tags\"}"); + return; + } + + neu_http_ok(aio, result); + + free(result); + if (tags) { + utarray_foreach(tags->infos, neu_resp_get_sub_driver_tags_info_t *, + info) + { + utarray_free(info->tags); + } + utarray_free(tags->infos); + } +} \ No newline at end of file diff --git a/plugins/restful/datalayers_handle.h b/plugins/restful/datalayers_handle.h new file mode 100644 index 00000000..432c4e1a --- /dev/null +++ b/plugins/restful/datalayers_handle.h @@ -0,0 +1,35 @@ +/** + * 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. + **/ + +#ifndef _NEU_PLUGIN_REST_DATALAYERS_HANDLE_H_ +#define _NEU_PLUGIN_REST_DATALAYERS_HANDLE_H_ + +#include + +#include "adapter.h" + +void handle_datalayers_get_groups(nng_aio *aio); +void handle_datalayers_get_groups_resp(nng_aio * aio, + neu_resp_get_subscribe_group_t *groups); + +void handle_datalayers_get_tags(nng_aio *aio); +void handle_datalayers_get_tags_resp(nng_aio * aio, + neu_resp_get_sub_driver_tags_t *tags); + +#endif \ No newline at end of file diff --git a/plugins/restful/handle.c b/plugins/restful/handle.c index 7ff39ed9..17a68be7 100644 --- a/plugins/restful/handle.c +++ b/plugins/restful/handle.c @@ -26,6 +26,7 @@ #include "adapter_handle.h" #include "cid_handle.h" +#include "datalayers_handle.h" #include "datatag_handle.h" #include "global_config_handle.h" #include "group_config_handle.h" @@ -148,6 +149,12 @@ static struct neu_http_handler cors_handler[] = { { .url = "/api/v2/users", }, + { + .url = "/api/v2/datalayers/group", + }, + { + .url = "/api/v2/datalayers/tags", + }, }; static struct neu_http_handler rest_handlers[] = { @@ -494,6 +501,18 @@ static struct neu_http_handler rest_handlers[] = { .url = "/api/v2/users", .value.handler = handle_delete_user, }, + { + .method = NEU_HTTP_METHOD_GET, + .type = NEU_HTTP_HANDLER_FUNCTION, + .url = "/api/v2/datalayers/groups", + .value.handler = handle_datalayers_get_groups, + }, + { + .method = NEU_HTTP_METHOD_GET, + .type = NEU_HTTP_HANDLER_FUNCTION, + .url = "/api/v2/datalayers/tags", + .value.handler = handle_datalayers_get_tags, + }, }; void neu_rest_handler(const struct neu_http_handler **handlers, uint32_t *size) diff --git a/plugins/restful/rest.c b/plugins/restful/rest.c index 6bcbdcb5..b8189393 100644 --- a/plugins/restful/rest.c +++ b/plugins/restful/rest.c @@ -26,6 +26,7 @@ #include "adapter_handle.h" #include "argparse.h" +#include "datalayers_handle.h" #include "datatag_handle.h" #include "define.h" #include "global_config_handle.h" @@ -283,6 +284,20 @@ static int dashb_plugin_request(neu_plugin_t * plugin, neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0); } break; + case NEU_RESP_GET_DATALAYERS_GROUPS: + handle_datalayers_get_groups_resp( + header->ctx, (neu_resp_get_subscribe_group_t *) data); + if (neu_otel_control_is_started() && trace) { + neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0); + } + break; + case NEU_RESP_GET_DATALAYERS_TAGS: + handle_datalayers_get_tags_resp( + header->ctx, (neu_resp_get_sub_driver_tags_t *) data); + if (neu_otel_control_is_started() && trace) { + neu_otel_scope_set_status_code2(scope, NEU_OTEL_STATUS_OK, 0); + } + break; case NEU_RESP_GET_NODE_SETTING: handle_get_node_setting_resp(header->ctx, (neu_resp_get_node_setting_t *) data); diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 7005242d..bc87c747 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -755,6 +755,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_RESP_GET_NODE_SETTING: case NEU_REQ_UPDATE_GROUP: case NEU_RESP_GET_SUBSCRIBE_GROUP: + case NEU_RESP_GET_DATALAYERS_GROUPS: + case NEU_RESP_GET_DATALAYERS_TAGS: case NEU_RESP_ADD_TAG: case NEU_RESP_ADD_GTAG: case NEU_RESP_UPDATE_TAG: diff --git a/src/base/msg_internal.h b/src/base/msg_internal.h index 537010a4..08fac744 100644 --- a/src/base/msg_internal.h +++ b/src/base/msg_internal.h @@ -30,108 +30,112 @@ extern "C" { #include "msg.h" -#define NEU_REQRESP_TYPE_MAP(XX) \ - XX(NEU_RESP_ERROR, neu_resp_error_t) \ - XX(NEU_REQ_READ_GROUP, neu_req_read_group_t) \ - XX(NEU_RESP_READ_GROUP, neu_resp_read_group_t) \ - XX(NEU_REQ_READ_GROUP_PAGINATE, neu_req_read_group_paginate_t) \ - XX(NEU_RESP_READ_GROUP_PAGINATE, neu_resp_read_group_paginate_t) \ - XX(NEU_REQ_TEST_READ_TAG, neu_req_test_read_tag_t) \ - XX(NEU_RESP_TEST_READ_TAG, neu_resp_test_read_tag_t) \ - XX(NEU_REQ_WRITE_TAG, neu_req_write_tag_t) \ - XX(NEU_REQ_WRITE_TAGS, neu_req_write_tags_t) \ - XX(NEU_REQ_WRITE_GTAGS, neu_req_write_gtags_t) \ - XX(NEU_REQ_SUBSCRIBE_GROUP, neu_req_subscribe_t) \ - XX(NEU_REQ_UNSUBSCRIBE_GROUP, neu_req_unsubscribe_t) \ - XX(NEU_REQ_UPDATE_SUBSCRIBE_GROUP, neu_req_subscribe_t) \ - XX(NEU_REQ_SUBSCRIBE_GROUPS, neu_req_subscribe_groups_t) \ - XX(NEU_REQ_GET_SUBSCRIBE_GROUP, neu_req_get_subscribe_group_t) \ - XX(NEU_RESP_GET_SUBSCRIBE_GROUP, neu_resp_get_subscribe_group_t) \ - XX(NEU_REQ_GET_SUB_DRIVER_TAGS, neu_req_get_sub_driver_tags_t) \ - XX(NEU_RESP_GET_SUB_DRIVER_TAGS, neu_resp_get_sub_driver_tags_t) \ - XX(NEU_REQ_NODE_INIT, neu_req_node_init_t) \ - XX(NEU_REQ_NODE_UNINIT, neu_req_node_uninit_t) \ - XX(NEU_RESP_NODE_UNINIT, neu_resp_node_uninit_t) \ - XX(NEU_REQ_ADD_NODE, neu_req_add_node_t) \ - XX(NEU_REQ_UPDATE_NODE, neu_req_update_node_t) \ - XX(NEU_REQ_DEL_NODE, neu_req_del_node_t) \ - XX(NEU_REQ_GET_NODE, neu_req_get_node_t) \ - XX(NEU_RESP_GET_NODE, neu_resp_get_node_t) \ - XX(NEU_REQ_NODE_SETTING, neu_req_node_setting_t) \ - XX(NEU_REQ_GET_NODE_SETTING, neu_req_get_node_setting_t) \ - XX(NEU_RESP_GET_NODE_SETTING, neu_resp_get_node_setting_t) \ - XX(NEU_REQ_GET_NODE_STATE, neu_req_get_node_state_t) \ - XX(NEU_RESP_GET_NODE_STATE, neu_resp_get_node_state_t) \ - XX(NEU_REQ_GET_NODES_STATE, neu_req_get_nodes_state_t) \ - XX(NEU_RESP_GET_NODES_STATE, neu_resp_get_nodes_state_t) \ - XX(NEU_REQ_NODE_CTL, neu_req_node_ctl_t) \ - XX(NEU_REQ_NODE_RENAME, neu_req_node_rename_t) \ - XX(NEU_RESP_NODE_RENAME, neu_resp_node_rename_t) \ - XX(NEU_REQ_ADD_GROUP, neu_req_add_group_t) \ - XX(NEU_REQ_DEL_GROUP, neu_req_del_group_t) \ - XX(NEU_REQ_UPDATE_GROUP, neu_req_update_group_t) \ - XX(NEU_REQ_UPDATE_DRIVER_GROUP, neu_req_update_group_t) \ - XX(NEU_RESP_UPDATE_DRIVER_GROUP, neu_resp_update_group_t) \ - XX(NEU_REQ_GET_GROUP, neu_req_get_group_t) \ - XX(NEU_RESP_GET_GROUP, neu_resp_get_group_t) \ - XX(NEU_REQ_GET_DRIVER_GROUP, neu_req_get_group_t) \ - XX(NEU_RESP_GET_DRIVER_GROUP, neu_resp_get_driver_group_t) \ - XX(NEU_REQ_ADD_TAG, neu_req_add_tag_t) \ - XX(NEU_RESP_ADD_TAG, neu_resp_add_tag_t) \ - XX(NEU_REQ_ADD_GTAG, neu_req_add_gtag_t) \ - XX(NEU_RESP_ADD_GTAG, neu_resp_add_tag_t) \ - XX(NEU_REQ_DEL_TAG, neu_req_del_tag_t) \ - XX(NEU_REQ_UPDATE_TAG, neu_req_update_tag_t) \ - XX(NEU_RESP_UPDATE_TAG, neu_resp_update_tag_t) \ - XX(NEU_REQ_GET_TAG, neu_req_get_tag_t) \ - XX(NEU_RESP_GET_TAG, neu_resp_get_tag_t) \ - XX(NEU_REQ_ADD_PLUGIN, neu_req_add_plugin_t) \ - XX(NEU_REQ_DEL_PLUGIN, neu_req_del_plugin_t) \ - XX(NEU_REQ_UPDATE_PLUGIN, neu_req_update_plugin_t) \ - XX(NEU_REQ_GET_PLUGIN, neu_req_get_plugin_t) \ - XX(NEU_RESP_GET_PLUGIN, neu_resp_get_plugin_t) \ - XX(NEU_REQRESP_TRANS_DATA, neu_reqresp_trans_data_t) \ - XX(NEU_REQRESP_NODES_STATE, neu_reqresp_nodes_state_t) \ - XX(NEU_REQRESP_NODE_DELETED, neu_reqresp_node_deleted_t) \ - XX(NEU_REQ_ADD_DRIVERS, neu_req_driver_array_t) \ - XX(NEU_REQ_UPDATE_LOG_LEVEL, neu_req_update_log_level_t) \ - XX(NEU_REQ_PRGFILE_UPLOAD, neu_req_prgfile_upload_t) \ - XX(NEU_REQ_PRGFILE_PROCESS, neu_req_prgfile_process_t) \ - XX(NEU_RESP_PRGFILE_PROCESS, neu_resp_prgfile_process_t) \ - XX(NEU_REQ_SCAN_TAGS, neu_req_scan_tags_t) \ - XX(NEU_RESP_SCAN_TAGS, neu_resp_scan_tags_t) \ - XX(NEU_REQ_CHECK_SCHEMA, neu_req_check_schema_t) \ - XX(NEU_RESP_CHECK_SCHEMA, neu_resp_check_schema_t) \ - XX(NEU_REQ_DRIVER_ACTION, neu_req_driver_action_t) \ - XX(NEU_RESP_DRIVER_ACTION, neu_resp_driver_action_t) \ - XX(NEU_REQ_DRIVER_DIRECTORY, neu_req_driver_directory_t) \ - XX(NEU_RESP_DRIVER_DIRECTORY, neu_resp_driver_directory_t) \ - XX(NEU_REQ_FUP_OPEN, neu_req_fup_open_t) \ - XX(NEU_RESP_FUP_OPEN, neu_resp_fup_open_t) \ - XX(NEU_REQ_FUP_DATA, neu_req_fup_data_t) \ - XX(NEU_RESP_FUP_DATA, neu_resp_fup_data_t) \ - XX(NEU_REQ_FDOWN_OPEN, neu_req_fdown_open_t) \ - XX(NEU_RESP_FDOWN_OPEN, neu_resp_fdown_open_t) \ - XX(NEU_REQ_FDOWN_DATA, neu_req_fdown_data_t) \ - XX(NEU_RESP_FDOWN_DATA, neu_resp_fdown_data_t) \ - XX(NEU_REQ_ADD_NODE_EVENT, neu_req_add_node_t) \ - XX(NEU_REQ_DEL_NODE_EVENT, neu_req_del_node_t) \ - XX(NEU_REQ_NODE_CTL_EVENT, neu_req_node_ctl_t) \ - XX(NEU_REQ_NODE_SETTING_EVENT, neu_req_node_setting_t) \ - XX(NEU_REQ_ADD_GROUP_EVENT, neu_req_add_group_t) \ - XX(NEU_REQ_DEL_GROUP_EVENT, neu_req_del_group_t) \ - XX(NEU_REQ_UPDATE_GROUP_EVENT, neu_req_update_group_t) \ - XX(NEU_REQ_ADD_TAG_EVENT, neu_req_add_tag_t) \ - XX(NEU_REQ_DEL_TAG_EVENT, neu_req_del_tag_t) \ - XX(NEU_REQ_UPDATE_TAG_EVENT, neu_req_update_tag_t) \ - XX(NEU_REQ_ADD_GTAG_EVENT, neu_req_add_gtag_t) \ - XX(NEU_REQ_ADD_PLUGIN_EVENT, neu_req_add_plugin_t) \ - XX(NEU_REQ_DEL_PLUGIN_EVENT, neu_req_del_plugin_t) \ - XX(NEU_REQ_SUBSCRIBE_GROUP_EVENT, neu_req_subscribe_t) \ - XX(NEU_REQ_UNSUBSCRIBE_GROUP_EVENT, neu_req_unsubscribe_t) \ - XX(NEU_REQ_UPDATE_SUBSCRIBE_GROUP_EVENT, neu_req_subscribe_t) \ - XX(NEU_REQ_SUBSCRIBE_GROUPS_EVENT, neu_req_subscribe_groups_t) \ - XX(NEU_RESP_WRITE_TAGS, neu_resp_write_tags_t) +#define NEU_REQRESP_TYPE_MAP(XX) \ + XX(NEU_RESP_ERROR, neu_resp_error_t) \ + XX(NEU_REQ_READ_GROUP, neu_req_read_group_t) \ + XX(NEU_RESP_READ_GROUP, neu_resp_read_group_t) \ + XX(NEU_REQ_READ_GROUP_PAGINATE, neu_req_read_group_paginate_t) \ + XX(NEU_RESP_READ_GROUP_PAGINATE, neu_resp_read_group_paginate_t) \ + XX(NEU_REQ_TEST_READ_TAG, neu_req_test_read_tag_t) \ + XX(NEU_RESP_TEST_READ_TAG, neu_resp_test_read_tag_t) \ + XX(NEU_REQ_WRITE_TAG, neu_req_write_tag_t) \ + XX(NEU_REQ_WRITE_TAGS, neu_req_write_tags_t) \ + XX(NEU_REQ_WRITE_GTAGS, neu_req_write_gtags_t) \ + XX(NEU_REQ_SUBSCRIBE_GROUP, neu_req_subscribe_t) \ + XX(NEU_REQ_UNSUBSCRIBE_GROUP, neu_req_unsubscribe_t) \ + XX(NEU_REQ_UPDATE_SUBSCRIBE_GROUP, neu_req_subscribe_t) \ + XX(NEU_REQ_SUBSCRIBE_GROUPS, neu_req_subscribe_groups_t) \ + XX(NEU_REQ_GET_SUBSCRIBE_GROUP, neu_req_get_subscribe_group_t) \ + XX(NEU_RESP_GET_SUBSCRIBE_GROUP, neu_resp_get_subscribe_group_t) \ + XX(NEU_REQ_GET_SUB_DRIVER_TAGS, neu_req_get_sub_driver_tags_t) \ + XX(NEU_RESP_GET_SUB_DRIVER_TAGS, neu_resp_get_sub_driver_tags_t) \ + XX(NEU_REQ_NODE_INIT, neu_req_node_init_t) \ + XX(NEU_REQ_NODE_UNINIT, neu_req_node_uninit_t) \ + XX(NEU_RESP_NODE_UNINIT, neu_resp_node_uninit_t) \ + XX(NEU_REQ_ADD_NODE, neu_req_add_node_t) \ + XX(NEU_REQ_UPDATE_NODE, neu_req_update_node_t) \ + XX(NEU_REQ_DEL_NODE, neu_req_del_node_t) \ + XX(NEU_REQ_GET_NODE, neu_req_get_node_t) \ + XX(NEU_RESP_GET_NODE, neu_resp_get_node_t) \ + XX(NEU_REQ_NODE_SETTING, neu_req_node_setting_t) \ + XX(NEU_REQ_GET_NODE_SETTING, neu_req_get_node_setting_t) \ + XX(NEU_RESP_GET_NODE_SETTING, neu_resp_get_node_setting_t) \ + XX(NEU_REQ_GET_NODE_STATE, neu_req_get_node_state_t) \ + XX(NEU_RESP_GET_NODE_STATE, neu_resp_get_node_state_t) \ + XX(NEU_REQ_GET_NODES_STATE, neu_req_get_nodes_state_t) \ + XX(NEU_RESP_GET_NODES_STATE, neu_resp_get_nodes_state_t) \ + XX(NEU_REQ_NODE_CTL, neu_req_node_ctl_t) \ + XX(NEU_REQ_NODE_RENAME, neu_req_node_rename_t) \ + XX(NEU_RESP_NODE_RENAME, neu_resp_node_rename_t) \ + XX(NEU_REQ_ADD_GROUP, neu_req_add_group_t) \ + XX(NEU_REQ_DEL_GROUP, neu_req_del_group_t) \ + XX(NEU_REQ_UPDATE_GROUP, neu_req_update_group_t) \ + XX(NEU_REQ_UPDATE_DRIVER_GROUP, neu_req_update_group_t) \ + XX(NEU_RESP_UPDATE_DRIVER_GROUP, neu_resp_update_group_t) \ + XX(NEU_REQ_GET_GROUP, neu_req_get_group_t) \ + XX(NEU_RESP_GET_GROUP, neu_resp_get_group_t) \ + XX(NEU_REQ_GET_DRIVER_GROUP, neu_req_get_group_t) \ + XX(NEU_RESP_GET_DRIVER_GROUP, neu_resp_get_driver_group_t) \ + XX(NEU_REQ_ADD_TAG, neu_req_add_tag_t) \ + XX(NEU_RESP_ADD_TAG, neu_resp_add_tag_t) \ + XX(NEU_REQ_ADD_GTAG, neu_req_add_gtag_t) \ + XX(NEU_RESP_ADD_GTAG, neu_resp_add_tag_t) \ + XX(NEU_REQ_DEL_TAG, neu_req_del_tag_t) \ + XX(NEU_REQ_UPDATE_TAG, neu_req_update_tag_t) \ + XX(NEU_RESP_UPDATE_TAG, neu_resp_update_tag_t) \ + XX(NEU_REQ_GET_TAG, neu_req_get_tag_t) \ + XX(NEU_RESP_GET_TAG, neu_resp_get_tag_t) \ + XX(NEU_REQ_ADD_PLUGIN, neu_req_add_plugin_t) \ + XX(NEU_REQ_DEL_PLUGIN, neu_req_del_plugin_t) \ + XX(NEU_REQ_UPDATE_PLUGIN, neu_req_update_plugin_t) \ + XX(NEU_REQ_GET_PLUGIN, neu_req_get_plugin_t) \ + XX(NEU_RESP_GET_PLUGIN, neu_resp_get_plugin_t) \ + XX(NEU_REQRESP_TRANS_DATA, neu_reqresp_trans_data_t) \ + XX(NEU_REQRESP_NODES_STATE, neu_reqresp_nodes_state_t) \ + XX(NEU_REQRESP_NODE_DELETED, neu_reqresp_node_deleted_t) \ + XX(NEU_REQ_ADD_DRIVERS, neu_req_driver_array_t) \ + XX(NEU_REQ_UPDATE_LOG_LEVEL, neu_req_update_log_level_t) \ + XX(NEU_REQ_PRGFILE_UPLOAD, neu_req_prgfile_upload_t) \ + XX(NEU_REQ_PRGFILE_PROCESS, neu_req_prgfile_process_t) \ + XX(NEU_RESP_PRGFILE_PROCESS, neu_resp_prgfile_process_t) \ + XX(NEU_REQ_SCAN_TAGS, neu_req_scan_tags_t) \ + XX(NEU_RESP_SCAN_TAGS, neu_resp_scan_tags_t) \ + XX(NEU_REQ_CHECK_SCHEMA, neu_req_check_schema_t) \ + XX(NEU_RESP_CHECK_SCHEMA, neu_resp_check_schema_t) \ + XX(NEU_REQ_DRIVER_ACTION, neu_req_driver_action_t) \ + XX(NEU_RESP_DRIVER_ACTION, neu_resp_driver_action_t) \ + XX(NEU_REQ_DRIVER_DIRECTORY, neu_req_driver_directory_t) \ + XX(NEU_RESP_DRIVER_DIRECTORY, neu_resp_driver_directory_t) \ + XX(NEU_REQ_FUP_OPEN, neu_req_fup_open_t) \ + XX(NEU_RESP_FUP_OPEN, neu_resp_fup_open_t) \ + XX(NEU_REQ_FUP_DATA, neu_req_fup_data_t) \ + XX(NEU_RESP_FUP_DATA, neu_resp_fup_data_t) \ + XX(NEU_REQ_FDOWN_OPEN, neu_req_fdown_open_t) \ + XX(NEU_RESP_FDOWN_OPEN, neu_resp_fdown_open_t) \ + XX(NEU_REQ_FDOWN_DATA, neu_req_fdown_data_t) \ + XX(NEU_RESP_FDOWN_DATA, neu_resp_fdown_data_t) \ + XX(NEU_REQ_ADD_NODE_EVENT, neu_req_add_node_t) \ + XX(NEU_REQ_DEL_NODE_EVENT, neu_req_del_node_t) \ + XX(NEU_REQ_NODE_CTL_EVENT, neu_req_node_ctl_t) \ + XX(NEU_REQ_NODE_SETTING_EVENT, neu_req_node_setting_t) \ + XX(NEU_REQ_ADD_GROUP_EVENT, neu_req_add_group_t) \ + XX(NEU_REQ_DEL_GROUP_EVENT, neu_req_del_group_t) \ + XX(NEU_REQ_UPDATE_GROUP_EVENT, neu_req_update_group_t) \ + XX(NEU_REQ_ADD_TAG_EVENT, neu_req_add_tag_t) \ + XX(NEU_REQ_DEL_TAG_EVENT, neu_req_del_tag_t) \ + XX(NEU_REQ_UPDATE_TAG_EVENT, neu_req_update_tag_t) \ + XX(NEU_REQ_ADD_GTAG_EVENT, neu_req_add_gtag_t) \ + XX(NEU_REQ_ADD_PLUGIN_EVENT, neu_req_add_plugin_t) \ + XX(NEU_REQ_DEL_PLUGIN_EVENT, neu_req_del_plugin_t) \ + XX(NEU_REQ_SUBSCRIBE_GROUP_EVENT, neu_req_subscribe_t) \ + XX(NEU_REQ_UNSUBSCRIBE_GROUP_EVENT, neu_req_unsubscribe_t) \ + XX(NEU_REQ_UPDATE_SUBSCRIBE_GROUP_EVENT, neu_req_subscribe_t) \ + XX(NEU_REQ_SUBSCRIBE_GROUPS_EVENT, neu_req_subscribe_groups_t) \ + XX(NEU_RESP_WRITE_TAGS, neu_resp_write_tags_t) \ + XX(NEU_REQ_GET_DATALAYERS_GROUPS, neu_req_get_subscribe_group_t) \ + XX(NEU_RESP_GET_DATALAYERS_GROUPS, neu_resp_get_subscribe_group_t) \ + XX(NEU_REQ_GET_DATALAYERS_TAGS, neu_req_get_sub_driver_tags_t) \ + XX(NEU_RESP_GET_DATALAYERS_TAGS, neu_resp_get_sub_driver_tags_t) static inline size_t neu_reqresp_size(neu_reqresp_type_e t) { diff --git a/src/core/manager.c b/src/core/manager.c index acfa419b..eb1f7892 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -1039,6 +1039,47 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) break; } + case NEU_REQ_GET_DATALAYERS_GROUPS: { + neu_req_get_subscribe_group_t *cmd = + (neu_req_get_subscribe_group_t *) &header[1]; + UT_array *groups = neu_manager_get_sub_group_deep_copy( + manager, cmd->app, cmd->driver, cmd->group); + neu_resp_get_subscribe_group_t resp = { .groups = groups }; + + strcpy(header->receiver, header->sender); + header->type = NEU_RESP_GET_DATALAYERS_GROUPS; + reply(manager, header, &resp); + break; + } + case NEU_REQ_GET_DATALAYERS_TAGS: { + neu_req_get_sub_driver_tags_t *cmd = + (neu_req_get_sub_driver_tags_t *) &header[1]; + neu_resp_get_sub_driver_tags_t resp = { 0 }; + UT_array *groups = neu_manager_get_sub_group(manager, cmd->app); + + utarray_new(resp.infos, neu_resp_get_sub_driver_tags_info_icd()); + utarray_foreach(groups, neu_resp_subscribe_info_t *, info) + { + neu_resp_get_sub_driver_tags_info_t in = { 0 }; + neu_adapter_t * driver = + neu_node_manager_find(manager->node_manager, info->driver); + assert(driver != NULL); + + strcpy(in.driver, info->driver); + strcpy(in.group, info->group); + neu_adapter_driver_get_value_tag((neu_adapter_driver_t *) driver, + info->group, &in.tags); + + utarray_push_back(resp.infos, &in); + } + utarray_free(groups); + + strcpy(header->receiver, header->sender); + header->type = NEU_RESP_GET_DATALAYERS_TAGS; + reply(manager, header, &resp); + + break; + } case NEU_REQ_GET_NODES_STATE: { neu_resp_get_nodes_state_t resp = { 0 }; UT_array *states = neu_node_manager_get_state(manager->node_manager); diff --git a/src/parser/neu_json_datalayers.c b/src/parser/neu_json_datalayers.c new file mode 100644 index 00000000..1cb8008f --- /dev/null +++ b/src/parser/neu_json_datalayers.c @@ -0,0 +1,190 @@ +/** + * 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. + **/ + +/* + * DO NOT EDIT THIS FILE MANUALLY! + * It was automatically generated by `json-autotype`. + */ + +#include +#include + +#include + +#include "msg.h" +#include "utils/log.h" +#include "json/json.h" + +#include "neu_json_datalayers.h" + +#include + +void generate_fixed_uuid(const char *input, char *out) +{ + unsigned char hash[MD5_DIGEST_LENGTH]; + MD5_CTX md5_ctx; + + MD5_Init(&md5_ctx); + MD5_Update(&md5_ctx, input, strlen(input)); + MD5_Final(hash, &md5_ctx); + + snprintf( + out, 37, + "%02x%02x%02x%02x-%02x%02x-%02x%02x-%02x%02x-%02x%02x%02x%02x%02x%02x", + hash[0], hash[1], hash[2], hash[3], hash[4], hash[5], hash[6], hash[7], + hash[8], hash[9], hash[10], hash[11], hash[12], hash[13], hash[14], + hash[15]); +} + +int neu_json_encode_datalayers_get_groups_resp(void *object, void *param) +{ + int ret = 0; + neu_json_get_subscribe_resp_t *resp = + (neu_json_get_subscribe_resp_t *) param; + + json_t *node_array = json_array(); + + for (int i = 0; i < resp->n_group; i++) { + neu_json_get_subscribe_resp_group_t *g = &resp->groups[i]; + if (!g->driver || !g->group) + continue; + + char node_id[37]; + generate_fixed_uuid(g->driver, node_id); + + json_t *node_obj = NULL; + json_t *children_array = NULL; + + for (size_t j = 0; j < json_array_size(node_array); j++) { + node_obj = json_array_get(node_array, j); + if (strcmp(json_string_value(json_object_get(node_obj, "id")), + node_id) == 0) { + break; + } + node_obj = NULL; + } + + if (!node_obj) { + node_obj = json_object(); + json_object_set_new(node_obj, "id", json_string(node_id)); + json_object_set_new(node_obj, "name", json_string(g->driver)); + json_object_set_new(node_obj, "category", json_integer(1)); + children_array = json_array(); + json_object_set_new(node_obj, "children", children_array); + json_array_append_new(node_array, node_obj); + } else { + children_array = json_object_get(node_obj, "children"); + } + + char group_id[37]; + char group_uuid_input[256]; + snprintf(group_uuid_input, sizeof(group_uuid_input), "%s-%s", g->driver, + g->group); + generate_fixed_uuid(group_uuid_input, group_id); + + json_t *group_obj = json_object(); + json_object_set_new(group_obj, "id", json_string(group_id)); + json_object_set_new(group_obj, "name", json_string(g->group)); + json_object_set_new(group_obj, "category", json_integer(2)); + json_object_set_new(group_obj, "children", json_array()); + + json_array_append_new(children_array, group_obj); + } + + json_t *final_obj = json_object(); + json_object_set_new(final_obj, "error", json_integer(0)); + json_object_set_new(final_obj, "children", node_array); + + neu_json_elem_t elems[] = { { + .name = NULL, + .t = NEU_JSON_OBJECT, + .v.val_object = final_obj, + } }; + + ret = neu_json_encode_field(object, elems, NEU_JSON_ELEM_SIZE(elems)); + + json_decref(final_obj); + + return ret; +} + +int neu_json_encode_datalayers_get_tags_resp(void *json_object_param, + void *param) +{ + neu_resp_get_sub_driver_tags_t *resp = + (neu_resp_get_sub_driver_tags_t *) param; + json_t *group_array = json_array(); + + for (unsigned int i = 0; i < utarray_len(resp->infos); i++) { + neu_resp_get_sub_driver_tags_info_t *info = + (neu_resp_get_sub_driver_tags_info_t *) utarray_eltptr(resp->infos, + i); + + char group_uuid_input[256], group_id[37]; + snprintf(group_uuid_input, sizeof(group_uuid_input), "%s-%s", + info->driver, info->group); + generate_fixed_uuid(group_uuid_input, group_id); + + json_t *group_obj = json_object(); + json_object_set_new(group_obj, "id", json_string(group_id)); + json_object_set_new(group_obj, "name", json_string(info->group)); + json_object_set_new(group_obj, "category", json_integer(2)); + + json_t *tag_array = json_array(); + + if (info->tags) { + for (unsigned int j = 0; j < utarray_len(info->tags); j++) { + neu_datatag_t *tag = + (neu_datatag_t *) utarray_eltptr(info->tags, j); + + char tag_uuid_input[512], tag_id[37]; + snprintf(tag_uuid_input, sizeof(tag_uuid_input), "%s-%s-%s", + info->driver, info->group, tag->name); + generate_fixed_uuid(tag_uuid_input, tag_id); + + json_t *tag_obj = json_object(); + json_object_set_new(tag_obj, "id", json_string(tag_id)); + json_object_set_new(tag_obj, "name", json_string(tag->name)); + json_object_set_new(tag_obj, "type", json_integer(tag->type)); + json_object_set_new(tag_obj, "category", json_integer(3)); + json_object_set_new(tag_obj, "children", json_array()); + + json_array_append_new(tag_array, tag_obj); + } + } + + json_object_set_new(group_obj, "children", tag_array); + json_array_append_new(group_array, group_obj); + } + + json_t *final_obj = json_object(); + json_object_set_new(final_obj, "error", json_integer(0)); + json_object_set_new(final_obj, "children", group_array); + + neu_json_elem_t elems[] = { { + .name = NULL, + .t = NEU_JSON_OBJECT, + .v.val_object = final_obj, + } }; + + int ret = neu_json_encode_field(json_object_param, elems, + NEU_JSON_ELEM_SIZE(elems)); + json_decref(final_obj); + return ret; +} \ No newline at end of file diff --git a/src/parser/neu_json_datalayers.h b/src/parser/neu_json_datalayers.h new file mode 100644 index 00000000..375f89e4 --- /dev/null +++ b/src/parser/neu_json_datalayers.h @@ -0,0 +1,54 @@ +/** + * 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. + **/ + +/* + * DO NOT EDIT THIS FILE MANUALLY! + * It was automatically generated by `json-autotype`. + */ + +#ifndef _NEU_JSON_API_NEU_JSON_DATALAYERS_H_ +#define _NEU_JSON_API_NEU_JSON_DATALAYERS_H_ + +#include "json/json.h" + +#ifdef __cplusplus +extern "C" { +#endif + +typedef struct { + char *group; + char *driver; + char *params; + char *static_tags; +} neu_json_get_subscribe_resp_group_t; + +typedef struct { + int n_group; + neu_json_get_subscribe_resp_group_t *groups; +} neu_json_get_subscribe_resp_t; + +int neu_json_encode_datalayers_get_groups_resp(void *json_object, void *param); + +int neu_json_encode_datalayers_get_tags_resp(void *json_object, void *param); + +#ifdef __cplusplus +} +#endif + +#endif diff --git a/src/persist/datalayer/client_example.c b/src/persist/datalayers/client_example.c similarity index 61% rename from src/persist/datalayer/client_example.c rename to src/persist/datalayers/client_example.c index 5f47c3aa..43f8d4a9 100644 --- a/src/persist/datalayer/client_example.c +++ b/src/persist/datalayers/client_example.c @@ -6,10 +6,11 @@ int main() { - Client *client = client_create("127.0.0.1", 8360, "admin", "public"); + neu_datalayers_client *client = + client_create("127.0.0.1", 8360, "admin", "public"); datatag tags[] = { - { "node_1", "group_1", "tag_1", { .string_value = "abc" }, STRING_TYPE } + { "node_1", "group_2", "tag_1", { .string_value = "abc" }, STRING_TYPE } }; size_t tag_count = sizeof(tags) / sizeof(tags[0]); @@ -57,6 +58,31 @@ int main() printf("No data found\n"); } + query_result *result_tag = client_query_all_data(client, STRING_TYPE); + if (result_tag) { + printf("Query returned %zu rows\n", result_tag->row_count); + for (size_t i = 0; i < result_tag->row_count; ++i) { + printf("Node: %s, Group: %s, Tag: %s\n", + result_tag->rows[i].node_name, + result_tag->rows[i].group_name, result_tag->rows[i].tag); + } + client_query_free(result_tag); + } else { + printf("No data found\n"); + } + + query_result *result_group = client_query_nodes_groups(client, STRING_TYPE); + if (result_group) { + printf("Query returned %zu rows\n", result_group->row_count); + for (size_t i = 0; i < result_group->row_count; ++i) { + printf("Node: %s, Group: %s\n", result_group->rows[i].node_name, + result_group->rows[i].group_name); + } + client_query_free(result_group); + } else { + printf("No data found\n"); + } + client_destroy(client); return 0; diff --git a/src/persist/datalayer/flight_sql_client.cpp b/src/persist/datalayers/flight_sql_client.cpp similarity index 67% rename from src/persist/datalayer/flight_sql_client.cpp rename to src/persist/datalayers/flight_sql_client.cpp index 2fb06636..ece28568 100644 --- a/src/persist/datalayer/flight_sql_client.cpp +++ b/src/persist/datalayers/flight_sql_client.cpp @@ -23,7 +23,10 @@ public: arrow::Status Execute(const std::string &sql); arrow::Result> Query(const std::string &sql); + bool IsInitialized() const { return initialized_; } + private: + bool initialized_ = false; flight::Location location_; std::shared_ptr flight_client_; std::unique_ptr client_; @@ -41,9 +44,9 @@ Client::Client(const std::string &host, int port, const std::string &username, arrow::Result location_result = flight::Location::ForGrpcTcp(host, port); if (!location_result.ok()) { - std::cerr << "Failed to create Flight location: " - << location_result.status().ToString() << std::endl; - exit(EXIT_FAILURE); + // std::cerr << "Failed to create Flight location: " << + // location_result.status().ToString() << std::endl; + return; } location_ = location_result.ValueOrDie(); @@ -52,21 +55,24 @@ Client::Client(const std::string &host, int port, const std::string &username, arrow::Result> flight_client_result = flight::FlightClient::Connect(location_, client_options); if (!flight_client_result.ok()) { - std::cerr << "Failed to connect to Flight client: " - << flight_client_result.status().ToString() << std::endl; - exit(EXIT_FAILURE); + // std::cerr << "Failed to connect to Flight client: " << + // flight_client_result.status().ToString() << std::endl; + return; } flight_client_ = flight_client_result.ValueOrDie(); flight::FlightCallOptions call_options; auto auth_result = AuthenticateBasicToken(call_options, username, password); if (!auth_result.ok()) { - std::cerr << "Authentication failed: " - << auth_result.status().ToString() << std::endl; - exit(EXIT_FAILURE); + // std::cerr << "Authentication failed: " << + // auth_result.status().ToString() << std::endl; + return; } bearer_token_ = auth_result.ValueOrDie(); client_ = std::make_unique(flight_client_); + + initialized_ = true; + // std::cerr << "Client initialization succeeded!" << std::endl; } arrow::Status Client::Execute(const std::string &sql) @@ -75,7 +81,7 @@ arrow::Status Client::Execute(const std::string &sql) call_options.headers.push_back( std::make_pair("authorization", "Bearer " + bearer_token_)); - std::cout << "Executing SQL: " << sql << std::endl; + // std::cout << "Executing SQL: " << sql << std::endl; arrow::Result> flight_info_result = client_->Execute(call_options, sql); @@ -156,12 +162,14 @@ Client::Query(const std::string &sql) extern "C" Client *client_create(const char *host, int port, const char *username, const char *password) { - try { - return new Client(host, port, username, password); - } catch (const std::exception &e) { - std::cerr << "Error creating client: " << e.what() << std::endl; + Client *client = new Client(host, port, username, password); + if (!client->IsInitialized()) { + delete client; + // std::cerr << "Client creation failed due to invalid parameters or + // connection issues." << std::endl; return nullptr; } + return client; } extern "C" int client_execute(Client *client, const char *sql) @@ -169,7 +177,7 @@ extern "C" int client_execute(Client *client, const char *sql) if (!client) return -1; auto status = client->Execute(sql); - return status.ok() ? 0 : 1; + return status.ok() ? 0 : -1; } extern "C" void client_destroy(Client *client) @@ -235,9 +243,9 @@ extern "C" int client_insert(Client *client, ValueType type, datatag *tags, } int status = client_execute(client, sql.c_str()); - if (status != 0) { + /*if (status != 0) { std::cerr << "Error executing SQL: " << status << std::endl; - } + }*/ return status; } @@ -259,6 +267,20 @@ std::string convert_timestamp_to_utc8(int64_t timestamp) return std::string(time_str); } +extern "C" void client_query_free(query_result *result) +{ + if (result) { + for (size_t i = 0; i < result->row_count; ++i) { + if (result->rows[i].value_type == STRING_TYPE && + result->rows[i].value.string_value) { + free((void *) result->rows[i].value.string_value); + } + } + free(result->rows); + free(result); + } +} + extern "C" query_result *client_query(Client *client, ValueType type, const char *node_name, const char *group_name, const char *tag) @@ -363,27 +385,15 @@ extern "C" query_result *client_query(Client *client, ValueType type, return result; } -extern "C" void client_query_free(query_result *result) -{ - if (result) { - for (size_t i = 0; i < result->row_count; ++i) { - if (result->rows[i].value_type == STRING_TYPE && - result->rows[i].value.string_value) { - free((void *) result->rows[i].value.string_value); - } - } - free(result->rows); - free(result); - } -} - extern "C" query_result *client_query_nodes_groups(Client * client, ValueType type) { - if (!client) + if (!client) { + std::cerr << "Client is not initialized." << std::endl; return nullptr; - std::string table_name; + } + std::string table_name; switch (type) { case INT_TYPE: table_name = "neuronex.neuron_int"; @@ -398,11 +408,13 @@ extern "C" query_result *client_query_nodes_groups(Client * client, table_name = "neuronex.neuron_string"; break; default: + std::cerr << "Invalid ValueType." << std::endl; return nullptr; } - std::string sql = "SELECT node_name, group_name FROM " + table_name; - std::cout << "Generated SQL: " << sql << std::endl; + std::string sql = + "SELECT DISTINCT node_name, group_name FROM " + table_name; + std::cout << "Generated SQL for nodes/groups: " << sql << std::endl; auto table_result = client->Query(sql); if (!table_result.ok()) { @@ -410,33 +422,52 @@ extern "C" query_result *client_query_nodes_groups(Client * client, << table_result.status().ToString() << std::endl; return nullptr; } + auto table = table_result.ValueOrDie(); - auto table = table_result.ValueOrDie(); int64_t num_rows = table->num_rows(); if (num_rows == 0) { - std::cout << "No rows found." << std::endl; + std::cout << "No nodes/groups found." << std::endl; return nullptr; } query_result *result = (query_result *) malloc(sizeof(query_result)); - result->row_count = num_rows; - result->rows = (datarow *) malloc(num_rows * sizeof(datarow)); - - auto node_col = std::dynamic_pointer_cast( - table->column(0)->chunk(0)); - auto group_col = std::dynamic_pointer_cast( - table->column(1)->chunk(0)); - - if (!node_col || !group_col) { - std::cerr << "Failed to cast column to StringArray" << std::endl; + if (!result) { + std::cerr << "Failed to allocate memory for result." << std::endl; + return nullptr; + } + result->row_count = num_rows; + result->rows = (datarow *) malloc(num_rows * sizeof(datarow)); + if (!result->rows) { + std::cerr << "Failed to allocate memory for rows." << std::endl; + free(result); return nullptr; } - for (int64_t i = 0; i < num_rows; ++i) { - strncpy(result->rows[i].node_name, node_col->GetString(i).c_str(), - sizeof(result->rows[i].node_name)); - strncpy(result->rows[i].group_name, group_col->GetString(i).c_str(), - sizeof(result->rows[i].group_name)); + int64_t row_index = 0; + arrow::ArrayVector node_chunks = table->column(0)->chunks(); + arrow::ArrayVector group_chunks = table->column(1)->chunks(); + + for (size_t chunk_idx = 0; chunk_idx < node_chunks.size(); ++chunk_idx) { + auto node_col = std::dynamic_pointer_cast( + node_chunks[chunk_idx]); + auto group_col = std::dynamic_pointer_cast( + group_chunks[chunk_idx]); + + if (!node_col || !group_col) { + std::cerr << "Chunk " << chunk_idx << " type mismatch." + << std::endl; + continue; + } + + for (int64_t i = 0; i < node_col->length(); ++i) { + snprintf(result->rows[row_index].node_name, + sizeof(result->rows[row_index].node_name), "%s", + node_col->GetString(i).c_str()); + snprintf(result->rows[row_index].group_name, + sizeof(result->rows[row_index].group_name), "%s", + group_col->GetString(i).c_str()); + row_index++; + } } return result; @@ -444,10 +475,12 @@ extern "C" query_result *client_query_nodes_groups(Client * client, extern "C" query_result *client_query_all_data(Client *client, ValueType type) { - if (!client) + if (!client) { + std::cerr << "Client is not initialized." << std::endl; return nullptr; - std::string table_name; + } + std::string table_name; switch (type) { case INT_TYPE: table_name = "neuronex.neuron_int"; @@ -462,11 +495,13 @@ extern "C" query_result *client_query_all_data(Client *client, ValueType type) table_name = "neuronex.neuron_string"; break; default: + std::cerr << "Invalid ValueType." << std::endl; return nullptr; } - std::string sql = "SELECT * FROM " + table_name; - std::cout << "Generated SQL: " << sql << std::endl; + std::string sql = "SELECT node_name, group_name, tag FROM " + table_name; + + std::cout << "Generated SQL for all data: " << sql << std::endl; auto table_result = client->Query(sql); if (!table_result.ok()) { @@ -474,145 +509,60 @@ extern "C" query_result *client_query_all_data(Client *client, ValueType type) << table_result.status().ToString() << std::endl; return nullptr; } + auto table = table_result.ValueOrDie(); - auto table = table_result.ValueOrDie(); int64_t num_rows = table->num_rows(); if (num_rows == 0) { - std::cout << "No rows found." << std::endl; + std::cout << "No node/group/tag found." << std::endl; return nullptr; } query_result *result = (query_result *) malloc(sizeof(query_result)); - result->row_count = num_rows; - result->rows = (datarow *) malloc(num_rows * sizeof(datarow)); - - auto time_col = std::dynamic_pointer_cast( - table->column(0)->chunk(0)); - auto node_col = std::dynamic_pointer_cast( - table->column(1)->chunk(0)); - auto group_col = std::dynamic_pointer_cast( - table->column(2)->chunk(0)); - auto tag_col = std::dynamic_pointer_cast( - table->column(3)->chunk(0)); - auto value_col = table->column(4); - - if (!time_col || !node_col || !group_col || !tag_col) { - std::cerr << "Failed to cast columns to StringArray" << std::endl; + if (!result) { + std::cerr << "Failed to allocate memory for result." << std::endl; return nullptr; } - for (int64_t i = 0; i < num_rows; ++i) { - auto timestamp = time_col->Value(i); - std::string time_str = convert_timestamp_to_utc8(timestamp); - strncpy(result->rows[i].time, time_str.c_str(), - sizeof(result->rows[i].time)); + result->row_count = num_rows; + result->rows = (datarow *) malloc(num_rows * sizeof(datarow)); + if (!result->rows) { + std::cerr << "Failed to allocate memory for rows." << std::endl; + free(result); + return nullptr; + } - strncpy(result->rows[i].node_name, node_col->GetString(i).c_str(), - sizeof(result->rows[i].node_name)); - strncpy(result->rows[i].group_name, group_col->GetString(i).c_str(), - sizeof(result->rows[i].group_name)); - strncpy(result->rows[i].tag, tag_col->GetString(i).c_str(), - sizeof(result->rows[i].tag)); + int64_t row_index = 0; + arrow::ArrayVector node_chunks = table->column(0)->chunks(); + arrow::ArrayVector group_chunks = table->column(1)->chunks(); + arrow::ArrayVector tag_chunks = table->column(2)->chunks(); - if (value_col->type()->Equals(arrow::int64())) { - auto value_col_int = std::dynamic_pointer_cast( - value_col->chunk(0)); - result->rows[i].value.int_value = value_col_int->Value(i); - result->rows[i].value_type = INT_TYPE; - } else if (value_col->type()->Equals(arrow::float64())) { - auto value_col_float = - std::dynamic_pointer_cast( - value_col->chunk(0)); - result->rows[i].value.float_value = value_col_float->Value(i); - result->rows[i].value_type = FLOAT_TYPE; - } else if (value_col->type()->Equals(arrow::boolean())) { - auto value_col_bool = - std::dynamic_pointer_cast( - value_col->chunk(0)); - result->rows[i].value.bool_value = value_col_bool->Value(i); - result->rows[i].value_type = BOOL_TYPE; - } else if (value_col->type()->Equals(arrow::utf8())) { - auto value_col_string = - std::dynamic_pointer_cast( - value_col->chunk(0)); - result->rows[i].value.string_value = - strdup(value_col_string->GetString(i).c_str()); - result->rows[i].value_type = STRING_TYPE; + for (size_t chunk_idx = 0; chunk_idx < node_chunks.size(); ++chunk_idx) { + auto node_col = std::dynamic_pointer_cast( + node_chunks[chunk_idx]); + auto group_col = std::dynamic_pointer_cast( + group_chunks[chunk_idx]); + auto tag_col = std::dynamic_pointer_cast( + tag_chunks[chunk_idx]); + + if (!node_col || !group_col || !tag_col) { + std::cerr << "Chunk " << chunk_idx << " type mismatch." + << std::endl; + continue; + } + + for (int64_t i = 0; i < node_col->length(); ++i) { + snprintf(result->rows[row_index].node_name, + sizeof(result->rows[row_index].node_name), "%s", + node_col->GetString(i).c_str()); + snprintf(result->rows[row_index].group_name, + sizeof(result->rows[row_index].group_name), "%s", + group_col->GetString(i).c_str()); + snprintf(result->rows[row_index].tag, + sizeof(result->rows[row_index].tag), "%s", + tag_col->GetString(i).c_str()); + row_index++; } } return result; -} - -extern "C" query_result *client_query_by_tag(Client *client, const char *tag) -{ - if (!client || !tag) - return nullptr; - std::vector table_names = { "neuronex.neuron_int", - "neuronex.neuron_float", - "neuronex.neuron_bool", - "neuronex.neuron_string" }; - - query_result *final_result = nullptr; - - for (const std::string &table_name : table_names) { - std::string sql = - "SELECT * FROM " + table_name + " WHERE tag = '" + tag + "'"; - std::cout << "Generated SQL: " << sql << std::endl; - - auto table_result = client->Query(sql); - if (!table_result.ok()) { - std::cerr << "Error executing query: " - << table_result.status().ToString() << std::endl; - continue; - } - - auto table = table_result.ValueOrDie(); - int64_t num_rows = table->num_rows(); - if (num_rows == 0) { - continue; - } - - query_result *result = (query_result *) malloc(sizeof(query_result)); - result->row_count = num_rows; - result->rows = (datarow *) malloc(num_rows * sizeof(datarow)); - - auto time_col = std::dynamic_pointer_cast( - table->column(0)->chunk(0)); - auto node_col = std::dynamic_pointer_cast( - table->column(1)->chunk(0)); - auto group_col = std::dynamic_pointer_cast( - table->column(2)->chunk(0)); - auto tag_col = std::dynamic_pointer_cast( - table->column(3)->chunk(0)); - - auto value_col = table->column(4); - - if (!time_col || !node_col || !group_col || !tag_col) { - std::cerr << "Failed to cast columns to StringArray" << std::endl; - continue; - } - - for (int64_t i = 0; i < num_rows; ++i) { - auto timestamp = time_col->Value(i); - std::string time_str = convert_timestamp_to_utc8(timestamp); - strncpy(result->rows[i].time, time_str.c_str(), - sizeof(result->rows[i].time)); - - strncpy(result->rows[i].node_name, node_col->GetString(i).c_str(), - sizeof(result->rows[i].node_name)); - strncpy(result->rows[i].group_name, group_col->GetString(i).c_str(), - sizeof(result->rows[i].group_name)); - strncpy(result->rows[i].tag, tag_col->GetString(i).c_str(), - sizeof(result->rows[i].tag)); - } - - if (!final_result) { - final_result = result; - } else { - ; - } - } - - return final_result; } \ No newline at end of file diff --git a/src/persist/datalayers/flight_sql_client.h b/src/persist/datalayers/flight_sql_client.h new file mode 100644 index 00000000..ce4b6c7b --- /dev/null +++ b/src/persist/datalayers/flight_sql_client.h @@ -0,0 +1,70 @@ +#ifndef FLIGHT_SQL_CLIENT_H +#define FLIGHT_SQL_CLIENT_H + +#include + +#ifdef __cplusplus +extern "C" { +#endif + +typedef enum { INT_TYPE, FLOAT_TYPE, BOOL_TYPE, STRING_TYPE } ValueType; + +typedef union { + int int_value; + float float_value; + bool bool_value; + const char *string_value; +} ValueUnion; + +typedef struct { + const char *node_name; + const char *group_name; + const char *tag; + ValueUnion value; + ValueType value_type; +} datatag; + +typedef struct { + ValueType value_type; + ValueUnion value; + char time[64]; + char node_name[128]; + char group_name[128]; + char tag[128]; +} datarow; + +typedef struct { + size_t row_count; + datarow *rows; +} query_result; + +typedef struct Client neu_datalayers_client; + +neu_datalayers_client *client_create(const char *host, int port, + const char *username, + const char *password); + +int client_execute(neu_datalayers_client *client, const char *sql); + +void client_destroy(neu_datalayers_client *client); + +int client_insert(neu_datalayers_client *client, ValueType type, datatag *tags, + size_t tag_count); + +query_result *client_query(neu_datalayers_client *client, ValueType type, + const char *node_name, const char *group_name, + const char *tag); + +void client_query_free(query_result *result); + +query_result *client_query_nodes_groups(neu_datalayers_client *client, + ValueType type); + +query_result *client_query_all_data(neu_datalayers_client *client, + ValueType type); + +#ifdef __cplusplus +} +#endif + +#endif \ No newline at end of file diff --git a/src/utils/http.c b/src/utils/http.c index 4450d467..9f3cec58 100644 --- a/src/utils/http.c +++ b/src/utils/http.c @@ -268,6 +268,11 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content) case NEU_ERR_MQTT_PUBLISH_FAILURE: case NEU_ERR_STRING_TOO_LONG: case NEU_ERR_FILE_WRITE_FAILURE: + case NEU_ERR_DATALAYERS_FAILURE: + case NEU_ERR_DATALAYERS_INIT_FAILURE: + case NEU_ERR_DATALAYERS_CONNECT_FAILURE: + case NEU_ERR_DATALAYERS_IS_NULL: + status = NNG_HTTP_STATUS_OK; break; case NEU_ERR_EINTERNAL: diff --git a/src/utils/json.c b/src/utils/json.c index 185ec420..e34f10f7 100644 --- a/src/utils/json.c +++ b/src/utils/json.c @@ -337,7 +337,16 @@ static json_t *encode_object(json_t *object, neu_json_elem_t ele) break; } case NEU_JSON_OBJECT: - json_object_set_new(ob, ele.name, ele.v.val_object); + if (ele.name == NULL) { + const char *key; + json_t * value; + json_object_foreach(ele.v.val_object, key, value) + { + json_object_set(ob, key, value); + } + } else { + json_object_set_new(ob, ele.name, ele.v.val_object); + } break; default: break;