diff --git a/.github/workflows/check.yml b/.github/workflows/check.yml index 1fe7dfe1..802e07e1 100644 --- a/.github/workflows/check.yml +++ b/.github/workflows/check.yml @@ -114,6 +114,7 @@ jobs: - uses: actions/checkout@v2 with: submodules: recursive + fetch-depth: 0 - name: compile run: | @@ -122,11 +123,25 @@ jobs: cmake .. -DCMAKE_TOOLCHAIN_FILE=../cmake/x86_64-linux-gnu.cmake make -j4 + - name: get neuron current branch + shell: bash + run: | + if [ '${{ github.event_name }}' == 'push' ]; then + branch=${{ github.ref_name }} + elif [ '${{ github.event_name }}' == 'pull_request' ]; then + branch=${{ github.event.pull_request.base.ref }} + elif [ '${{ github.event_name }}' == 'release' ]; then + branch=${{ github.event.release.target_commitish }} + fi + echo "BRANCH=$branch" >> $GITHUB_ENV + echo "neuron current branch is $branch" + - name: fetch neu-ft uses: actions/checkout@v2 with: repository: neugates/neu-ft path: neu-ft + ref: ${{ env.BRANCH }} - name: function test run: | diff --git a/CMakeLists.txt b/CMakeLists.txt index d545ff00..c7ad71c4 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -129,6 +129,7 @@ set(NEURON_SOURCES plugins/restful/handle.c plugins/restful/log_handle.c plugins/restful/file_handle.c + plugins/restful/metric_handle.c plugins/restful/normal_handle.c plugins/restful/rw_handle.c plugins/restful/adapter_handle.c diff --git a/ft/metrics_test.robot b/ft/metrics_test.robot index cedf2c80..d161506b 100644 --- a/ft/metrics_test.robot +++ b/ft/metrics_test.robot @@ -5,8 +5,8 @@ Resource resource/api.resource Resource resource/common.resource Resource resource/error.resource -Suite Setup Start Neuronx -Suite Teardown Stop Neuronx +Suite Setup Metrics Test Setup +Suite Teardown Metrics Test Teardown *** Test Cases *** @@ -30,4 +30,14 @@ Check node state, it should return success ${nlines}= Get metrics lines ${parameter_lines}= Get parameter lines ${nlines} - Check node state ${parameter_lines} \ No newline at end of file + Check node state ${parameter_lines} + +*** Keywords *** +Metrics Test Setup + Start Neuronx + ${res}= Add Node monitor ${PLUGIN_MONITOR} + Check Response Status ${res} 200 + +Metrics Test Teardown + Del Node monitor + Stop Neuronx diff --git a/ft/node_test.robot b/ft/node_test.robot index 8a65b412..72d30d6f 100644 --- a/ft/node_test.robot +++ b/ft/node_test.robot @@ -93,18 +93,6 @@ Update node name to the same, it should fail as name conflicts Check Response Status ${res} 409 Check Error Code ${res} ${NEU_ERR_NODE_EXIST} -Update monitor node, it should fail - ${res} = Update Node monitor monitor-new - - Check Response Status ${res} 400 - Check Error Code ${res} ${NEU_ERR_NODE_NOT_ALLOW_UPDATE} - -Update node name to monitor, it should fail - ${res} = Update Node modbus-node monitor - - Check Response Status ${res} 409 - Check Error Code ${res} ${NEU_ERR_NODE_EXIST} - Update driver node name, it should success ${res} = Update Node modbus-node modbus-node-new Check Response Status ${res} 200 diff --git a/ft/resource/common.resource b/ft/resource/common.resource index 8a7b846b..4080dbf2 100644 --- a/ft/resource/common.resource +++ b/ft/resource/common.resource @@ -18,6 +18,7 @@ ${test_data_dir} ft/data ${PLUGIN_MODBUS_TCP} Modbus TCP +${PLUGIN_MONITOR} Monitor ${PLUGIN_MQTT} MQTT ${MQTT_CMD_NONE} diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index b0605556..316f9e90 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -391,6 +391,30 @@ typedef struct { neu_datatag_t *tags; } neu_req_add_tag_t, neu_req_update_tag_t; +static inline void neu_req_add_tag_fini(neu_req_add_tag_t *req) +{ + for (uint16_t i = 0; i < req->n_tag; i++) { + neu_tag_fini(&req->tags[i]); + } + free(req->tags); +} + +static inline int neu_req_add_tag_copy(neu_req_add_tag_t *dst, + neu_req_add_tag_t *src) +{ + strcpy(dst->driver, src->driver); + strcpy(dst->group, src->group); + dst->tags = (neu_datatag_t *) calloc(src->n_tag, sizeof(src->tags[0])); + if (NULL == dst->tags) { + return -1; + } + dst->n_tag = src->n_tag; + for (uint16_t i = 0; i < src->n_tag; i++) { + neu_tag_copy(&dst->tags[i], &src->tags[i]); + } + return 0; +} + typedef struct { uint16_t index; int error; @@ -403,6 +427,36 @@ typedef struct neu_req_del_tag { char ** tags; } neu_req_del_tag_t; +static inline void neu_req_del_tag_fini(neu_req_del_tag_t *req) +{ + for (uint16_t i = 0; i < req->n_tag; i++) { + free(req->tags[i]); + } + free(req->tags); +} + +static inline int neu_req_del_tag_copy(neu_req_del_tag_t *dst, + neu_req_del_tag_t *src) +{ + strcpy(dst->driver, src->driver); + strcpy(dst->group, src->group); + dst->tags = (char **) calloc(src->n_tag, sizeof(src->tags[0])); + if (NULL == dst->tags) { + return -1; + } + dst->n_tag = src->n_tag; + for (uint16_t i = 0; i < src->n_tag; ++i) { + dst->tags[i] = strdup(src->tags[i]); + if (NULL == dst->tags[i]) { + while (i-- > 0) { + free(dst->tags[i]); + } + free(dst->tags); + } + } + return 0; +} + typedef struct neu_req_get_tag { char driver[NEU_NODE_NAME_LEN]; char group[NEU_GROUP_NAME_LEN]; @@ -491,6 +545,19 @@ typedef struct neu_req_node_setting { char *setting; } neu_req_node_setting_t; +static inline void neu_req_node_setting_fini(neu_req_node_setting_t *req) +{ + free(req->setting); +} + +static inline int neu_req_node_setting_copy(neu_req_node_setting_t *dst, + neu_req_node_setting_t *src) +{ + strcpy(dst->node, src->node); + dst->setting = strdup(src->setting); + return dst->setting ? 0 : -1; +} + typedef struct neu_req_get_node_setting { char node[NEU_NODE_NAME_LEN]; } neu_req_get_node_setting_t; diff --git a/plugins/monitor/CMakeLists.txt b/plugins/monitor/CMakeLists.txt index 10bc1e4e..47691e71 100644 --- a/plugins/monitor/CMakeLists.txt +++ b/plugins/monitor/CMakeLists.txt @@ -6,7 +6,6 @@ file(COPY ${CMAKE_SOURCE_DIR}/plugins/monitor/monitor.json DESTINATION ${CMAKE_B set(src monitor.c monitor_config.c - metric_handle.c mqtt_handle.c ) diff --git a/plugins/monitor/monitor.c b/plugins/monitor/monitor.c index 3120c05d..f396a34f 100644 --- a/plugins/monitor/monitor.c +++ b/plugins/monitor/monitor.c @@ -23,7 +23,6 @@ #include "errcodes.h" #include "event/event.h" -#include "metric_handle.h" #include "monitor.h" #include "mqtt_handle.h" #include "utils/http_handler.h" @@ -48,59 +47,6 @@ static void disconnect_cb(void *data) neu_plugin_module.module_name); } -static nng_http_server *server_init() -{ - nng_url * url = NULL; - nng_http_server *server = NULL; - - nlog_notice("monitor bind url: %s", host_port); - - int ret = nng_url_parse(&url, host_port); - if (0 != ret) { - return NULL; - } - - ret = nng_http_server_hold(&server, url); - if (0 != ret) { - nlog_error("monitor server bind error: %d", ret); - } - - nng_url_free(url); - return server; -} - -static struct neu_http_handler api_handlers[] = { - { - .method = NEU_HTTP_METHOD_GET, - .type = NEU_HTTP_HANDLER_FUNCTION, - .url = "/api/v2/metrics", - .value.handler = handle_get_metric, - }, -}; - -static int add_handlers(nng_http_server *server) -{ - struct neu_http_handler cors = { - .method = NEU_HTTP_METHOD_OPTIONS, - .type = NEU_HTTP_HANDLER_FUNCTION, - .value.handler = neu_http_handle_cors, - }; - - for (size_t i = 0; i < sizeof(api_handlers) / sizeof(api_handlers[0]); - ++i) { - if (0 != neu_http_add_handler(server, &api_handlers[0])) { - return -1; - } - - cors.url = api_handlers[i].url; - if (0 != neu_http_add_handler(server, &cors)) { - return -1; - } - } - - return 0; -} - static int config_mqtt_client(neu_plugin_t *plugin, neu_mqtt_client_t *client, const monitor_config_t *config) { @@ -253,26 +199,6 @@ static neu_plugin_t *monitor_plugin_open(void) neu_plugin_common_init(&plugin->common); - if (0 != pthread_mutex_init(&plugin->mutex, NULL)) { - free(plugin); - return NULL; - } - - plugin->api_server = server_init(); - if (NULL == plugin->api_server) { - nlog_error("fail to create monitor server"); - free(plugin); - return NULL; - } - - if (0 != add_handlers(plugin->api_server) || - 0 != nng_http_server_start(plugin->api_server)) { - nlog_error("fail to start monitor server"); - nng_http_server_release(plugin->api_server); - free(plugin); - return NULL; - } - g_monitor_plugin_ = plugin; return plugin; } @@ -281,9 +207,6 @@ static int monitor_plugin_close(neu_plugin_t *plugin) { int rv = 0; - nng_http_server_stop(plugin->api_server); - nng_http_server_release(plugin->api_server); - pthread_mutex_destroy(&plugin->mutex); plog_notice(plugin, "Success to free plugin: %s", neu_plugin_module.module_name); free(plugin); @@ -493,7 +416,7 @@ const neu_plugin_module_t neu_plugin_module = { .kind = NEU_PLUGIN_KIND_SYSTEM, .type = NEU_NA_TYPE_APP, .display = true, - .single = true, + .single = false, .single_name = "monitor", }; diff --git a/plugins/monitor/monitor.h b/plugins/monitor/monitor.h index 726e33ce..4581bc17 100644 --- a/plugins/monitor/monitor.h +++ b/plugins/monitor/monitor.h @@ -33,11 +33,9 @@ struct neu_plugin { neu_plugin_common_t common; - pthread_mutex_t mutex; bool started; neu_events_t * events; neu_event_timer_t * heartbeat_timer; - nng_http_server * api_server; monitor_config_t * config; neu_mqtt_client_t * mqtt_client; }; diff --git a/plugins/monitor/mqtt_handle.c b/plugins/monitor/mqtt_handle.c index d9852c33..a57e505f 100644 --- a/plugins/monitor/mqtt_handle.c +++ b/plugins/monitor/mqtt_handle.c @@ -322,20 +322,12 @@ int handle_events(neu_plugin_t *plugin, neu_reqresp_type_e event, void *data) end: if (NEU_REQ_NODE_SETTING_EVENT == event) { - free(((neu_req_node_setting_t *) data)->setting); + neu_req_node_setting_fini(data); } else if (NEU_REQ_ADD_TAG_EVENT == event || NEU_REQ_UPDATE_TAG_EVENT == event) { - neu_req_add_tag_t *add_tag = data; - for (uint16_t i = 0; i < add_tag->n_tag; i++) { - neu_tag_fini(&add_tag->tags[i]); - } - free(add_tag->tags); + neu_req_add_tag_fini(data); } else if (NEU_REQ_DEL_TAG_EVENT == event) { - neu_req_del_tag_t *del_tag = data; - for (uint16_t i = 0; i < del_tag->n_tag; i++) { - free(del_tag->tags[i]); - } - free(del_tag->tags); + neu_req_del_tag_fini(data); } return rv; } diff --git a/plugins/restful/global_config_handle.c b/plugins/restful/global_config_handle.c index 5b1c1386..d41ce0d1 100644 --- a/plugins/restful/global_config_handle.c +++ b/plugins/restful/global_config_handle.c @@ -103,7 +103,10 @@ static int add_setting(context_t *ctx, neu_json_node_setting_req_t *data); static inline bool is_static_node(const char *name, const char *plugin) { - return 0 == strcmp("monitor", name) && 0 == strcmp("Monitor", plugin); + (void) name; + (void) plugin; + return false; + // return 0 == strcmp("monitor", name) && 0 == strcmp("Monitor", plugin); } static context_t *context_new(nng_aio *aio, bool get) diff --git a/plugins/restful/handle.c b/plugins/restful/handle.c index 1a37797a..af0aecfa 100644 --- a/plugins/restful/handle.c +++ b/plugins/restful/handle.c @@ -30,6 +30,7 @@ #include "global_config_handle.h" #include "group_config_handle.h" #include "log_handle.h" +#include "metric_handle.h" #include "ndriver_handle.h" #include "normal_handle.h" #include "plugin_handle.h" @@ -143,6 +144,9 @@ static struct neu_http_handler cors_handler[] = { { .url = "/api/v2/ndriver/tag", }, + { + .url = "/api/v2/metrics", + }, }; static struct neu_http_handler rest_handlers[] = { @@ -502,6 +506,12 @@ static struct neu_http_handler rest_handlers[] = { .url = "/api/v2/ndriver/tag", .value.handler = handle_get_ndriver_tags, }, + { + .method = NEU_HTTP_METHOD_GET, + .type = NEU_HTTP_HANDLER_FUNCTION, + .url = "/api/v2/metrics", + .value.handler = handle_get_metric, + }, }; void neu_rest_handler(const struct neu_http_handler **handlers, uint32_t *size) diff --git a/plugins/monitor/metric_handle.c b/plugins/restful/metric_handle.c similarity index 97% rename from plugins/monitor/metric_handle.c rename to plugins/restful/metric_handle.c index 7c43c889..575b8196 100644 --- a/plugins/monitor/metric_handle.c +++ b/plugins/restful/metric_handle.c @@ -1,6 +1,6 @@ /** * NEURON IIoT System for Industry 4.0 - * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved. + * Copyright (C) 2020-2023 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 @@ -22,12 +22,11 @@ #include "define.h" #include "metrics.h" #include "utils/asprintf.h" +#include "utils/http.h" +#include "utils/http_handler.h" #include "utils/log.h" #include "metric_handle.h" -#include "monitor.h" -#include "utils/http.h" -#include "utils/http_handler.h" // clang-format off #define METRIC_GLOBAL_TMPL \ @@ -298,19 +297,18 @@ static void gen_node_metrics(const neu_metrics_t *metrics, struct context *ctx) void handle_get_metric(nng_aio *aio) { - int status = NNG_HTTP_STATUS_OK; - char * result = NULL; - size_t len = 0; - FILE * stream = NULL; - neu_plugin_t *plugin = neu_monitor_get_plugin(); + int status = NNG_HTTP_STATUS_OK; + char * result = NULL; + size_t len = 0; + FILE * stream = NULL; neu_metrics_category_e cat = NEU_METRICS_CATEGORY_ALL; size_t cat_param_len = 0; const char *cat_param = neu_http_get_param(aio, "category", &cat_param_len); if (NULL != cat_param && !parse_metrics_catgory(cat_param, cat_param_len, &cat)) { - plog_error(plugin, "invalid metrics category: %.*s", - (int) cat_param_len, cat_param); + nlog_error("invalid metrics category: %.*s", (int) cat_param_len, + cat_param); status = NNG_HTTP_STATUS_BAD_REQUEST; goto end; } diff --git a/plugins/monitor/metric_handle.h b/plugins/restful/metric_handle.h similarity index 86% rename from plugins/monitor/metric_handle.h rename to plugins/restful/metric_handle.h index dcad7739..ebd92a2f 100644 --- a/plugins/monitor/metric_handle.h +++ b/plugins/restful/metric_handle.h @@ -1,6 +1,6 @@ /** * NEURON IIoT System for Industry 4.0 - * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved. + * Copyright (C) 2020-2023 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 @@ -17,8 +17,8 @@ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ -#ifndef NEU_PLUGIN_MONITOR_METRIC_H -#define NEU_PLUGIN_MONITOR_METRIC_H +#ifndef NEU_PLUGIN_REST_METRIC_H +#define NEU_PLUGIN_REST_METRIC_H #include diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index ed0bb6ac..7d694bb4 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -1347,7 +1347,7 @@ inline static void notify_monitor(neu_adapter_t * adapter, neu_reqresp_type_e event, void *data) { neu_reqresp_head_t header = { - .receiver = "monitor", + .receiver = "manager", .type = event, }; diff --git a/src/core/manager.c b/src/core/manager.c index 2627fd51..5567cc25 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -58,8 +58,8 @@ inline static void forward_msg_dup(neu_manager_t *manager, nng_msg *msg, nng_pipe pipe, msg_dup dup); inline static void forward_msg(neu_manager_t *manager, nng_msg *msg, const char *node); -inline static void notify_monitor(neu_manager_t * manager, - neu_reqresp_type_e event, void *data); +inline static void notify_monitors(neu_manager_t * manager, + neu_reqresp_type_e event, void *data); static void start_static_adapter(neu_manager_t *manager, const char *name); static int update_timestamp(void *usr_data); static void start_single_adapter(neu_manager_t *manager, const char *name, @@ -588,7 +588,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) if (error == NEU_ERR_SUCCESS) { manager_storage_add_node(manager, cmd->node); - notify_monitor(manager, NEU_REQ_ADD_NODE_EVENT, cmd); + notify_monitors(manager, NEU_REQ_ADD_NODE_EVENT, cmd); } header->type = NEU_RESP_ERROR; @@ -600,12 +600,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_req_update_node_t *cmd = (neu_req_update_node_t *) &header[1]; neu_resp_error_t e = { 0 }; - if (0 == strcmp("monitor", cmd->node)) { - e.error = NEU_ERR_NODE_NOT_ALLOW_UPDATE; - } else if (0 == strcmp("monitor", cmd->new_name)) { - e.error = NEU_ERR_NODE_EXIST; - } else if (NULL == - neu_node_manager_find(manager->node_manager, cmd->node)) { + if (NULL == neu_node_manager_find(manager->node_manager, cmd->node)) { e.error = NEU_ERR_NODE_NOT_EXIST; } else if (NULL != neu_node_manager_find(manager->node_manager, @@ -657,7 +652,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) } header->type = NEU_REQ_NODE_UNINIT; forward_msg(manager, msg, header->receiver); - notify_monitor(manager, NEU_REQ_DEL_NODE_EVENT, cmd); + notify_monitors(manager, NEU_REQ_DEL_NODE_EVENT, cmd); if (neu_adapter_get_type(adapter) == NEU_NA_TYPE_DRIVER) { UT_array *apps = neu_subscribe_manager_find_by_driver( manager->subscribe_manager, cmd->node); @@ -936,16 +931,14 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_REQ_ADD_NODE_EVENT: case NEU_REQ_DEL_NODE_EVENT: case NEU_REQ_NODE_CTL_EVENT: + case NEU_REQ_NODE_SETTING_EVENT: case NEU_REQ_ADD_GROUP_EVENT: case NEU_REQ_DEL_GROUP_EVENT: case NEU_REQ_UPDATE_GROUP_EVENT: case NEU_REQ_ADD_TAG_EVENT: case NEU_REQ_DEL_TAG_EVENT: case NEU_REQ_UPDATE_TAG_EVENT: { - if (neu_node_manager_find(manager->node_manager, header->receiver)) { - forward_msg(manager, msg, header->receiver); - } - + notify_monitors(manager, header->type, &header[1]); break; } @@ -1027,8 +1020,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) break; } - case NEU_REQ_NODE_SETTING: - case NEU_REQ_NODE_SETTING_EVENT: { + case NEU_REQ_NODE_SETTING: { neu_req_node_setting_t *cmd = (neu_req_node_setting_t *) &header[1]; if (neu_node_manager_find(manager->node_manager, header->receiver) == @@ -1225,30 +1217,94 @@ inline static void forward_msg(neu_manager_t *manager, nng_msg *msg, } } -inline static void notify_monitor(neu_manager_t * manager, - neu_reqresp_type_e event, void *data) +struct notify_monitor_ctx { + neu_manager_t * manager; + neu_reqresp_type_e event; + void * data; +}; + +static int notify_monitor(const char *name, nng_pipe pipe, void *arg) { - nng_pipe pipe = neu_node_manager_get_pipe(manager->node_manager, "monitor"); + struct notify_monitor_ctx *ctx = arg; + void * data = NULL; + neu_req_node_setting_t setting = { 0 }; + neu_req_add_tag_t mod_tag = { 0 }; + neu_req_del_tag_t del_tag = { 0 }; + + // monitor not ready, ignore if (0 == pipe.id) { - nlog_error("no monitor node"); - return; + return 0; } - neu_reqresp_head_t header = { - .sender = "manager", - .receiver = "monitor", - .type = event, - }; + // copy message data + switch (ctx->event) { + case NEU_REQ_NODE_SETTING_EVENT: { + if (0 != neu_req_node_setting_copy(&setting, ctx->data)) { + return -1; + } + data = &setting; + break; + } + case NEU_REQ_ADD_TAG_EVENT: + case NEU_REQ_UPDATE_TAG_EVENT: { + if (0 != neu_req_add_tag_copy(&mod_tag, ctx->data)) { + return -1; + } + data = &mod_tag; + break; + } + case NEU_REQ_DEL_TAG_EVENT: { + if (0 != neu_req_del_tag_copy(&del_tag, ctx->data)) { + return -1; + } + data = &del_tag; + break; + } + default: + data = ctx->data; + } + // message header + neu_reqresp_head_t header = { + .sender = "manager", + .type = ctx->event, + }; + strcpy(header.receiver, name); + + // send message nng_msg *msg = neu_msg_gen(&header, data); nng_msg_set_pipe(msg, pipe); - int ret = nng_sendmsg(manager->socket, msg, 0); + int ret = nng_sendmsg(ctx->manager->socket, msg, 0); if (ret != 0) { nng_msg_free(msg); nlog_warn("notify %s of %s, error: %s", header.receiver, neu_reqresp_type_string(header.type), nng_strerror(ret)); } + + return 0; +} + +inline static void notify_monitors(neu_manager_t * manager, + neu_reqresp_type_e event, void *data) +{ + struct notify_monitor_ctx ctx = { + .manager = manager, + .event = event, + .data = data, + }; + + neu_node_manager_for_each_monitor(manager->node_manager, notify_monitor, + &ctx); + + if (NEU_REQ_NODE_SETTING_EVENT == event) { + neu_req_node_setting_fini(data); + } else if (NEU_REQ_ADD_TAG_EVENT == event || + NEU_REQ_UPDATE_TAG_EVENT == event) { + neu_req_add_tag_fini(data); + } else if (NEU_REQ_DEL_TAG_EVENT == event) { + neu_req_del_tag_fini(data); + } } static void start_static_adapter(neu_manager_t *manager, const char *name) diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c index 71938ec6..74c7eb28 100644 --- a/src/core/manager_internal.c +++ b/src/core/manager_internal.c @@ -684,7 +684,7 @@ int neu_manager_subscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group, const char *params) { - if (0 == strcmp(app, "monitor")) { + if (neu_node_manager_is_monitor(manager->node_manager, app)) { // filter out monitor node return NEU_ERR_NODE_NOT_ALLOW_SUBSCRIBE; } diff --git a/src/core/node_manager.c b/src/core/node_manager.c index 0d96b9b6..79f9ff0e 100644 --- a/src/core/node_manager.c +++ b/src/core/node_manager.c @@ -32,6 +32,7 @@ typedef struct node_entity { bool is_static; bool display; bool single; + bool is_monitor; nng_pipe pipe; UT_hash_handle hh; @@ -40,12 +41,17 @@ typedef struct node_entity { struct neu_node_manager { node_entity_t *nodes; + UT_array * monitors; }; neu_node_manager_t *neu_node_manager_create() { neu_node_manager_t *node_manager = calloc(1, sizeof(neu_node_manager_t)); + if (node_manager) { + utarray_new(node_manager->monitors, &ut_ptr_icd); + } + return node_manager; } @@ -60,6 +66,7 @@ void neu_node_manager_destroy(neu_node_manager_t *mgr) free(el); } + utarray_free(mgr->monitors); free(mgr); } @@ -70,9 +77,15 @@ int neu_node_manager_add(neu_node_manager_t *mgr, neu_adapter_t *adapter) node->adapter = adapter; node->name = strdup(adapter->name); node->display = true; + node->is_monitor = + (0 == strcmp(node->adapter->module->module_name, "Monitor")); HASH_ADD_STR(mgr->nodes, name, node); + if (node->is_monitor) { + utarray_push_back(mgr->monitors, &node); + } + return 0; } @@ -149,6 +162,10 @@ void neu_node_manager_del(neu_node_manager_t *mgr, const char *name) HASH_FIND_STR(mgr->nodes, name, node); if (node != NULL) { HASH_DEL(mgr->nodes, node); + if (node->is_monitor) { + utarray_erase(mgr->monitors, utarray_eltidx(mgr->monitors, node), + 1); + } free(node->name); free(node); } @@ -354,6 +371,31 @@ nng_pipe neu_node_manager_get_pipe(neu_node_manager_t *mgr, const char *name) return pipe; } +bool neu_node_manager_is_monitor(neu_node_manager_t *mgr, const char *name) +{ + node_entity_t *node = NULL; + HASH_FIND_STR(mgr->nodes, name, node); + return node ? node->is_monitor : false; +} + +int neu_node_manager_for_each_monitor(neu_node_manager_t *mgr, + int (*cb)(const char *name, nng_pipe pipe, + void *data), + void *data) +{ + int rv = 0; + + utarray_foreach(mgr->monitors, node_entity_t **, node) + { + rv = cb((*node)->name, (*node)->pipe, data); + if (0 != rv) { + break; + } + } + + return rv; +} + UT_array *neu_node_manager_get_state(neu_node_manager_t *mgr) { UT_icd icd = { sizeof(neu_nodes_state_t), NULL, NULL, NULL }; diff --git a/src/core/node_manager.h b/src/core/node_manager.h index 6df3abe2..3b7640d8 100644 --- a/src/core/node_manager.h +++ b/src/core/node_manager.h @@ -62,6 +62,12 @@ UT_array *neu_node_manager_get_pipes(neu_node_manager_t *mgr, int type); UT_array *neu_node_manager_get_pipes_all(neu_node_manager_t *mgr); nng_pipe neu_node_manager_get_pipe(neu_node_manager_t *mgr, const char *name); +bool neu_node_manager_is_monitor(neu_node_manager_t *mgr, const char *name); +int neu_node_manager_for_each_monitor(neu_node_manager_t *mgr, + int (*)(const char *name, nng_pipe pipe, + void *data), + void *data); + // neu_nodes_state_t array UT_array *neu_node_manager_get_state(neu_node_manager_t *mgr); diff --git a/src/persist/persist.c b/src/persist/persist.c index 0540a4b8..23cb8b3a 100644 --- a/src/persist/persist.c +++ b/src/persist/persist.c @@ -600,21 +600,12 @@ void neu_persister_destroy() int neu_persister_store_node(neu_persist_node_info_t *info) { int rv = 0; - if (strcmp(info->plugin_name, "Monitor") == 0) { - rv = execute_sql( - global_db, - "INSERT OR IGNORE INTO nodes (name, type, state, plugin_name) " - "VALUES (%Q, %i, %i, %Q)", - info->name, info->type, info->state, info->plugin_name); - } else { - rv = - execute_sql(global_db, - "INSERT INTO nodes (name, type, state, plugin_name) " - "VALUES (%Q, %i, %i, %Q)", - info->name, info->type, info->state, info->plugin_name); - if (0 == rv) { - node_count_add(1); - } + rv = execute_sql(global_db, + "INSERT INTO nodes (name, type, state, plugin_name) " + "VALUES (%Q, %i, %i, %Q)", + info->name, info->type, info->state, info->plugin_name); + if (0 == rv) { + node_count_add(1); } return rv; }