diff --git a/CMakeLists.txt b/CMakeLists.txt index 044086bd..a81ca07d 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -91,6 +91,8 @@ set(NEURON_SOURCES src/core/subscribe.c src/core/plugin_manager.c src/core/node_manager.c + src/core/storage.c + src/adapter/storage.c src/adapter/adapter.c src/adapter/driver/cache.c src/adapter/driver/driver.c @@ -107,7 +109,6 @@ set(NEURON_SOURCES plugins/restful/version_handle.c plugins/restful/rest.c plugins/restful/http.c - plugins/dummy/dummy.c ${PERSIST_SOURCES}) set(CMAKE_BUILD_RPATH ./) diff --git a/ft/tag_test.robot b/ft/tag_test.robot index 5a493358..994fb503 100644 --- a/ft/tag_test.robot +++ b/ft/tag_test.robot @@ -17,7 +17,7 @@ ${tag_desc} {"name": "tag_desc", "description": "description info", Add tags to non-existent node, it should return failure ${res}= Add Tags modbus-node group ${tag1},${tag2} - Check Response Status ${res} 200 + Check Response Status ${res} 404 Check Error Code ${res} ${NEU_ERR_NODE_NOT_EXIST} Add tags to the non-existent group, it should return failure diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 4459606b..71edb913 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -41,7 +41,6 @@ typedef enum neu_reqresp_type { NEU_RESP_ERROR, NEU_REQRESP_TRANS_DATA, NEU_REQ_UPDATE_LICENSE, - NEU_REQRESP_PERSISTENCE_LOAD, NEU_REQ_READ_GROUP, NEU_RESP_READ_GROUP, @@ -56,6 +55,8 @@ typedef enum neu_reqresp_type { NEU_REQ_APP_UNSUBSCRIBE_GROUP, NEU_REQ_NODE_INIT, + NEU_REQ_NODE_UNINIT, + NEU_RESP_NODE_UNINIT, NEU_REQ_ADD_NODE, NEU_REQ_DEL_NODE, NEU_REQ_GET_NODE, @@ -98,9 +99,9 @@ typedef struct neu_resp_error { int error; } neu_resp_error_t; -typedef struct neu_req_node_init { +typedef struct { char node[NEU_NODE_NAME_LEN]; -} neu_req_node_init_t; +} neu_req_node_init_t, neu_req_node_uninit_t, neu_resp_node_uninit_t; typedef struct neu_req_add_plugin { char library[NEU_PLUGIN_LIBRARY_LEN]; @@ -304,9 +305,9 @@ inline static nng_msg *neu_msg_gen(neu_reqresp_head_t *header, void *data) size_t data_size = 0; switch (header->type) { - case NEU_REQRESP_PERSISTENCE_LOAD: - break; case NEU_REQ_NODE_INIT: + case NEU_REQ_NODE_UNINIT: + case NEU_RESP_NODE_UNINIT: data_size = sizeof(neu_req_node_init_t); break; case NEU_RESP_ERROR: diff --git a/plugins/dummy/dummy.c b/plugins/dummy/dummy.c deleted file mode 100644 index 7eba629c..00000000 --- a/plugins/dummy/dummy.c +++ /dev/null @@ -1,126 +0,0 @@ -/** - * NEURON IIoT System for Industry 4.0 - * Copyright (C) 2020-2022 EMQ Technologies Co., Ltd All rights reserved. - * - * This program is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 3 of the License, or (at your option) any later version. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public License - * along with this program; if not, write to the Free Software Foundation, - * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. - **/ - -#include -#include -#include - -#include -#include - -#include "dummy.h" -#include "plugin.h" -#include "utils/log.h" - -#define neu_plugin_module default_dummy_plugin_module - -struct neu_plugin { - neu_plugin_common_t common; -}; - -static neu_plugin_t *dummy_plugin_open() -{ - neu_plugin_t *plugin = calloc(1, sizeof(neu_plugin_t)); - - neu_plugin_common_init(&plugin->common); - - nlog_info("Success to create plugin: %s", neu_plugin_module.module_name); - return plugin; -} - -static int dummy_plugin_close(neu_plugin_t *plugin) -{ - free(plugin); - nlog_info("Success to free plugin: %s", neu_plugin_module.module_name); - return 0; -} - -static int dummy_plugin_init(neu_plugin_t *plugin) -{ - int rv = 0; - - (void) plugin; - - nlog_info("Initialize plugin: %s", neu_plugin_module.module_name); - return rv; -} - -static int dummy_plugin_uninit(neu_plugin_t *plugin) -{ - int rv = 0; - - (void) plugin; - - nlog_info("Uninitialize plugin: %s", neu_plugin_module.module_name); - return rv; -} - -static int dummy_plugin_config(neu_plugin_t *plugin, const char *configs) -{ - int rv = 0; - - (void) plugin; - (void) configs; - - nlog_info("config plugin: %s", neu_plugin_module.module_name); - return rv; -} - -static int dummy_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, - void *data) -{ - (void) plugin; - (void) head; - (void) data; - return 0; -} - -static int dummy_plugin_start(neu_plugin_t *plugin) -{ - (void) plugin; - return 0; -} - -static int dummy_plugin_stop(neu_plugin_t *plugin) -{ - (void) plugin; - return 0; -} - -static const neu_plugin_intf_funs_t plugin_intf_funs = { - .open = dummy_plugin_open, - .close = dummy_plugin_close, - .init = dummy_plugin_init, - .uninit = dummy_plugin_uninit, - .start = dummy_plugin_start, - .stop = dummy_plugin_stop, - .setting = dummy_plugin_config, - .request = dummy_plugin_request, -}; - -#define DEFAULT_DUMMY_PLUGIN_DESCR "A plugin that does nothing" - -const neu_plugin_module_t default_dummy_plugin_module = { - .version = NEURON_PLUGIN_VER_1_0, - .module_name = "neuron-default-dummy", - .module_descr = DEFAULT_DUMMY_PLUGIN_DESCR, - .intf_funs = &plugin_intf_funs, - .kind = NEU_PLUGIN_KIND_STATIC, - .type = NEU_NA_TYPE_APP, -}; diff --git a/plugins/mqtt/mqtt.c b/plugins/mqtt/mqtt.c index ac99767c..e5935224 100644 --- a/plugins/mqtt/mqtt.c +++ b/plugins/mqtt/mqtt.c @@ -519,10 +519,10 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, error = trans_data(plugin, data); break; case NEU_REQ_APP_SUBSCRIBE_GROUP: { - neu_req_app_subscribe_group_t *sub = - (neu_req_app_subscribe_group_t *) data; + // neu_req_app_subscribe_group_t *sub = + //(neu_req_app_subscribe_group_t *) data; - utarray_free(sub->tags); + // utarray_free(sub->tags); break; } default: diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 601b4bbc..7a9953df 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -29,403 +29,117 @@ #include #include +#include "utils/log.h" + #include "adapter.h" #include "adapter_internal.h" #include "driver/driver_internal.h" #include "errcodes.h" #include "persist/persist.h" #include "plugin.h" -#include "utils/log.h" +#include "storage.h" -static neu_persister_t *g_persister_singleton = NULL; +static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data); +static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header, + void *data); +static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header, + void *data); +inline static void reply(neu_adapter_t *adapter, neu_reqresp_head_t *header, + void *data); -static void persister_singleton_init() +static const adapter_callbacks_t callback_funs = { + .command = adapter_command, + .response = adapter_response, +}; + +neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool start) { - const char *persistence_dir = "persistence"; + int rv = 0; + neu_adapter_t * adapter = NULL; + neu_event_io_param_t param = { 0 }; - neu_persister_t *persister = neu_persister_create(persistence_dir); - if (NULL == persister) { - nlog_error("can't create persister"); - exit(EXIT_FAILURE); - } - g_persister_singleton = persister; -} - -// NOTE: this function is not thread safe! -static neu_persister_t *persister_singleton_get() -{ - if (NULL == g_persister_singleton) { - persister_singleton_init(); - } - return g_persister_singleton; -} - -static int persister_singleton_load_plugins(neu_adapter_t *adapter) -{ - UT_array * plugin_infos = NULL; - neu_persister_t *persister = persister_singleton_get(); - - int rv = neu_persister_load_plugins(persister, &plugin_infos); - if (rv != 0) { - return rv; - } - - utarray_foreach(plugin_infos, char **, name) - { - rv = neu_manager_add_plugin(adapter->manager, *name); - const char *ok_or_err = (0 == rv) ? "success" : "fail"; - nlog_info("%s load plugin %s, lib:%s", adapter->name, ok_or_err, *name); - } - - utarray_foreach(plugin_infos, char **, name) { free(*name); } - utarray_free(plugin_infos); - - return rv; -} - -static int persister_singleton_load_setting(neu_adapter_t *adapter, - const char * adapter_name) -{ - neu_persister_t *persister = persister_singleton_get(); - const char * setting = NULL; - int rv = - neu_persister_load_adapter_setting(persister, adapter_name, &setting); - if (0 != rv) { - const char *fail_or_ignore = "fail"; - if (rv != 0) { - // ignore this error, no setting ever set - rv = 0; - fail_or_ignore = "ignore"; - } - nlog_info("%s %s load setting of %s", adapter->name, fail_or_ignore, - adapter_name); - return rv; - } - - rv = neu_manager_node_setting(adapter->manager, adapter_name, setting); - const char *ok_or_err = (0 == rv) ? "success" : "fail"; - nlog_info("%s %s set setting of %s %s", adapter->name, ok_or_err, - adapter_name, setting); - - free((char *) setting); - - return rv; -} - -static int persister_singleton_load_datatags(neu_adapter_t *adapter, - const char * driver, - const char * group) -{ - neu_persister_t *persister = persister_singleton_get(); - UT_array * tags = NULL; - int rv = neu_persister_load_datatags(persister, driver, group, &tags); - if (0 != rv) { - const char *fail_or_ignore = "fail"; - if (rv != 0) { - // ignore no datatags - rv = 0; - fail_or_ignore = "ignore"; - } - nlog_info("%s %s load datatags info of adapter:%s grp:%s", - adapter->name, fail_or_ignore, driver, group); - return rv; - } - - utarray_foreach(tags, neu_datatag_t *, tag) - { - uint16_t index = 0; - neu_manager_add_tag(adapter->manager, driver, group, 1, tag, &index); - (void) index; - } - - neu_persist_datatag_infos_free(tags); - - return rv; -} - -static int persister_singleton_load_grp_and_tags(neu_adapter_t *adapter, - const char * adapter_name) -{ - UT_array * group_config_infos = NULL; - neu_persister_t *persister = persister_singleton_get(); - - int rv = neu_persister_load_group_configs(persister, adapter_name, - &group_config_infos); - if (0 != rv) { - const char *fail_or_ignore = "fail"; - if (rv != 0) { - // ignore no group configs - rv = 0; - fail_or_ignore = "ignore"; - } - nlog_error("%s %s load grp config of %s", adapter->name, fail_or_ignore, - adapter_name); - return rv; - } - - utarray_foreach(group_config_infos, neu_persist_group_config_info_t *, p) - { - neu_manager_add_group(adapter->manager, p->adapter_name, - p->group_config_name, p->read_interval); - - rv = persister_singleton_load_datatags(adapter, p->adapter_name, - p->group_config_name); - if (0 != rv) { - nlog_warn("%s load datatags of adapter:%s grp:%s fail, ignore", - adapter->name, p->adapter_name, p->group_config_name); - } - } - - neu_persist_group_config_infos_free(group_config_infos); - return rv; -} - -static int persister_singleton_load_subscriptions(neu_adapter_t *adapter, - const char * adapter_name) -{ - int rv = 0; - neu_persister_t *persister = persister_singleton_get(); - UT_array * sub_infos = NULL; - - rv = neu_persister_load_subscriptions(persister, adapter_name, &sub_infos); - if (0 != rv) { - const char *fail_or_ignore = "fail"; - if (rv != 0) { - // ignore no subscriptions - rv = 0; - fail_or_ignore = "ignore"; - } - nlog_error("%s %s load subscription infos of %s", adapter->name, - fail_or_ignore, adapter_name); - return rv; - } - - utarray_foreach(sub_infos, neu_persist_subscription_info_t *, info) - { - rv = neu_manager_subscribe(adapter->manager, info->sub_adapter_name, - info->src_adapter_name, - info->group_config_name); - const char *ok_or_err = (0 == rv) ? "success" : "fail"; - nlog_info("%s %s load subscription app:%s driver:%s grp:%s", - adapter->name, ok_or_err, adapter_name, - info->src_adapter_name, info->group_config_name); - } - - neu_persist_subscription_infos_free(sub_infos); - - return rv; -} - -static int persister_singleton_load_data(neu_adapter_t *adapter) -{ - UT_array * adapter_infos = NULL; - neu_persister_t *persister = persister_singleton_get(); - - nlog_info("%s start persistence loading", adapter->name); - - int rv = persister_singleton_load_plugins(adapter); - if (0 != rv) { - return rv; - } - - rv = neu_persister_load_adapters(persister, &adapter_infos); - if (0 != rv) { - nlog_error("%s failed to load adapter infos", adapter->name); - goto error_load_adapters; - } - - // sort by adapter id - // qsort(adapter_infos->data, adapter_infos->size, - // sizeof(neu_persist_adapter_info_t), persist_adapter_info_cmp); - - utarray_foreach(adapter_infos, neu_persist_adapter_info_t *, adapter_info) - { - rv = neu_manager_add_node(adapter->manager, adapter_info->name, - adapter_info->plugin_name); - const char *ok_or_err = (0 == rv) ? "success" : "fail"; - nlog_info("%s load adapter %s type:%" PRId64 ", name:%s plugin:%s", - adapter->name, ok_or_err, adapter_info->type, - adapter_info->name, adapter_info->plugin_name); - if (0 != rv) { - continue; - } - - persister_singleton_load_setting(adapter, adapter_info->name); - persister_singleton_load_grp_and_tags(adapter, adapter_info->name); - - if (ADAPTER_STATE_RUNNING == adapter_info->state) { - rv = neu_manager_node_ctl(adapter->manager, adapter_info->name, - NEU_ADAPTER_CTL_START); - - if (0 != rv) { - nlog_error("%s fail start adapter %s", adapter->name, - adapter_info->name); - } - } - } - - utarray_foreach(adapter_infos, neu_persist_adapter_info_t *, adapter_info) - { - persister_singleton_load_subscriptions(adapter, adapter_info->name); - } - - neu_persist_adapter_infos_free(adapter_infos); -error_load_adapters: - return rv; -} - -static int persister_singleton_handle_nodes(neu_adapter_t * adapter, - neu_reqresp_type_e event, - const char * node_name) -{ - int rv = 0; - neu_persister_t * persister = persister_singleton_get(); - neu_persist_adapter_info_t adapter_info = {}; - - nlog_info("%s handling node event %d of %s", adapter->name, event, - node_name); - - if (NEU_REQ_DEL_NODE == event) { - rv = neu_persister_delete_adapter(persister, node_name); - if (0 != rv) { - nlog_error("%s failed to del adapter %s", adapter->name, node_name); - } - return rv; - } - - // MSG_EVENT_ADD_NODE - rv = neu_manager_get_adapter_info(adapter->manager, node_name, - &adapter_info); - if (0 != rv) { - nlog_error("%s unable to get adapter:%s info", adapter->name, - node_name); - return rv; - } - - rv = neu_persister_store_adapter(persister, &adapter_info); - if (0 != rv) { - nlog_error("%s failed to store adapter info", adapter->name); - } - - neu_persist_adapter_info_fini(&adapter_info); - return rv; -} - -static int persister_singleton_handle_plugins(neu_adapter_t *adapter) -{ - neu_persister_t *persister = persister_singleton_get(); - UT_array * plugin_infos = NULL; - - nlog_info("%s handling plugin event", adapter->name); - - plugin_infos = neu_manager_get_plugins(adapter->manager); - - // store the current set of plugin infos - int rv = neu_persister_store_plugins(persister, plugin_infos); - if (0 != rv) { - nlog_error("%s failed to store plugins infos", adapter->name); - } - - neu_persist_plugin_infos_free(plugin_infos); - return rv; -} - -static int persister_singleton_handle_grp_config(neu_adapter_t * adapter, - neu_reqresp_type_e event, - char *node_name, char *group, - uint32_t interval) -{ - neu_persister_t *persister = persister_singleton_get(); - int rv = 0; - - nlog_info("%s handling grp config event %d", adapter->name, event); - - if (NEU_REQ_DEL_GROUP != event) { - neu_persist_group_config_info_t info = { - .group_config_name = group, - .read_interval = interval, - .adapter_name = node_name, - }; - rv = neu_persister_store_group_config(persister, node_name, &info); - if (0 != rv) { - nlog_error("%s failed to store group config infos", node_name); - } - } else { - rv = neu_persister_delete_group_config(persister, node_name, group); - if (0 != rv) { - nlog_error("%s failed to del group config infos", group); - } - } - - return rv; -} - -static int persister_singleton_handle_datatags(neu_adapter_t *adapter, - const char * node, - const char * group) -{ - neu_persister_t *persister = persister_singleton_get(); - UT_array * tags = NULL; - - int rv = neu_manager_get_tag(adapter->manager, node, group, &tags); - if (rv != 0) { - nlog_warn("%s fail get tag infos", adapter->name); - return rv; - } - - rv = neu_persister_store_datatags(persister, node, group, tags); - if (0 != rv) { - nlog_error("fail store adapter:%s grp:%s datatag infos", node, group); - } - - neu_persist_datatag_infos_free(tags); - return rv; -} - -static int persister_singleton_handle_subscriptions(neu_adapter_t *adapter, - const char * app) -{ - neu_persister_t *persister = persister_singleton_get(); - - UT_array *sub_infos = neu_manager_get_sub_group(adapter->manager, app); - - int rv = neu_persister_store_subscriptions(persister, app, sub_infos); - if (0 != rv) { - nlog_error("%s fail store adapter:%s subscription infos", app, app); - } - - utarray_free(sub_infos); - - return rv; -} - -neu_node_running_state_e -neu_adapter_state_to_plugin_state(neu_adapter_t *adapter) -{ - neu_node_running_state_e state; - - switch (adapter->state) { - case ADAPTER_STATE_IDLE: - state = NEU_NODE_RUNNING_STATE_IDLE; + switch (info->module->type) { + case NEU_NA_TYPE_DRIVER: + adapter = (neu_adapter_t *) neu_adapter_driver_create(); break; - case ADAPTER_STATE_INIT: - state = NEU_NODE_RUNNING_STATE_INIT; - break; - case ADAPTER_STATE_READY: - state = NEU_NODE_RUNNING_STATE_READY; - break; - case ADAPTER_STATE_RUNNING: - state = NEU_NODE_RUNNING_STATE_RUNNING; - break; - case ADAPTER_STATE_STOPPED: - state = NEU_NODE_RUNNING_STATE_STOPPED; - break; - default: - state = NEU_NODE_RUNNING_STATE_IDLE; + case NEU_NA_TYPE_APP: + adapter = calloc(1, sizeof(neu_adapter_t)); break; } - return state; + adapter->name = strdup(info->name); + adapter->events = neu_event_new(); + adapter->state = NEU_NODE_RUNNING_STATE_INIT; + adapter->handle = info->handle; + adapter->cb_funs.command = callback_funs.command; + adapter->cb_funs.response = callback_funs.response; + adapter->module = info->module; + adapter->persister = neu_persister_create("persistence"); + + nng_pair1_open(&adapter->sock); + nng_socket_set_ms(adapter->sock, NNG_OPT_SENDTIMEO, 100); + nng_socket_get_int(adapter->sock, NNG_OPT_RECVFD, &adapter->recv_fd); + + switch (info->module->type) { + case NEU_NA_TYPE_DRIVER: + neu_adapter_driver_init((neu_adapter_driver_t *) adapter); + break; + case NEU_NA_TYPE_APP: + break; + } + + adapter->plugin = adapter->module->intf_funs->open(); + assert(adapter->plugin != NULL); + assert(neu_plugin_common_check(adapter->plugin)); + neu_plugin_common_t *common = neu_plugin_to_plugin_common(adapter->plugin); + common->adapter = adapter; + common->adapter_callbacks = &adapter->cb_funs; + common->link_state = NEU_NODE_LINK_STATE_DISCONNECTED; + common->log = zlog_get_category(adapter->name); + + adapter->module->intf_funs->init(adapter->plugin); + if (adapter_load_setting(adapter->persister, adapter->name, + &adapter->setting) == 0) { + if (adapter->module->intf_funs->setting(adapter->plugin, + adapter->setting) == 0) { + adapter->state = NEU_NODE_RUNNING_STATE_READY; + } else { + free(adapter->setting); + } + if (adapter->state == NEU_NODE_RUNNING_STATE_READY && start) { + adapter->module->intf_funs->start(adapter->plugin); + adapter->state = NEU_NODE_RUNNING_STATE_RUNNING; + } + } + + if (info->module->type == NEU_NA_TYPE_DRIVER) { + adapter_load_group_and_tag(adapter->persister, + (neu_adapter_driver_t *) adapter); + } + + param.fd = adapter->recv_fd; + param.usr_data = (void *) adapter; + param.cb = adapter_loop; + + adapter->nng_io = neu_event_add_io(adapter->events, param); + rv = nng_dial(adapter->sock, neu_manager_get_url(), &adapter->dialer, 0); + assert(rv == 0); + + nlog_info("Success to create adapter: %s", adapter->name); + + return adapter; +} + +void neu_adapter_init(neu_adapter_t *adapter) +{ + neu_reqresp_head_t header = { .type = NEU_REQ_NODE_INIT }; + neu_req_node_init_t init = { 0 }; + nng_msg * init_msg = NULL; + + strcpy(init.node, adapter->name); + init_msg = neu_msg_gen(&header, &init); + + nng_sendmsg(adapter->sock, init_msg, 0); } static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header, @@ -442,13 +156,36 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header, strcpy(header.receiver, cmd->driver); break; } + case NEU_REQ_GET_GROUP: + case NEU_REQ_DEL_GROUP: + case NEU_REQ_ADD_GROUP: { + neu_req_add_group_t *cmd = (neu_req_add_group_t *) data; + strcpy(header.receiver, cmd->driver); + break; + } + case NEU_REQ_GET_TAG: + case NEU_REQ_UPDATE_TAG: + case NEU_REQ_DEL_TAG: + case NEU_REQ_ADD_TAG: { + neu_req_add_tag_t *cmd = (neu_req_add_tag_t *) data; + strcpy(header.receiver, cmd->driver); + break; + } + case NEU_REQ_NODE_CTL: + case NEU_REQ_GET_NODE_STATE: + case NEU_REQ_GET_NODE_SETTING: + case NEU_REQ_NODE_SETTING: { + neu_req_node_setting_t *cmd = (neu_req_node_setting_t *) data; + strcpy(header.receiver, cmd->node); + break; + } default: break; } msg = neu_msg_gen(&header, data); - ret = nng_sendmsg(adapter->nng.sock, msg, 0); + ret = nng_sendmsg(adapter->sock, msg, 0); if (ret != 0) { nng_msg_free(msg); } @@ -470,68 +207,351 @@ static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header, nng_msg *msg = neu_msg_gen(header, data); - return nng_sendmsg(adapter->nng.sock, msg, 0); + return nng_sendmsg(adapter->sock, msg, 0); } -// clang-format off -static const adapter_callbacks_t callback_funs = { - .command = adapter_command, - .response = adapter_response, -}; -// clang-format on - -neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, - neu_manager_t * manager) +static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) { - int rv = 0; - neu_adapter_t *adapter = calloc(1, sizeof(neu_adapter_t)); + neu_adapter_t * adapter = (neu_adapter_t *) usr_data; + nng_msg * msg = NULL; + neu_reqresp_head_t *header = NULL; - adapter->cb_funs = callback_funs; - switch (info->module->type) { - case NEU_NA_TYPE_DRIVER: - adapter = (neu_adapter_t *) neu_adapter_driver_create(adapter); + if (type != NEU_EVENT_IO_READ) { + nlog_warn("adapter: %s recv close, exit loop, fd: %d", adapter->name, + fd); + return 0; + } + + nng_recvmsg(adapter->sock, &msg, 0); + header = (neu_reqresp_head_t *) nng_msg_body(msg); + + switch (header->type) { + case NEU_REQ_APP_SUBSCRIBE_GROUP: + case NEU_REQ_APP_UNSUBSCRIBE_GROUP: + case NEU_RESP_GET_NODE_STATE: + case NEU_RESP_GET_NODE_SETTING: + case NEU_RESP_READ_GROUP: + case NEU_RESP_GET_SUBSCRIBE_GROUP: + case NEU_RESP_ADD_TAG: + case NEU_RESP_UPDATE_TAG: + case NEU_RESP_GET_TAG: + case NEU_RESP_GET_NODE: + case NEU_RESP_GET_PLUGIN: + case NEU_RESP_GET_GROUP: + case NEU_RESP_ERROR: + case NEU_REQRESP_TRANS_DATA: + adapter->module->intf_funs->request( + adapter->plugin, (neu_reqresp_head_t *) header, &header[1]); break; - default: + case NEU_REQ_READ_GROUP: { + neu_resp_error_t error = { 0 }; + + if (adapter->module->type != NEU_NA_TYPE_DRIVER) { + error.error = NEU_ERR_GROUP_NOT_ALLOW; + header->type = NEU_RESP_ERROR; + neu_msg_exchange(header); + reply(adapter, header, &error); + } else { + neu_adapter_driver_read_group((neu_adapter_driver_t *) adapter, + header); + } + + break; + } + case NEU_REQ_WRITE_TAG: { + neu_resp_error_t error = { 0 }; + + if (adapter->module->type != NEU_NA_TYPE_DRIVER) { + error.error = NEU_ERR_GROUP_NOT_ALLOW; + header->type = NEU_RESP_ERROR; + neu_msg_exchange(header); + reply(adapter, header, &error); + } else { + neu_adapter_driver_write_tag((neu_adapter_driver_t *) adapter, + header); + } + + break; + } + case NEU_REQ_NODE_SETTING: { + neu_req_node_setting_t *cmd = (neu_req_node_setting_t *) &header[1]; + neu_resp_error_t error = { 0 }; + + error.error = neu_adapter_set_setting(adapter, cmd->setting); + if (error.error == NEU_ERR_SUCCESS) { + adapter_storage_setting(adapter->persister, adapter->name, + cmd->setting); + } + free(cmd->setting); + + header->type = NEU_RESP_ERROR; + neu_msg_exchange(header); + reply(adapter, header, &error); + break; + } + case NEU_REQ_GET_NODE_SETTING: { + neu_resp_get_node_setting_t resp = { 0 }; + neu_resp_error_t error = { 0 }; + + neu_msg_exchange(header); + error.error = neu_adapter_get_setting(adapter, &resp.setting); + if (error.error != NEU_ERR_SUCCESS) { + header->type = NEU_RESP_ERROR; + reply(adapter, header, &error); + } else { + header->type = NEU_RESP_GET_NODE_SETTING; + reply(adapter, header, &resp); + } + + break; + } + case NEU_REQ_GET_NODE_STATE: { + neu_resp_get_node_state_t resp = { 0 }; + + resp.state = neu_adapter_get_state(adapter); + header->type = NEU_RESP_GET_NODE_STATE; + neu_msg_exchange(header); + reply(adapter, header, &resp); + break; + } + case NEU_REQ_GET_GROUP: { + neu_msg_exchange(header); + + if (adapter->module->type != NEU_NA_TYPE_DRIVER) { + neu_resp_error_t error = { .error = NEU_ERR_GROUP_NOT_ALLOW }; + + header->type = NEU_RESP_ERROR; + reply(adapter, header, &error); + } else { + neu_resp_get_group_t resp = { + .groups = neu_adapter_driver_get_group( + (neu_adapter_driver_t *) adapter) + }; + header->type = NEU_RESP_GET_GROUP; + reply(adapter, header, &resp); + } + break; + } + case NEU_REQ_GET_TAG: { + neu_req_get_tag_t *cmd = (neu_req_get_tag_t *) &header[1]; + neu_resp_error_t error = { .error = 0 }; + UT_array * tags = NULL; + + if (adapter->module->type != NEU_NA_TYPE_DRIVER) { + error.error = NEU_ERR_GROUP_NOT_ALLOW; + } else { + error.error = neu_adapter_driver_get_tag( + (neu_adapter_driver_t *) adapter, cmd->group, &tags); + } + + neu_msg_exchange(header); + if (error.error != NEU_ERR_SUCCESS) { + header->type = NEU_RESP_ERROR; + reply(adapter, header, &error); + } else { + neu_resp_get_tag_t resp = { .tags = tags }; + + header->type = NEU_RESP_GET_TAG; + reply(adapter, header, &resp); + } + + break; + } + case NEU_REQ_ADD_GROUP: { + neu_req_add_group_t *cmd = (neu_req_add_group_t *) &header[1]; + neu_resp_error_t error = { 0 }; + + if (cmd->interval < NEU_GROUP_INTERVAL_LIMIT) { + error.error = NEU_ERR_GROUP_PARAMETER_INVALID; + } else { + if (adapter->module->type != NEU_NA_TYPE_DRIVER) { + error.error = NEU_ERR_GROUP_NOT_ALLOW; + } else { + error.error = neu_adapter_driver_add_group( + (neu_adapter_driver_t *) adapter, cmd->group, + cmd->interval); + } + } + + if (error.error == NEU_ERR_SUCCESS) { + adapter_storage_add_group(adapter->persister, adapter->name, + cmd->group, cmd->interval); + } + + neu_msg_exchange(header); + header->type = NEU_RESP_ERROR; + reply(adapter, header, &error); + break; + } + case NEU_REQ_DEL_GROUP: { + neu_req_del_group_t *cmd = (neu_req_del_group_t *) &header[1]; + neu_resp_error_t error = { 0 }; + + if (adapter->module->type != NEU_NA_TYPE_DRIVER) { + error.error = NEU_ERR_GROUP_NOT_ALLOW; + } else { + error.error = neu_adapter_driver_del_group( + (neu_adapter_driver_t *) adapter, cmd->group); + } + + if (error.error == NEU_ERR_SUCCESS) { + adapter_storage_del_group(adapter->persister, cmd->driver, + cmd->group); + } + + neu_msg_exchange(header); + header->type = NEU_RESP_ERROR; + reply(adapter, header, &error); + break; + } + case NEU_REQ_NODE_CTL: { + neu_req_node_ctl_t *cmd = (neu_req_node_ctl_t *) &header[1]; + neu_resp_error_t error = { 0 }; + + switch (cmd->ctl) { + case NEU_ADAPTER_CTL_START: + error.error = neu_adapter_start(adapter); + break; + case NEU_ADAPTER_CTL_STOP: + error.error = neu_adapter_stop(adapter); + break; + } + + neu_msg_exchange(header); + header->type = NEU_RESP_ERROR; + reply(adapter, header, &error); + break; + } + case NEU_REQ_DEL_TAG: { + neu_req_del_tag_t *cmd = (neu_req_del_tag_t *) &header[1]; + neu_resp_error_t error = { 0 }; + + if (adapter->module->type != NEU_NA_TYPE_DRIVER) { + error.error = NEU_ERR_GROUP_NOT_ALLOW; + } else { + for (int i = 0; i < cmd->n_tag; i++) { + neu_adapter_driver_del_tag((neu_adapter_driver_t *) adapter, + cmd->group, cmd->tags[i]); + } + } + + for (int i = 0; i < cmd->n_tag; i++) { + free(cmd->tags[i]); + } + free(cmd->tags); + + if (error.error == NEU_ERR_SUCCESS) { + adapter_storage_tag(adapter->persister, adapter, cmd->driver, + cmd->group); + } + + neu_msg_exchange(header); + header->type = NEU_RESP_ERROR; + reply(adapter, header, &error); + break; + } + case NEU_REQ_ADD_TAG: { + neu_req_add_tag_t *cmd = (neu_req_add_tag_t *) &header[1]; + neu_resp_add_tag_t resp = { 0 }; + + if (adapter->module->type != NEU_NA_TYPE_DRIVER) { + resp.error = NEU_ERR_GROUP_NOT_ALLOW; + } else { + for (int i = 0; i < cmd->n_tag; i++) { + int ret = + neu_adapter_driver_add_tag((neu_adapter_driver_t *) adapter, + cmd->group, &cmd->tags[i]); + if (ret == 0) { + resp.index += 1; + } else { + resp.error = ret; + break; + } + } + } + for (int i = 0; i < cmd->n_tag; i++) { + free(cmd->tags[i].address); + free(cmd->tags[i].name); + free(cmd->tags[i].description); + } + free(cmd->tags); + + if (resp.index > 0) { + adapter_storage_tag(adapter->persister, adapter, cmd->driver, + cmd->group); + } + + neu_msg_exchange(header); + header->type = NEU_RESP_ADD_TAG; + reply(adapter, header, &resp); + break; + } + case NEU_REQ_UPDATE_TAG: { + neu_req_update_tag_t *cmd = (neu_req_update_tag_t *) &header[1]; + neu_resp_update_tag_t resp = { 0 }; + + if (adapter->module->type != NEU_NA_TYPE_DRIVER) { + resp.error = NEU_ERR_GROUP_NOT_ALLOW; + } else { + for (int i = 0; i < cmd->n_tag; i++) { + int ret = neu_adapter_driver_update_tag( + (neu_adapter_driver_t *) adapter, cmd->group, + &cmd->tags[i]); + if (ret == 0) { + resp.index += 1; + } else { + resp.error = ret; + break; + } + } + } + for (int i = 0; i < cmd->n_tag; i++) { + free(cmd->tags[i].address); + free(cmd->tags[i].name); + free(cmd->tags[i].description); + } + free(cmd->tags); + + if (resp.index > 0) { + adapter_storage_tag(adapter->persister, adapter, cmd->driver, + cmd->group); + } + + neu_msg_exchange(header); + header->type = NEU_RESP_UPDATE_TAG; + reply(adapter, header, &resp); + break; + } + case NEU_REQ_NODE_UNINIT: { + neu_req_node_uninit_t *cmd = (neu_req_node_uninit_t *) &header[1]; + nng_msg * uninit_msg = NULL; + + neu_adapter_uninit(adapter); + + header->type = NEU_RESP_NODE_UNINIT; + neu_msg_exchange(header); + strcpy(cmd->node, adapter->name); + uninit_msg = neu_msg_gen(header, cmd); + + nng_sendmsg(adapter->sock, uninit_msg, 0); break; } - adapter->name = strdup(info->name); - adapter->state = ADAPTER_STATE_IDLE; - adapter->manager = manager; - adapter->plugin_info.handle = info->handle; - adapter->plugin_info.module = info->module; - adapter->req_id = 1; + default: + assert(false); + break; + } + nng_msg_free(msg); - rv = nng_mtx_alloc(&adapter->nng.mtx); - assert(rv == 0); - rv = nng_mtx_alloc(&adapter->nng.sub_grp_mtx); - assert(rv == 0); - - adapter->plugin_info.plugin = - adapter->plugin_info.module->intf_funs->open(); - assert(adapter->plugin_info.plugin != NULL); - assert(neu_plugin_common_check(adapter->plugin_info.plugin)); - neu_plugin_common_t *common = - neu_plugin_to_plugin_common(adapter->plugin_info.plugin); - common->adapter = adapter; - common->adapter_callbacks = &adapter->cb_funs; - common->link_state = NEU_NODE_LINK_STATE_DISCONNECTED; - common->log = zlog_get_category(adapter->name); - - adapter->events = neu_event_new(); - - nlog_info("Success to create adapter: %s", adapter->name); - - return adapter; + return 0; } void neu_adapter_destroy(neu_adapter_t *adapter) { - if (adapter->plugin_info.module->type == NEU_NA_TYPE_DRIVER) { - neu_adapter_driver_destroy((neu_adapter_driver_t *) adapter); - } + nng_dialer_close(adapter->dialer); + nng_close(adapter->sock); + neu_persister_destroy(adapter->persister); - adapter->plugin_info.module->intf_funs->close(adapter->plugin_info.plugin); if (adapter->name != NULL) { free(adapter->name); } @@ -539,223 +559,56 @@ void neu_adapter_destroy(neu_adapter_t *adapter) free(adapter->setting); } - nng_mtx_free(adapter->nng.mtx); - nng_mtx_free(adapter->nng.sub_grp_mtx); neu_event_close(adapter->events); free(adapter); } -int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) -{ - neu_adapter_t * adapter = (neu_adapter_t *) usr_data; - int rv = 0; - nng_msg * msg = NULL; - neu_reqresp_head_t *header = NULL; - - switch (type) { - case NEU_EVENT_IO_READ: - rv = nng_recvmsg(adapter->nng.sock, &msg, 0); - if (rv == NNG_ECLOSED) { - nlog_warn("nng socket closed."); - return 0; - } - header = (neu_reqresp_head_t *) nng_msg_body(msg); - switch (header->type) { - case NEU_REQRESP_PERSISTENCE_LOAD: - persister_singleton_load_data(adapter); - break; - case NEU_REQ_APP_SUBSCRIBE_GROUP: - case NEU_REQ_APP_UNSUBSCRIBE_GROUP: - case NEU_RESP_GET_NODE_STATE: - case NEU_RESP_GET_NODE_SETTING: - case NEU_REQRESP_TRANS_DATA: - case NEU_RESP_READ_GROUP: - case NEU_RESP_GET_SUBSCRIBE_GROUP: - case NEU_RESP_ADD_TAG: - case NEU_RESP_UPDATE_TAG: - case NEU_RESP_GET_TAG: - case NEU_RESP_GET_NODE: - case NEU_RESP_GET_PLUGIN: - case NEU_RESP_GET_GROUP: - case NEU_RESP_ERROR: - adapter->plugin_info.module->intf_funs->request( - adapter->plugin_info.plugin, (neu_reqresp_head_t *) header, - &header[1]); - break; - case NEU_REQ_ADD_PLUGIN: - case NEU_REQ_DEL_PLUGIN: - persister_singleton_handle_plugins(adapter); - break; - case NEU_REQ_ADD_NODE: - case NEU_REQ_DEL_NODE: { - neu_req_add_node_t *cmd = (neu_req_add_node_t *) &header[1]; - persister_singleton_handle_nodes(adapter, header->type, cmd->node); - break; - } - case NEU_REQ_ADD_GROUP: { - neu_req_add_group_t *cmd = (neu_req_add_group_t *) &header[1]; - persister_singleton_handle_grp_config( - adapter, header->type, cmd->driver, cmd->group, cmd->interval); - break; - } - case NEU_REQ_DEL_GROUP: { - neu_req_del_group_t *cmd = (neu_req_del_group_t *) &header[1]; - persister_singleton_handle_grp_config(adapter, header->type, - cmd->driver, cmd->group, 0); - break; - } - case NEU_REQ_ADD_TAG: - case NEU_REQ_DEL_TAG: - case NEU_REQ_UPDATE_TAG: { - neu_req_add_tag_t *cmd = (neu_req_add_tag_t *) &header[1]; - persister_singleton_handle_datatags(adapter, cmd->driver, - cmd->group); - break; - } - case NEU_REQ_SUBSCRIBE_GROUP: - case NEU_REQ_UNSUBSCRIBE_GROUP: { - neu_req_subscribe_t *cmd = (neu_req_subscribe_t *) &header[1]; - persister_singleton_handle_subscriptions(adapter, cmd->app); - - break; - } - case NEU_REQ_NODE_SETTING: { - neu_req_node_setting_t *cmd = (neu_req_node_setting_t *) &header[1]; - neu_persister_t * persister = persister_singleton_get(); - - neu_persister_store_adapter_setting(persister, cmd->node, - cmd->setting); - - free(cmd->setting); - break; - } - case NEU_REQ_NODE_CTL: { - neu_req_node_ctl_t *cmd = (neu_req_node_ctl_t *) &header[1]; - persister_singleton_handle_nodes(adapter, NEU_REQ_ADD_NODE, - cmd->node); - break; - } - case NEU_REQ_READ_GROUP: { - assert(adapter->plugin_info.module->type == NEU_NA_TYPE_DRIVER); - neu_adapter_driver_read_group((neu_adapter_driver_t *) adapter, - header); - break; - } - case NEU_REQ_WRITE_TAG: { - assert(adapter->plugin_info.module->type == NEU_NA_TYPE_DRIVER); - neu_adapter_driver_write_tag((neu_adapter_driver_t *) adapter, - header); - break; - } - default: - assert(false); - break; - } - - nng_msg_free(msg); - break; - case NEU_EVENT_IO_CLOSED: - case NEU_EVENT_IO_HUP: - nlog_warn("adapter %s, fd: %d, recv closed or hup.", adapter->name, fd); - break; - } - - return 0; -} - -int neu_adapter_init(neu_adapter_t *adapter) -{ - int rv = nng_pair1_open(&adapter->nng.sock); - - neu_event_io_param_t param = { - .usr_data = (void *) adapter, - .cb = adapter_loop, - }; - - assert(rv == 0); - - nng_socket_set_ms(adapter->nng.sock, NNG_OPT_SENDTIMEO, 100); - nng_socket_get_int(adapter->nng.sock, NNG_OPT_RECVFD, &adapter->recv_fd); - param.fd = adapter->recv_fd; - - adapter->nng_io = neu_event_add_io(adapter->events, param); - - if (adapter->plugin_info.module->type == NEU_NA_TYPE_DRIVER) { - neu_adapter_driver_init((neu_adapter_driver_t *) adapter); - } - - rv = nng_dial(adapter->nng.sock, neu_manager_get_url(adapter->manager), - &adapter->nng.dialer, 0); - assert(rv == 0); - - neu_reqresp_head_t header = { 0 }; - neu_req_node_init_t init = { 0 }; - - header.type = NEU_REQ_NODE_INIT; - strcpy(init.node, adapter->name); - - nng_msg *out_msg = neu_msg_gen(&header, &init); - - nng_sendmsg(adapter->nng.sock, out_msg, 0); - - if (adapter->state == ADAPTER_STATE_IDLE) { - adapter->state = ADAPTER_STATE_INIT; - } - adapter->plugin_info.module->intf_funs->init(adapter->plugin_info.plugin); - return rv; -} - int neu_adapter_uninit(neu_adapter_t *adapter) { - int rv = 0; - - nng_dialer_close(adapter->nng.dialer); - nng_close(adapter->nng.sock); - - if (adapter->plugin_info.module->type == NEU_NA_TYPE_DRIVER) { + if (adapter->module->type == NEU_NA_TYPE_DRIVER) { neu_adapter_driver_uninit((neu_adapter_driver_t *) adapter); } - adapter->plugin_info.module->intf_funs->uninit(adapter->plugin_info.plugin); + adapter->module->intf_funs->uninit(adapter->plugin); neu_event_del_io(adapter->events, adapter->nng_io); + if (adapter->module->type == NEU_NA_TYPE_DRIVER) { + neu_adapter_driver_destroy((neu_adapter_driver_t *) adapter); + } + + adapter->module->intf_funs->close(adapter->plugin); + nlog_info("Stop the adapter(%s)", adapter->name); - return rv; + return 0; } int neu_adapter_start(neu_adapter_t *adapter) { - // TODO start by send msg - const neu_plugin_intf_funs_t *intf_funs = - adapter->plugin_info.module->intf_funs; - neu_err_code_e error = NEU_ERR_SUCCESS; + const neu_plugin_intf_funs_t *intf_funs = adapter->module->intf_funs; + neu_err_code_e error = NEU_ERR_SUCCESS; - nng_mtx_lock(adapter->nng.mtx); switch (adapter->state) { - case ADAPTER_STATE_IDLE: - case ADAPTER_STATE_INIT: - if (strcmp(adapter->plugin_info.module->module_name, "ekuiper") != 0) { + case NEU_NODE_RUNNING_STATE_IDLE: + case NEU_NODE_RUNNING_STATE_INIT: + if (strcmp(adapter->module->module_name, "ekuiper") != 0) { error = NEU_ERR_NODE_NOT_READY; } break; - case ADAPTER_STATE_RUNNING: + case NEU_NODE_RUNNING_STATE_RUNNING: error = NEU_ERR_NODE_IS_RUNNING; break; - case ADAPTER_STATE_READY: - case ADAPTER_STATE_STOPPED: + case NEU_NODE_RUNNING_STATE_READY: + case NEU_NODE_RUNNING_STATE_STOPPED: break; } - nng_mtx_unlock(adapter->nng.mtx); if (error != NEU_ERR_SUCCESS) { return error; } - error = intf_funs->start(adapter->plugin_info.plugin); + error = intf_funs->start(adapter->plugin); if (error == NEU_ERR_SUCCESS) { - nng_mtx_lock(adapter->nng.mtx); - adapter->state = ADAPTER_STATE_RUNNING; - nng_mtx_unlock(adapter->nng.mtx); + adapter->state = NEU_NODE_RUNNING_STATE_RUNNING; } return error; @@ -763,84 +616,50 @@ int neu_adapter_start(neu_adapter_t *adapter) int neu_adapter_stop(neu_adapter_t *adapter) { - // TODO stop by send msg + const neu_plugin_intf_funs_t *intf_funs = adapter->module->intf_funs; + neu_err_code_e error = NEU_ERR_SUCCESS; - const neu_plugin_intf_funs_t *intf_funs = - adapter->plugin_info.module->intf_funs; - neu_err_code_e error = NEU_ERR_SUCCESS; - - nng_mtx_lock(adapter->nng.mtx); switch (adapter->state) { - case ADAPTER_STATE_IDLE: - case ADAPTER_STATE_INIT: - case ADAPTER_STATE_READY: + case NEU_NODE_RUNNING_STATE_IDLE: + case NEU_NODE_RUNNING_STATE_INIT: + case NEU_NODE_RUNNING_STATE_READY: error = NEU_ERR_NODE_NOT_RUNNING; break; - case ADAPTER_STATE_STOPPED: + case NEU_NODE_RUNNING_STATE_STOPPED: error = NEU_ERR_NODE_IS_STOPED; break; - case ADAPTER_STATE_RUNNING: + case NEU_NODE_RUNNING_STATE_RUNNING: break; } - nng_mtx_unlock(adapter->nng.mtx); if (error != NEU_ERR_SUCCESS) { return error; } - error = intf_funs->stop(adapter->plugin_info.plugin); + error = intf_funs->stop(adapter->plugin); if (error == NEU_ERR_SUCCESS) { - nng_mtx_lock(adapter->nng.mtx); - adapter->state = ADAPTER_STATE_STOPPED; - nng_mtx_unlock(adapter->nng.mtx); + adapter->state = NEU_NODE_RUNNING_STATE_STOPPED; } return error; } -const char *neu_adapter_get_name(neu_adapter_t *adapter) -{ - if (adapter == NULL) { - return NULL; - } - - return (const char *) adapter->name; -} - -neu_manager_t *neu_adapter_get_manager(neu_adapter_t *adapter) -{ - if (adapter == NULL) { - return NULL; - } - - return (neu_manager_t *) adapter->manager; -} - -neu_adapter_type_e neu_adapter_get_type(neu_adapter_t *adapter) -{ - if (adapter == NULL) { - return 0; - } - - return (neu_adapter_type_e) adapter->plugin_info.module->type; -} - int neu_adapter_set_setting(neu_adapter_t *adapter, const char *setting) { int rv = -1; const neu_plugin_intf_funs_t *intf_funs; - intf_funs = adapter->plugin_info.module->intf_funs; - rv = intf_funs->setting(adapter->plugin_info.plugin, setting); + intf_funs = adapter->module->intf_funs; + rv = intf_funs->setting(adapter->plugin, setting); if (rv == 0) { if (adapter->setting != NULL) { free(adapter->setting); } adapter->setting = strdup(setting); - if (adapter->state == ADAPTER_STATE_INIT) { - adapter->state = ADAPTER_STATE_READY; + if (adapter->state == NEU_NODE_RUNNING_STATE_INIT) { + adapter->state = NEU_NODE_RUNNING_STATE_READY; } } else { rv = NEU_ERR_NODE_SETTING_INVALID; @@ -851,11 +670,6 @@ int neu_adapter_set_setting(neu_adapter_t *adapter, const char *setting) int neu_adapter_get_setting(neu_adapter_t *adapter, char **config) { - if (adapter == NULL) { - nlog_error("Get Config adapter with NULL adapter"); - return NEU_ERR_NODE_NOT_EXIST; - } - if (adapter->setting != NULL) { *config = strdup(adapter->setting); return NEU_ERR_SUCCESS; @@ -866,23 +680,21 @@ int neu_adapter_get_setting(neu_adapter_t *adapter, char **config) neu_node_state_t neu_adapter_get_state(neu_adapter_t *adapter) { - neu_node_state_t state = { 0 }; - neu_plugin_common_t *common = - neu_plugin_to_plugin_common(adapter->plugin_info.plugin); + neu_node_state_t state = { 0 }; + neu_plugin_common_t *common = neu_plugin_to_plugin_common(adapter->plugin); state.link = common->link_state; - state.running = neu_adapter_state_to_plugin_state(adapter); + state.running = adapter->state; return state; } int neu_adapter_validate_tag(neu_adapter_t *adapter, neu_datatag_t *tag) { - const neu_plugin_intf_funs_t *intf_funs = - adapter->plugin_info.module->intf_funs; - neu_err_code_e error = NEU_ERR_SUCCESS; + const neu_plugin_intf_funs_t *intf_funs = adapter->module->intf_funs; + neu_err_code_e error = NEU_ERR_SUCCESS; - error = intf_funs->driver.validate_tag(adapter->plugin_info.plugin, tag); + error = intf_funs->driver.validate_tag(adapter->plugin, tag); return error; } @@ -897,3 +709,11 @@ void neu_adapter_del_timer(neu_adapter_t *adapter, neu_event_timer_t *timer) { neu_event_del_timer(adapter->events, timer); } + +inline static void reply(neu_adapter_t *adapter, neu_reqresp_head_t *header, + void *data) +{ + nng_msg *msg = neu_msg_gen(header, data); + + nng_sendmsg(adapter->sock, msg, 0); +} \ No newline at end of file diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h index a6f753a6..0736d339 100644 --- a/src/adapter/adapter_internal.h +++ b/src/adapter/adapter_internal.h @@ -30,62 +30,51 @@ #include "adapter_info.h" #include "core/manager.h" -typedef enum adapter_state { - ADAPTER_STATE_IDLE = 0, - ADAPTER_STATE_INIT, - ADAPTER_STATE_READY, - ADAPTER_STATE_RUNNING, - ADAPTER_STATE_STOPPED, -} adapter_state_e; - struct neu_adapter { - adapter_state_e state; - char * name; - neu_manager_t * manager; + char *name; + char *setting; - uint32_t req_id; + neu_node_running_state_e state; + neu_persister_t * persister; adapter_callbacks_t cb_funs; - char * setting; - struct { - void * handle; // handle of dynamic lib - neu_plugin_module_t *module; - neu_plugin_t * plugin; - } plugin_info; + void * handle; + neu_plugin_module_t *module; + neu_plugin_t * plugin; - struct { - nng_socket sock; - nng_dialer dialer; - nng_mtx * mtx; - nng_mtx * sub_grp_mtx; - } nng; + nng_socket sock; + nng_dialer dialer; neu_events_t * events; neu_event_io_t *nng_io; int recv_fd; }; -neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, - neu_manager_t * manager); -void neu_adapter_destroy(neu_adapter_t *adapter); -int neu_adapter_start(neu_adapter_t *adapter); -int neu_adapter_stop(neu_adapter_t *adapter); -int neu_adapter_init(neu_adapter_t *adapter); -int neu_adapter_uninit(neu_adapter_t *adapter); +typedef void (*adapter_handler)(neu_adapter_t * adapter, + neu_reqresp_head_t *header); +typedef struct adapter_msg_handler { + neu_reqresp_type_e type; + adapter_handler handler; +} adapter_msg_handler_t; + +neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool start); +void neu_adapter_init(neu_adapter_t *adapter); + +int neu_adapter_start(neu_adapter_t *adapter); +int neu_adapter_stop(neu_adapter_t *adapter); + +int neu_adapter_uninit(neu_adapter_t *adapter); +void neu_adapter_destroy(neu_adapter_t *adapter); neu_event_timer_t *neu_adapter_add_timer(neu_adapter_t * adapter, neu_event_timer_param_t param); void neu_adapter_del_timer(neu_adapter_t *adapter, neu_event_timer_t *timer); -const char * neu_adapter_get_name(neu_adapter_t *adapter); -neu_manager_t * neu_adapter_get_manager(neu_adapter_t *adapter); -neu_adapter_type_e neu_adapter_get_type(neu_adapter_t *adapter); int neu_adapter_set_setting(neu_adapter_t *adapter, const char *config); int neu_adapter_get_setting(neu_adapter_t *adapter, char **config); neu_node_state_t neu_adapter_get_state(neu_adapter_t *adapter); -neu_node_running_state_e - neu_adapter_state_to_plugin_state(neu_adapter_t *adapter); + int neu_adapter_validate_tag(neu_adapter_t *adapter, neu_datatag_t *tag); #endif diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 809a086c..2fa8f654 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -128,13 +128,10 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag, } } -neu_adapter_driver_t *neu_adapter_driver_create(neu_adapter_t *adapter) +neu_adapter_driver_t *neu_adapter_driver_create() { neu_adapter_driver_t *driver = calloc(1, sizeof(neu_adapter_driver_t)); - driver->adapter = *adapter; - free(adapter); - nng_mtx_alloc(&driver->mtx); driver->cache = neu_driver_cache_new(); driver->driver_events = neu_event_new(); @@ -265,9 +262,8 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver, break; } - driver->adapter.plugin_info.module->intf_funs->driver.write_tag( - driver->adapter.plugin_info.plugin, (void *) req, tag, - cmd->value.value); + driver->adapter.module->intf_funs->driver.write_tag( + driver->adapter.plugin, (void *) req, tag, cmd->value.value); free(tag->address); free(tag->name); free(tag->description); @@ -341,8 +337,8 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name) neu_group_destroy(find->group); free(find); - neu_plugin_to_plugin_common(driver->adapter.plugin_info.plugin) - ->tag_size -= tag_size; + neu_plugin_to_plugin_common(driver->adapter.plugin)->tag_size -= + tag_size; ret = NEU_ERR_SUCCESS; } @@ -406,8 +402,8 @@ int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group, int ret = NEU_ERR_SUCCESS; group_t *find = NULL; - ret = driver->adapter.plugin_info.module->intf_funs->driver.validate_tag( - driver->adapter.plugin_info.plugin, tag); + ret = driver->adapter.module->intf_funs->driver.validate_tag( + driver->adapter.plugin, tag); if (ret != NEU_ERR_SUCCESS) { return ret; } @@ -422,8 +418,7 @@ int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group, nng_mtx_unlock(driver->mtx); if (ret == NEU_ERR_SUCCESS) { - neu_plugin_to_plugin_common(driver->adapter.plugin_info.plugin) - ->tag_size += 1; + neu_plugin_to_plugin_common(driver->adapter.plugin)->tag_size += 1; } return ret; @@ -445,8 +440,7 @@ int neu_adapter_driver_del_tag(neu_adapter_driver_t *driver, const char *group, nng_mtx_unlock(driver->mtx); if (ret == NEU_ERR_SUCCESS) { - neu_plugin_to_plugin_common(driver->adapter.plugin_info.plugin) - ->tag_size -= 1; + neu_plugin_to_plugin_common(driver->adapter.plugin)->tag_size -= 1; } return ret; @@ -458,8 +452,8 @@ int neu_adapter_driver_update_tag(neu_adapter_driver_t *driver, int ret = NEU_ERR_SUCCESS; group_t *find = NULL; - ret = driver->adapter.plugin_info.module->intf_funs->driver.validate_tag( - driver->adapter.plugin_info.plugin, tag); + ret = driver->adapter.module->intf_funs->driver.validate_tag( + driver->adapter.plugin, tag); if (ret != NEU_ERR_SUCCESS) { return ret; } @@ -514,8 +508,7 @@ static int report_callback(void *usr_data) { group_t * group = (group_t *) usr_data; neu_reqresp_head_t header = { 0 }; - neu_node_running_state_e state = - neu_adapter_state_to_plugin_state((neu_adapter_t *) group->driver); + neu_node_running_state_e state = group->driver->adapter.state; if (state != NEU_NODE_RUNNING_STATE_RUNNING) { return 0; } @@ -531,7 +524,7 @@ static int report_callback(void *usr_data) sizeof(neu_reqresp_trans_data_t) + utarray_len(tags) * sizeof(neu_resp_tag_value_t)); - strcpy(data->driver, neu_adapter_get_name((neu_adapter_t *) group->driver)); + strcpy(data->driver, group->driver->adapter.name); strcpy(data->group, group->name); data->n_tag = utarray_len(tags); read_group(group->driver->cache, group->name, tags, @@ -580,8 +573,7 @@ static void group_change(void *arg, int64_t timestamp, UT_array *tags, static int read_callback(void *usr_data) { group_t * group = (group_t *) usr_data; - neu_node_running_state_e state = - neu_adapter_state_to_plugin_state((neu_adapter_t *) group->driver); + neu_node_running_state_e state = group->driver->adapter.state; if (state != NEU_NODE_RUNNING_STATE_RUNNING) { return 0; } @@ -589,9 +581,8 @@ static int read_callback(void *usr_data) nng_mtx_lock(group->driver->mtx); if (!neu_group_is_change(group->group, group->timestamp)) { if (group->grp.tags != NULL) { - group->driver->adapter.plugin_info.module->intf_funs->driver - .group_timer(group->driver->adapter.plugin_info.plugin, - &group->grp); + group->driver->adapter.module->intf_funs->driver.group_timer( + group->driver->adapter.plugin, &group->grp); } } else { neu_group_change_test(group->group, group->timestamp, (void *) group, diff --git a/src/adapter/driver/driver_internal.h b/src/adapter/driver/driver_internal.h index 6123aaf2..f166628b 100644 --- a/src/adapter/driver/driver_internal.h +++ b/src/adapter/driver/driver_internal.h @@ -22,7 +22,7 @@ #include "adapter.h" -neu_adapter_driver_t *neu_adapter_driver_create(neu_adapter_t *adapter); +neu_adapter_driver_t *neu_adapter_driver_create(); void neu_adapter_driver_destroy(neu_adapter_driver_t *driver); int neu_adapter_driver_start(neu_adapter_driver_t *driver); diff --git a/src/adapter/storage.c b/src/adapter/storage.c new file mode 100644 index 00000000..2fc8edab --- /dev/null +++ b/src/adapter/storage.c @@ -0,0 +1,120 @@ +/** + * 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 "utils/log.h" + +#include "adapter_internal.h" +#include "driver/driver_internal.h" +#include "storage.h" + +void adapter_storage_setting(neu_persister_t *persister, const char *node, + const char *setting) +{ + neu_persister_store_adapter_setting(persister, node, setting); +} + +void adapter_storage_add_group(neu_persister_t *persister, const char *node, + const char *group, uint32_t interval) +{ + neu_persist_group_config_info_t info = { + .group_config_name = (char *) group, + .read_interval = interval, + .adapter_name = (char *) node, + }; + + neu_persister_store_group_config(persister, node, &info); +} + +void adapter_storage_del_group(neu_persister_t *persister, const char *node, + const char *group) +{ + neu_persister_delete_group_config(persister, node, group); +} + +void adapter_storage_tag(neu_persister_t *persister, neu_adapter_t *adapter, + const char *node, const char *group) +{ + UT_array *tags = NULL; + + int rv = neu_adapter_driver_get_tag((neu_adapter_driver_t *) adapter, group, + &tags); + if (rv != 0) { + nlog_warn("fail get tag infos"); + return; + } + + rv = neu_persister_store_datatags(persister, node, group, tags); + if (0 != rv) { + nlog_error("fail store adapter:%s grp:%s datatag infos", node, group); + } + + neu_persist_datatag_infos_free(tags); +} + +int adapter_load_setting(neu_persister_t *persister, const char *node, + char **setting) +{ + int rv = neu_persister_load_adapter_setting(persister, node, + (const char **) setting); + if (0 != rv) { + nlog_info("load %s setting fail", node); + return -1; + } + + return 0; +} + +int adapter_load_group_and_tag(neu_persister_t * persister, + neu_adapter_driver_t *driver) +{ + UT_array * group_infos = NULL; + neu_adapter_t *adapter = (neu_adapter_t *) driver; + + int rv = neu_persister_load_group_configs(persister, adapter->name, + &group_infos); + if (0 != rv) { + nlog_warn("load %s group fail", adapter->name); + return rv; + } + + utarray_foreach(group_infos, neu_persist_group_config_info_t *, p) + { + UT_array *tags = NULL; + neu_adapter_driver_add_group(driver, p->group_config_name, + p->read_interval); + + rv = neu_persister_load_datatags(persister, adapter->name, + p->group_config_name, &tags); + if (0 != rv) { + nlog_warn("load %s:%s tags fail", adapter->name, + p->group_config_name); + continue; + } + + utarray_foreach(tags, neu_datatag_t *, tag) + { + neu_adapter_driver_add_tag(driver, p->group_config_name, tag); + } + + neu_persist_datatag_infos_free(tags); + } + + neu_persist_group_config_infos_free(group_infos); + return rv; +} \ No newline at end of file diff --git a/src/adapter/storage.h b/src/adapter/storage.h new file mode 100644 index 00000000..e09b0ea1 --- /dev/null +++ b/src/adapter/storage.h @@ -0,0 +1,41 @@ +/** + * 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 ADAPTER_STORAGE_H +#define ADAPTER_STORAGE_H + +#include "persist/persist.h" + +#include "adapter_internal.h" + +void adapter_storage_setting(neu_persister_t *persister, const char *node, + const char *setting); +void adapter_storage_add_group(neu_persister_t *persister, const char *node, + const char *group, uint32_t interval); +void adapter_storage_del_group(neu_persister_t *persister, const char *node, + const char *group); +void adapter_storage_tag(neu_persister_t *persister, neu_adapter_t *adapter, + const char *node, const char *group); + +int adapter_load_setting(neu_persister_t *persister, const char *node, + char **setting); +int adapter_load_group_and_tag(neu_persister_t * persister, + neu_adapter_driver_t *driver); + +#endif diff --git a/src/core/manager.c b/src/core/manager.c index beddcd56..be3c7aa5 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -17,6 +17,7 @@ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ #include +#include #include #include @@ -33,6 +34,7 @@ #include "node_manager.h" #include "plugin_manager.h" +#include "storage.h" #include "subscribe.h" #include "manager.h" @@ -40,7 +42,6 @@ // definition for adapter names #define DEFAULT_DASHBOARD_ADAPTER_NAME DEFAULT_DASHBOARD_PLUGIN_NAME -#define DEFAULT_PERSIST_ADAPTER_NAME DEFAULT_DUMMY_PLUGIN_NAME static const char *const url = "inproc://neu_manager"; @@ -52,7 +53,6 @@ inline static void forward_msg_dup(neu_manager_t *manager, nng_msg *msg, inline static void forward_msg(neu_manager_t *manager, nng_msg *msg, const char *ndoe); static void start_static_adapter(neu_manager_t *manager, const char *name); -static void stop_static_adapter(neu_manager_t *manager, const char *name); neu_manager_t *neu_manager_create() { @@ -67,7 +67,7 @@ neu_manager_t *neu_manager_create() manager->plugin_manager = neu_plugin_manager_create(); manager->node_manager = neu_node_manager_create(); manager->subscribe_manager = neu_subscribe_manager_create(); - manager->stop = false; + manager->persister = neu_persister_create("persistence"); nng_pair1_open_poly(&manager->socket); @@ -78,17 +78,41 @@ neu_manager_t *neu_manager_create() manager->loop = neu_event_add_io(manager->events, param); start_static_adapter(manager, DEFAULT_DASHBOARD_PLUGIN_NAME); - start_static_adapter(manager, DEFAULT_DUMMY_PLUGIN_NAME); + + if (manager_load_plugin(manager) != 0) { + nlog_warn("load plugin error"); + } else { + manager_load_node(manager); + manager_load_subscribe(manager); + } return manager; } void neu_manager_destroy(neu_manager_t *manager) { - stop_static_adapter(manager, DEFAULT_DASHBOARD_PLUGIN_NAME); - stop_static_adapter(manager, DEFAULT_DUMMY_PLUGIN_NAME); + neu_reqresp_head_t header = { .type = NEU_REQ_NODE_UNINIT }; + neu_req_node_init_t uninit = { 0 }; + nng_msg * uninit_msg = NULL; + UT_array *pipes = neu_node_manager_get_pipes_all(manager->node_manager); - manager->stop = true; + utarray_foreach(pipes, nng_pipe *, pipe) + { + uninit_msg = neu_msg_gen(&header, &uninit); + nng_msg_set_pipe(uninit_msg, *pipe); + + nng_sendmsg(manager->socket, uninit_msg, 0); + } + utarray_free(pipes); + + while (1) { + usleep(1000 * 100); + if (neu_node_manager_size(manager->node_manager) == 0) { + break; + } + } + + neu_persister_destroy(manager->persister); neu_subscribe_manager_destroy(manager->subscribe_manager); neu_node_manager_destroy(manager->node_manager); neu_plugin_manager_destroy(manager->plugin_manager); @@ -101,9 +125,8 @@ void neu_manager_destroy(neu_manager_t *manager) nlog_warn("manager exit"); } -const char *neu_manager_get_url(neu_manager_t *manager) +const char *neu_manager_get_url() { - (void) manager; return url; } @@ -127,44 +150,52 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) header = (neu_reqresp_head_t *) nng_msg_body(msg); - if (manager->stop) { - nlog_warn("manager stopped, drop from %s msg: %d", header->sender, - header->type); - nng_msg_free(msg); - return 0; - } - nlog_info("manager recv msg from: %s, type: %d", header->sender, header->type); switch (header->type) { + case NEU_REQRESP_TRANS_DATA: { + neu_reqresp_trans_data_t *cmd = (neu_reqresp_trans_data_t *) &header[1]; + UT_array *apps = neu_subscribe_manager_find(manager->subscribe_manager, + cmd->driver, cmd->group); + if (apps != NULL) { + utarray_foreach(apps, neu_app_subscribe_t *, app) + { + forward_msg_dup(manager, msg, app->pipe); + nlog_debug("forward trans data to pipe: %d", app->pipe.id); + } + utarray_free(apps); + break; + } + break; + } + case NEU_REQ_UPDATE_LICENSE: { + UT_array *pipes = neu_node_manager_get_pipes(manager->node_manager, + NEU_NA_TYPE_DRIVER); + + utarray_foreach(pipes, nng_pipe *, pipe) + { + forward_msg_dup(manager, msg, *pipe); + nlog_info("forward license update to pipe: %d", pipe->id); + } + utarray_free(pipes); + + break; + } case NEU_REQ_NODE_INIT: { neu_req_node_init_t *init = (neu_req_node_init_t *) &header[1]; nng_pipe pipe = nng_msg_get_pipe(msg); neu_node_manager_update(manager->node_manager, init->node, pipe); nlog_notice("bind node %s to pipe(%d)", init->node, pipe.id); - if (strcmp(DEFAULT_DUMMY_PLUGIN_NAME, init->node) == 0) { - header->type = NEU_REQRESP_PERSISTENCE_LOAD; - nng_msg *out_msg = neu_msg_gen(header, NULL); - - manager->persist_pipe = pipe; - nng_msg_set_pipe(out_msg, manager->persist_pipe); - - nng_sendmsg(manager->socket, out_msg, 0); - nlog_info("send persistence load to pipe: %d", pipe.id); - } break; } - case NEU_RESP_ERROR: - forward_msg(manager, msg, header->receiver); - break; case NEU_REQ_ADD_PLUGIN: { neu_req_add_plugin_t *cmd = (neu_req_add_plugin_t *) &header[1]; int error = neu_manager_add_plugin(manager, cmd->library); neu_resp_error_t e = { .error = error }; if (error == NEU_ERR_SUCCESS) { - forward_msg_dup(manager, msg, manager->persist_pipe); + manager_strorage_plugin(manager); } header->type = NEU_RESP_ERROR; @@ -177,6 +208,10 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) int error = neu_manager_del_plugin(manager, cmd->plugin); neu_resp_error_t e = { .error = error }; + if (error == NEU_ERR_SUCCESS) { + manager_strorage_plugin(manager); + } + header->type = NEU_RESP_ERROR; strcpy(header->receiver, header->sender); reply(manager, header, &e); @@ -197,7 +232,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_resp_error_t e = { .error = error }; if (error == NEU_ERR_SUCCESS) { - forward_msg_dup(manager, msg, manager->persist_pipe); + manager_storage_add_node(manager, cmd->node); } header->type = NEU_RESP_ERROR; @@ -207,16 +242,31 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) } case NEU_REQ_DEL_NODE: { neu_req_del_node_t *cmd = (neu_req_del_node_t *) &header[1]; - int error = neu_manager_del_node(manager, cmd->node); - neu_resp_error_t e = { .error = error }; + neu_resp_error_t error = { 0 }; - if (error == NEU_ERR_SUCCESS) { - forward_msg_dup(manager, msg, manager->persist_pipe); + strcpy(header->receiver, cmd->node); + if (neu_node_manager_find(manager->node_manager, header->receiver) != + NULL) { + header->type = NEU_REQ_NODE_UNINIT; + forward_msg(manager, msg, header->receiver); + } else { + error.error = NEU_ERR_NODE_NOT_EXIST; + header->type = NEU_RESP_ERROR; + neu_msg_exchange(header); + reply(manager, header, &error); } + break; + } + case NEU_RESP_NODE_UNINIT: { + neu_resp_node_uninit_t *cmd = (neu_resp_node_uninit_t *) &header[1]; - header->type = NEU_RESP_ERROR; - strcpy(header->receiver, header->sender); - reply(manager, header, &e); + neu_manager_del_node(manager, cmd->node); + manager_storage_del_node(manager, cmd->node); + if (strlen(header->receiver) > 0) { + neu_resp_error_t error = { 0 }; + header->type = NEU_RESP_ERROR; + reply(manager, header, &error); + } break; } case NEU_REQ_GET_NODE: { @@ -229,137 +279,6 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) reply(manager, header, &resp); break; } - case NEU_REQ_ADD_GROUP: { - neu_req_add_group_t *cmd = (neu_req_add_group_t *) &header[1]; - int error = neu_manager_add_group(manager, cmd->driver, cmd->group, - cmd->interval); - neu_resp_error_t e = { .error = error }; - - if (error == NEU_ERR_SUCCESS) { - forward_msg_dup(manager, msg, manager->persist_pipe); - } - - header->type = NEU_RESP_ERROR; - strcpy(header->receiver, header->sender); - reply(manager, header, &e); - break; - } - case NEU_REQ_DEL_GROUP: { - neu_req_del_group_t *cmd = (neu_req_del_group_t *) &header[1]; - int error = neu_manager_del_group(manager, cmd->driver, cmd->group); - neu_resp_error_t e = { .error = error }; - - if (error == NEU_ERR_SUCCESS) { - forward_msg_dup(manager, msg, manager->persist_pipe); - } - - header->type = NEU_RESP_ERROR; - strcpy(header->receiver, header->sender); - reply(manager, header, &e); - break; - } - case NEU_REQ_GET_GROUP: { - neu_req_get_group_t *cmd = (neu_req_get_group_t *) &header[1]; - UT_array * groups = NULL; - int ret = neu_manager_get_group(manager, cmd->driver, &groups); - neu_resp_error_t e = { .error = ret }; - neu_resp_get_group_t resp = { .groups = groups }; - - strcpy(header->receiver, header->sender); - if (ret != NEU_ERR_SUCCESS) { - header->type = NEU_RESP_ERROR; - reply(manager, header, &e); - } else { - header->type = NEU_RESP_GET_GROUP; - reply(manager, header, &resp); - } - - break; - } - case NEU_REQ_ADD_TAG: { - neu_req_add_tag_t *cmd = (neu_req_add_tag_t *) &header[1]; - neu_resp_add_tag_t resp = { 0 }; - - resp.error = neu_manager_add_tag(manager, cmd->driver, cmd->group, - cmd->n_tag, cmd->tags, &resp.index); - - if (resp.error == NEU_ERR_SUCCESS) { - forward_msg_dup(manager, msg, manager->persist_pipe); - neu_manager_notify_app_sub_update(manager, cmd->driver, cmd->group); - } - for (int i = 0; i < cmd->n_tag; i++) { - free(cmd->tags[i].address); - free(cmd->tags[i].name); - free(cmd->tags[i].description); - } - free(cmd->tags); - - header->type = NEU_RESP_ADD_TAG; - strcpy(header->receiver, header->sender); - reply(manager, header, &resp); - break; - } - case NEU_REQ_DEL_TAG: { - neu_req_del_tag_t *cmd = (neu_req_del_tag_t *) &header[1]; - neu_resp_error_t error = { 0 }; - - error.error = neu_manager_del_tag(manager, cmd->driver, cmd->group, - cmd->n_tag, cmd->tags); - - if (error.error == NEU_ERR_SUCCESS) { - forward_msg_dup(manager, msg, manager->persist_pipe); - neu_manager_notify_app_sub_update(manager, cmd->driver, cmd->group); - } - for (int i = 0; i < cmd->n_tag; i++) { - free(cmd->tags[i]); - } - free(cmd->tags); - - header->type = NEU_RESP_ERROR; - strcpy(header->receiver, header->sender); - reply(manager, header, &error); - break; - } - case NEU_REQ_UPDATE_TAG: { - neu_req_update_tag_t *cmd = (neu_req_update_tag_t *) &header[1]; - neu_resp_update_tag_t resp = { 0 }; - - resp.error = neu_manager_update_tag(manager, cmd->driver, cmd->group, - cmd->n_tag, cmd->tags, &resp.index); - - if (resp.error == NEU_ERR_SUCCESS) { - forward_msg_dup(manager, msg, manager->persist_pipe); - neu_manager_notify_app_sub_update(manager, cmd->driver, cmd->group); - } - for (int i = 0; i < cmd->n_tag; i++) { - free(cmd->tags[i].address); - free(cmd->tags[i].name); - free(cmd->tags[i].description); - } - free(cmd->tags); - - header->type = NEU_RESP_UPDATE_TAG; - strcpy(header->receiver, header->sender); - reply(manager, header, &resp); - break; - } - case NEU_REQ_GET_TAG: { - neu_req_get_tag_t *cmd = (neu_req_get_tag_t *) &header[1]; - UT_array * tags = NULL; - int ret = neu_manager_get_tag(manager, cmd->driver, cmd->group, &tags); - neu_resp_error_t e = { .error = ret }; - neu_resp_get_tag_t resp = { .tags = tags }; - - strcpy(header->receiver, header->sender); - if (ret != NEU_ERR_SUCCESS) { - header->type = NEU_RESP_ERROR; - reply(manager, header, &e); - } else { - header->type = NEU_RESP_GET_TAG; - reply(manager, header, &resp); - } - break; - } case NEU_REQ_SUBSCRIBE_GROUP: { neu_req_subscribe_t *cmd = (neu_req_subscribe_t *) &header[1]; neu_resp_error_t error = { 0 }; @@ -368,7 +287,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_manager_subscribe(manager, cmd->app, cmd->driver, cmd->group); if (error.error == NEU_ERR_SUCCESS) { - forward_msg_dup(manager, msg, manager->persist_pipe); + manager_storage_subscribe(manager, cmd->app); neu_manager_notify_app_sub(manager, cmd->app, cmd->driver, cmd->group); } @@ -386,7 +305,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) neu_manager_unsubscribe(manager, cmd->app, cmd->driver, cmd->group); if (error.error == NEU_ERR_SUCCESS) { - forward_msg_dup(manager, msg, manager->persist_pipe); + manager_storage_subscribe(manager, cmd->app); neu_manager_notify_app_unsub(manager, cmd->app, cmd->driver, cmd->group); } @@ -407,109 +326,111 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) reply(manager, header, &resp); break; } - case NEU_REQ_NODE_SETTING: { - neu_req_node_setting_t *cmd = (neu_req_node_setting_t *) &header[1]; - neu_resp_error_t error = { 0 }; - error.error = - neu_manager_node_setting(manager, cmd->node, cmd->setting); - - if (error.error == NEU_ERR_SUCCESS) { - forward_msg_dup(manager, msg, manager->persist_pipe); - } else { - free(cmd->setting); - } - - header->type = NEU_RESP_ERROR; - strcpy(header->receiver, header->sender); - reply(manager, header, &error); - break; - } - case NEU_REQ_GET_NODE_SETTING: { - neu_req_get_node_setting_t *cmd = - (neu_req_get_node_setting_t *) &header[1]; - neu_resp_get_node_setting_t resp = { 0 }; - neu_resp_error_t error = { 0 }; - - error.error = - neu_manager_node_get_setting(manager, cmd->ndoe, &resp.setting); - - strcpy(header->receiver, header->sender); - if (error.error != NEU_ERR_SUCCESS) { - header->type = NEU_RESP_ERROR; - reply(manager, header, &error); - } else { - header->type = NEU_RESP_GET_NODE_SETTING; - reply(manager, header, &resp); - } - break; - } - case NEU_REQ_GET_NODE_STATE: { - neu_req_get_node_state_t *cmd = (neu_req_get_node_state_t *) &header[1]; - neu_resp_get_node_state_t resp = { 0 }; - neu_resp_error_t error = { 0 }; - - error.error = - neu_manager_get_node_state(manager, cmd->node, &resp.state); - - strcpy(header->receiver, header->sender); - if (error.error != NEU_ERR_SUCCESS) { - header->type = NEU_RESP_ERROR; - reply(manager, header, &error); - } else { - header->type = NEU_RESP_GET_NODE_STATE; - reply(manager, header, &resp); - } - - break; - } - case NEU_REQ_NODE_CTL: { - neu_req_node_ctl_t *cmd = (neu_req_node_ctl_t *) &header[1]; - neu_resp_error_t error = { 0 }; - - error.error = neu_manager_node_ctl(manager, cmd->node, cmd->ctl); - if (error.error == NEU_ERR_SUCCESS) { - forward_msg_dup(manager, msg, manager->persist_pipe); - } - - strcpy(header->receiver, header->sender); - header->type = NEU_RESP_ERROR; - reply(manager, header, &error); - break; - } + case NEU_REQ_GET_NODE_SETTING: case NEU_REQ_READ_GROUP: case NEU_REQ_WRITE_TAG: + case NEU_REQ_GET_NODE_STATE: + case NEU_REQ_GET_GROUP: + case NEU_REQ_GET_TAG: + case NEU_REQ_NODE_CTL: + case NEU_REQ_ADD_GROUP: { + if (neu_node_manager_find(manager->node_manager, header->receiver) == + NULL) { + neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST }; + header->type = NEU_RESP_ERROR; + neu_msg_exchange(header); + reply(manager, header, &e); + } else { + forward_msg(manager, msg, header->receiver); + } + + break; + } + + case NEU_REQ_DEL_GROUP: { + neu_req_del_group_t *cmd = (neu_req_del_group_t *) &header[1]; + + if (neu_node_manager_find(manager->node_manager, header->receiver) == + NULL) { + neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST }; + header->type = NEU_RESP_ERROR; + neu_msg_exchange(header); + reply(manager, header, &e); + } else { + forward_msg(manager, msg, header->receiver); + neu_subscribe_manager_remove(manager->subscribe_manager, + cmd->driver, cmd->group); + } + break; + } + case NEU_REQ_DEL_TAG: { + neu_req_del_tag_t *cmd = (neu_req_del_tag_t *) &header[1]; + + if (neu_node_manager_find(manager->node_manager, header->receiver) == + NULL) { + neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST }; + header->type = NEU_RESP_ERROR; + neu_msg_exchange(header); + reply(manager, header, &e); + for (int i = 0; i < cmd->n_tag; i++) { + free(cmd->tags[i]); + } + free(cmd->tags); + } else { + forward_msg(manager, msg, header->receiver); + } + + break; + } + case NEU_REQ_UPDATE_TAG: + case NEU_REQ_ADD_TAG: { + neu_req_add_tag_t *cmd = (neu_req_add_tag_t *) &header[1]; + + if (neu_node_manager_find(manager->node_manager, header->receiver) == + NULL) { + neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST }; + header->type = NEU_RESP_ERROR; + neu_msg_exchange(header); + reply(manager, header, &e); + for (int i = 0; i < cmd->n_tag; i++) { + free(cmd->tags[i].address); + free(cmd->tags[i].name); + free(cmd->tags[i].description); + } + free(cmd->tags); + } else { + forward_msg(manager, msg, header->receiver); + } + + break; + } + 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) == + NULL) { + neu_resp_error_t e = { .error = NEU_ERR_NODE_NOT_EXIST }; + header->type = NEU_RESP_ERROR; + neu_msg_exchange(header); + reply(manager, header, &e); + free(cmd->setting); + } else { + forward_msg(manager, msg, header->receiver); + } + break; + } + + case NEU_RESP_ADD_TAG: + case NEU_RESP_UPDATE_TAG: + case NEU_RESP_GET_TAG: + case NEU_RESP_GET_GROUP: + case NEU_RESP_GET_NODE_SETTING: + case NEU_RESP_GET_NODE_STATE: + case NEU_RESP_ERROR: case NEU_RESP_READ_GROUP: forward_msg(manager, msg, header->receiver); break; - case NEU_REQ_UPDATE_LICENSE: { - UT_array *pipes = neu_node_manager_get_pipes(manager->node_manager, - NEU_NA_TYPE_DRIVER); - - utarray_foreach(pipes, nng_pipe *, pipe) - { - forward_msg_dup(manager, msg, *pipe); - nlog_info("forward license update to pipe: %d", pipe->id); - } - utarray_free(pipes); - - break; - } - case NEU_REQRESP_TRANS_DATA: { - neu_reqresp_trans_data_t *cmd = (neu_reqresp_trans_data_t *) &header[1]; - UT_array *apps = neu_subscribe_manager_find(manager->subscribe_manager, - cmd->driver, cmd->group); - if (apps != NULL) { - utarray_foreach(apps, neu_app_subscribe_t *, app) - { - forward_msg_dup(manager, msg, app->pipe); - nlog_debug("forward trans data to pipe: %d", app->pipe.id); - } - utarray_free(apps); - break; - } - break; - } default: assert(false); break; @@ -560,15 +481,6 @@ static void start_static_adapter(neu_manager_t *manager, const char *name) neu_adapter_start(adapter); } -static void stop_static_adapter(neu_manager_t *manager, const char *name) -{ - neu_adapter_t *adapter = neu_node_manager_find(manager->node_manager, name); - - neu_adapter_uninit(adapter); - neu_adapter_destroy(adapter); - neu_node_manager_del(manager->node_manager, name); -} - inline static void reply(neu_manager_t *manager, neu_reqresp_head_t *header, void *data) { diff --git a/src/core/manager.h b/src/core/manager.h index 7505a157..d2312bd5 100644 --- a/src/core/manager.h +++ b/src/core/manager.h @@ -29,6 +29,6 @@ neu_manager_t *neu_manager_create(); void neu_manager_destroy(neu_manager_t *manager); -const char *neu_manager_get_url(neu_manager_t *manager); +const char *neu_manager_get_url(void); #endif diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c index e9783f4e..18bf246c 100644 --- a/src/core/manager_internal.c +++ b/src/core/manager_internal.c @@ -66,7 +66,7 @@ int neu_manager_add_node(neu_manager_t *manager, const char *node_name, adapter_info.handle = instance.handle; adapter_info.module = instance.module; - adapter = neu_adapter_create(&adapter_info, manager); + adapter = neu_adapter_create(&adapter_info, true); neu_node_manager_add(manager->node_manager, adapter); neu_adapter_init(adapter); @@ -82,9 +82,7 @@ int neu_manager_del_node(neu_manager_t *manager, const char *node_name) return NEU_ERR_NODE_NOT_EXIST; } - neu_adapter_uninit(adapter); neu_adapter_destroy(adapter); - neu_subscribe_manager_remove(manager->subscribe_manager, node_name, NULL); neu_node_manager_del(manager->node_manager, node_name); return NEU_ERR_SUCCESS; @@ -95,180 +93,6 @@ UT_array *neu_manager_get_nodes(neu_manager_t *manager, neu_node_type_e type) return neu_node_manager_get(manager->node_manager, type); } -int neu_manager_add_group(neu_manager_t *manager, const char *driver, - const char *group, uint32_t interval) -{ - if (interval < NEU_GROUP_INTERVAL_LIMIT) { - return NEU_ERR_GROUP_PARAMETER_INVALID; - } - - if (strlen(group) > NEU_GROUP_NAME_LEN) { - return NEU_ERR_GROUP_PARAMETER_INVALID; - } - - neu_adapter_t *adapter = - neu_node_manager_find(manager->node_manager, driver); - if (adapter == NULL) { - return NEU_ERR_NODE_NOT_EXIST; - } - - if (adapter->plugin_info.module->type != NEU_NA_TYPE_DRIVER) { - return NEU_ERR_GROUP_NOT_ALLOW; - } - - return neu_adapter_driver_add_group((neu_adapter_driver_t *) adapter, group, - interval); -} - -int neu_manager_del_group(neu_manager_t *manager, const char *driver, - const char *group) -{ - neu_adapter_t *adapter = - neu_node_manager_find(manager->node_manager, driver); - if (adapter == NULL) { - return NEU_ERR_NODE_NOT_EXIST; - } - - if (adapter->plugin_info.module->type != NEU_NA_TYPE_DRIVER) { - return NEU_ERR_GROUP_NOT_ALLOW; - } - - neu_subscribe_manager_remove(manager->subscribe_manager, driver, group); - return neu_adapter_driver_del_group((neu_adapter_driver_t *) adapter, - group); -} - -int neu_manager_update_group(neu_manager_t *manager, const char *driver, - const char *group, uint32_t interval) -{ - if (interval < NEU_GROUP_INTERVAL_LIMIT) { - return NEU_ERR_GROUP_PARAMETER_INVALID; - } - (void) manager; - (void) driver; - (void) group; - (void) interval; - - return 0; -} - -int neu_manager_get_group(neu_manager_t *manager, const char *driver, - UT_array **groups) -{ - neu_adapter_t *adapter = - neu_node_manager_find(manager->node_manager, driver); - - if (adapter == NULL) { - return NEU_ERR_NODE_NOT_EXIST; - } - - if (adapter->plugin_info.module->type != NEU_NA_TYPE_DRIVER) { - return NEU_ERR_GROUP_NOT_ALLOW; - } - - *groups = neu_adapter_driver_get_group((neu_adapter_driver_t *) adapter); - - return NEU_ERR_SUCCESS; -} - -int neu_manager_add_tag(neu_manager_t *manager, const char *driver, - const char *group, uint16_t n_tag, neu_datatag_t *tags, - uint16_t *index) -{ - neu_adapter_t *adapter = - neu_node_manager_find(manager->node_manager, driver); - *index = 0; - - if (adapter == NULL) { - return NEU_ERR_NODE_NOT_EXIST; - } - - if (adapter->plugin_info.module->type != NEU_NA_TYPE_DRIVER) { - return NEU_ERR_GROUP_NOT_ALLOW; - } - - for (int i = 0; i < n_tag; i++) { - int ret = neu_adapter_driver_add_tag((neu_adapter_driver_t *) adapter, - group, &tags[i]); - if (ret == NEU_ERR_SUCCESS) { - *index += 1; - } else { - return ret; - } - } - - return NEU_ERR_SUCCESS; -} - -int neu_manager_del_tag(neu_manager_t *manager, const char *driver, - const char *group, uint16_t n_tag, char **tags) -{ - neu_adapter_t *adapter = - neu_node_manager_find(manager->node_manager, driver); - - if (adapter == NULL) { - return NEU_ERR_NODE_NOT_EXIST; - } - - if (adapter->plugin_info.module->type != NEU_NA_TYPE_DRIVER) { - return NEU_ERR_GROUP_NOT_ALLOW; - } - - for (int i = 0; i < n_tag; i++) { - neu_adapter_driver_del_tag((neu_adapter_driver_t *) adapter, group, - tags[i]); - } - - return NEU_ERR_SUCCESS; -} - -int neu_manager_update_tag(neu_manager_t *manager, const char *driver, - const char *group, uint16_t n_tag, - neu_datatag_t *tags, uint16_t *index) -{ - neu_adapter_t *adapter = - neu_node_manager_find(manager->node_manager, driver); - *index = 0; - - if (adapter == NULL) { - return NEU_ERR_NODE_NOT_EXIST; - } - - if (adapter->plugin_info.module->type != NEU_NA_TYPE_DRIVER) { - return NEU_ERR_GROUP_NOT_ALLOW; - } - - for (int i = 0; i < n_tag; i++) { - int ret = neu_adapter_driver_update_tag( - (neu_adapter_driver_t *) adapter, group, &tags[i]); - if (ret == NEU_ERR_SUCCESS) { - *index += 1; - } else { - return ret; - } - } - - return NEU_ERR_SUCCESS; -} - -int neu_manager_get_tag(neu_manager_t *manager, const char *driver, - const char *group, UT_array **tags) -{ - neu_adapter_t *adapter = - neu_node_manager_find(manager->node_manager, driver); - - if (adapter == NULL) { - return NEU_ERR_NODE_NOT_EXIST; - } - - if (adapter->plugin_info.module->type != NEU_NA_TYPE_DRIVER) { - return NEU_ERR_GROUP_NOT_ALLOW; - } - - return neu_adapter_driver_get_tag((neu_adapter_driver_t *) adapter, group, - tags); -} - int neu_manager_subscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group) { @@ -309,62 +133,6 @@ UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app) return neu_subscribe_manager_get(manager->subscribe_manager, app); } -int neu_manager_get_node_state(neu_manager_t *manager, const char *node, - neu_node_state_t *state) -{ - neu_adapter_t *adapter = neu_node_manager_find(manager->node_manager, node); - if (adapter == NULL) { - return NEU_ERR_NODE_NOT_EXIST; - } - - *state = neu_adapter_get_state(adapter); - - return NEU_ERR_SUCCESS; -} - -int neu_manager_node_ctl(neu_manager_t *manager, const char *node, - neu_adapter_ctl_e ctl) -{ - int ret = NEU_ERR_SUCCESS; - neu_adapter_t *adapter = neu_node_manager_find(manager->node_manager, node); - if (adapter == NULL) { - return NEU_ERR_NODE_NOT_EXIST; - } - - switch (ctl) { - case NEU_ADAPTER_CTL_START: - ret = neu_adapter_start(adapter); - break; - case NEU_ADAPTER_CTL_STOP: - ret = neu_adapter_stop(adapter); - break; - } - - return ret; -} - -int neu_manager_node_setting(neu_manager_t *manager, const char *node, - const char *setting) -{ - neu_adapter_t *adapter = neu_node_manager_find(manager->node_manager, node); - if (adapter == NULL) { - return NEU_ERR_NODE_NOT_EXIST; - } - - return neu_adapter_set_setting(adapter, setting); -} - -int neu_manager_node_get_setting(neu_manager_t *manager, const char *node, - char **setting) -{ - neu_adapter_t *adapter = neu_node_manager_find(manager->node_manager, node); - if (adapter == NULL) { - return NEU_ERR_NODE_NOT_EXIST; - } - - return neu_adapter_get_setting(adapter, setting); -} - int neu_manager_get_adapter_info(neu_manager_t *manager, const char *name, neu_persist_adapter_info_t *info) { @@ -372,8 +140,8 @@ int neu_manager_get_adapter_info(neu_manager_t *manager, const char *name, if (adapter != NULL) { info->name = strdup(name); - info->type = adapter->plugin_info.module->type; - info->plugin_name = strdup(adapter->plugin_info.module->module_name); + info->type = adapter->module->type; + info->plugin_name = strdup(adapter->module->module_name); info->state = adapter->state; return 0; } @@ -392,8 +160,8 @@ void neu_manager_notify_app_sub(neu_manager_t *manager, const char *app, header.type = NEU_REQ_APP_SUBSCRIBE_GROUP; strcpy(sub.driver, driver); strcpy(sub.group, group); - int ret = neu_manager_get_tag(manager, driver, group, &sub.tags); - assert(ret == 0); + // int ret = neu_manager_get_tag(manager, driver, group, &sub.tags); + // assert(ret == 0); msg = neu_msg_gen(&header, (void *) &sub); nng_msg_set_pipe(msg, pipe); diff --git a/src/core/manager_internal.h b/src/core/manager_internal.h index 6aca1b0a..b8c31a4c 100644 --- a/src/core/manager_internal.h +++ b/src/core/manager_internal.h @@ -24,6 +24,7 @@ #include #include "event/event.h" +#include "persist/persist.h" #include "node_manager.h" #include "plugin_manager.h" @@ -34,13 +35,11 @@ typedef struct neu_manager { neu_events_t * events; neu_event_io_t *loop; - bool stop; - neu_plugin_manager_t *plugin_manager; neu_node_manager_t * node_manager; neu_subscribe_mgr_t * subscribe_manager; - nng_pipe persist_pipe; + neu_persister_t *persister; } neu_manager_t; int neu_manager_add_plugin(neu_manager_t *manager, const char *library); @@ -52,42 +51,12 @@ int neu_manager_add_node(neu_manager_t *manager, const char *node_name, int neu_manager_del_node(neu_manager_t *manager, const char *node_name); UT_array *neu_manager_get_nodes(neu_manager_t *manager, neu_node_type_e type); -int neu_manager_add_group(neu_manager_t *manager, const char *driver, - const char *group, uint32_t interval); -int neu_manager_del_group(neu_manager_t *manager, const char *driver, - const char *group); -int neu_manager_update_group(neu_manager_t *manager, const char *driver, - const char *group, uint32_t interval); -int neu_manager_get_group(neu_manager_t *manager, const char *driver, - UT_array **groups); - -int neu_manager_add_tag(neu_manager_t *manager, const char *driver, - const char *group, uint16_t n_tag, neu_datatag_t *tags, - uint16_t *index); -int neu_manager_del_tag(neu_manager_t *manager, const char *driver, - const char *group, uint16_t n_tag, char **tags); -int neu_manager_update_tag(neu_manager_t *manager, const char *driver, - const char *group, uint16_t n_tag, - neu_datatag_t *tags, uint16_t *index); -int neu_manager_get_tag(neu_manager_t *manager, const char *driver, - const char *group, UT_array **tags); - int neu_manager_subscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group); int neu_manager_unsubscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group); UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app); -int neu_manager_get_node_state(neu_manager_t *manager, const char *node, - neu_node_state_t *state); -int neu_manager_node_ctl(neu_manager_t *manager, const char *node, - neu_adapter_ctl_e ctl); - -int neu_manager_node_setting(neu_manager_t *manager, const char *node, - const char *setting); -int neu_manager_node_get_setting(neu_manager_t *manager, const char *node, - char **setting); - int neu_manager_get_adapter_info(neu_manager_t *manager, const char *name, neu_persist_adapter_info_t *info); diff --git a/src/core/node_manager.c b/src/core/node_manager.c index 5b67f46c..7051592b 100644 --- a/src/core/node_manager.c +++ b/src/core/node_manager.c @@ -52,8 +52,6 @@ void neu_node_manager_destroy(neu_node_manager_t *mgr) HASH_ITER(hh, mgr->nodes, el, tmp) { HASH_DEL(mgr->nodes, el); - neu_adapter_uninit(el->adapter); - neu_adapter_destroy(el->adapter); free(el->name); free(el); } @@ -110,6 +108,11 @@ void neu_node_manager_del(neu_node_manager_t *mgr, const char *name) } } +uint16_t neu_node_manager_size(neu_node_manager_t *mgr) +{ + return HASH_COUNT(mgr->nodes); +} + UT_array *neu_node_manager_get(neu_node_manager_t *mgr, neu_node_type_e type) { UT_array * array = NULL; @@ -121,11 +124,10 @@ UT_array *neu_node_manager_get(neu_node_manager_t *mgr, neu_node_type_e type) HASH_ITER(hh, mgr->nodes, el, tmp) { if (!el->is_static) { - if (el->adapter->plugin_info.module->type == type) { + if (el->adapter->module->type == type) { neu_resp_node_info_t info = { 0 }; strcpy(info.node, el->adapter->name); - strcpy(info.plugin, - el->adapter->plugin_info.module->module_name); + strcpy(info.plugin, el->adapter->module->module_name); utarray_push_back(array, &info); } } @@ -159,7 +161,7 @@ UT_array *neu_node_manager_get_pipes(neu_node_manager_t *mgr, HASH_ITER(hh, mgr->nodes, el, tmp) { if (!el->is_static) { - if (el->adapter->plugin_info.module->type == type) { + if (el->adapter->module->type == type) { nng_pipe pipe = el->pipe; utarray_push_back(pipes, &pipe); } @@ -169,6 +171,23 @@ UT_array *neu_node_manager_get_pipes(neu_node_manager_t *mgr, return pipes; } +UT_array *neu_node_manager_get_pipes_all(neu_node_manager_t *mgr) +{ + UT_icd icd = { sizeof(nng_pipe), NULL, NULL, NULL }; + UT_array * pipes = NULL; + node_entity_t *el = NULL, *tmp = NULL; + + utarray_new(pipes, &icd); + + HASH_ITER(hh, mgr->nodes, el, tmp) + { + nng_pipe pipe = el->pipe; + utarray_push_back(pipes, &pipe); + } + + return pipes; +} + nng_pipe neu_node_manager_get_pipe(neu_node_manager_t *mgr, const char *name) { nng_pipe pipe = { 0 }; diff --git a/src/core/node_manager.h b/src/core/node_manager.h index c20dfd33..dcec1332 100644 --- a/src/core/node_manager.h +++ b/src/core/node_manager.h @@ -32,20 +32,22 @@ typedef struct neu_node_manager neu_node_manager_t; neu_node_manager_t *neu_node_manager_create(); void neu_node_manager_destroy(neu_node_manager_t *mgr); -int neu_node_manager_add(neu_node_manager_t *mgr, neu_adapter_t *adapter); -int neu_node_manager_add_static(neu_node_manager_t *mgr, - neu_adapter_t * adapter); -int neu_node_manager_update(neu_node_manager_t *mgr, const char *name, - nng_pipe pipe); -void neu_node_manager_del(neu_node_manager_t *mgr, const char *name); +int neu_node_manager_add(neu_node_manager_t *mgr, neu_adapter_t *adapter); +int neu_node_manager_add_static(neu_node_manager_t *mgr, + neu_adapter_t * adapter); +int neu_node_manager_update(neu_node_manager_t *mgr, const char *name, + nng_pipe pipe); +void neu_node_manager_del(neu_node_manager_t *mgr, const char *name); +uint16_t neu_node_manager_size(neu_node_manager_t *mgr); -// neu_node_info array +// neu_resp_node_info array UT_array *neu_node_manager_get(neu_node_manager_t *mgr, neu_node_type_e type); neu_adapter_t *neu_node_manager_find(neu_node_manager_t *mgr, const char *name); // nng_pipe array UT_array *neu_node_manager_get_pipes(neu_node_manager_t *mgr, neu_node_type_e 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); #endif \ No newline at end of file diff --git a/src/core/plugin_manager.c b/src/core/plugin_manager.c index 87b267c2..a197f759 100644 --- a/src/core/plugin_manager.c +++ b/src/core/plugin_manager.c @@ -27,7 +27,6 @@ #include "adapter.h" #include "plugin.h" -#include "dummy/dummy.h" #include "restful/rest.h" #include "plugin_manager.h" @@ -226,9 +225,6 @@ void neu_plugin_manager_load_static(neu_plugin_manager_t * mgr, instance->module = (neu_plugin_module_t *) &default_dashboard_plugin_module; } - if (strcmp(DEFAULT_DUMMY_PLUGIN_NAME, plugin_name) == 0) { - instance->module = (neu_plugin_module_t *) &default_dummy_plugin_module; - } } void neu_plugin_manager_destroy_instance(neu_plugin_manager_t * mgr, diff --git a/src/core/plugin_manager.h b/src/core/plugin_manager.h index 55e937ee..abd3c371 100644 --- a/src/core/plugin_manager.h +++ b/src/core/plugin_manager.h @@ -27,7 +27,6 @@ #include "plugin.h" #define DEFAULT_DASHBOARD_PLUGIN_NAME "default-dashboard" -#define DEFAULT_DUMMY_PLUGIN_NAME "default-dummy" typedef struct neu_plugin_instance { void * handle; diff --git a/src/core/storage.c b/src/core/storage.c new file mode 100644 index 00000000..49f6ce73 --- /dev/null +++ b/src/core/storage.c @@ -0,0 +1,155 @@ +/** + * 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 "utils/log.h" + +#include "storage.h" + +void manager_strorage_plugin(neu_manager_t *manager) +{ + UT_array *plugin_infos = NULL; + + plugin_infos = neu_manager_get_plugins(manager); + + int rv = neu_persister_store_plugins(manager->persister, plugin_infos); + if (0 != rv) { + nlog_error("failed to store plugins infos"); + } + + neu_persist_plugin_infos_free(plugin_infos); + return; +} + +void manager_storage_add_node(neu_manager_t *manager, const char *node) +{ + int rv = 0; + neu_persist_adapter_info_t adapter_info = {}; + + rv = neu_manager_get_adapter_info(manager, node, &adapter_info); + if (0 != rv) { + nlog_error("unable to get adapter:%s info", node); + return; + } + + rv = neu_persister_store_adapter(manager->persister, &adapter_info); + if (0 != rv) { + nlog_error("failed to store adapter info"); + } + + neu_persist_adapter_info_fini(&adapter_info); +} + +void manager_storage_del_node(neu_manager_t *manager, const char *node) +{ + neu_persister_delete_adapter(manager->persister, node); +} + +void manager_storage_subscribe(neu_manager_t *manager, const char *app) +{ + + UT_array *sub_infos = neu_manager_get_sub_group(manager, app); + + int rv = + neu_persister_store_subscriptions(manager->persister, app, sub_infos); + if (0 != rv) { + nlog_error("fail store adapter:%s subscription infos", app); + } + + utarray_free(sub_infos); +} + +int manager_load_plugin(neu_manager_t *manager) +{ + UT_array *plugin_infos = NULL; + + int rv = neu_persister_load_plugins(manager->persister, &plugin_infos); + if (rv != 0) { + return rv; + } + + utarray_foreach(plugin_infos, char **, name) + { + rv = neu_manager_add_plugin(manager, *name); + const char *ok_or_err = (0 == rv) ? "success" : "fail"; + nlog_info("load plugin %s, lib:%s", ok_or_err, *name); + } + + utarray_foreach(plugin_infos, char **, name) { free(*name); } + utarray_free(plugin_infos); + + return rv; +} + +int manager_load_node(neu_manager_t *manager) +{ + UT_array *adapter_infos = NULL; + int rv = 0; + + rv = neu_persister_load_adapters(manager->persister, &adapter_infos); + if (0 != rv) { + nlog_error("failed to load adapter infos"); + return -1; + } + + utarray_foreach(adapter_infos, neu_persist_adapter_info_t *, adapter_info) + { + rv = neu_manager_add_node(manager, adapter_info->name, + adapter_info->plugin_name); + const char *ok_or_err = (0 == rv) ? "success" : "fail"; + nlog_info("load adapter %s type:%" PRId64 ", name:%s plugin:%s", + ok_or_err, adapter_info->type, adapter_info->name, + adapter_info->plugin_name); + } + + neu_persist_adapter_infos_free(adapter_infos); + return rv; +} + +int manager_load_subscribe(neu_manager_t *manager) +{ + UT_array *nodes = + neu_node_manager_get(manager->node_manager, NEU_NA_TYPE_APP); + + utarray_foreach(nodes, neu_resp_node_info_t *, node) + { + int rv = 0; + UT_array *sub_infos = NULL; + + rv = neu_persister_load_subscriptions(manager->persister, node->node, + &sub_infos); + if (0 != rv) { + nlog_warn("load %s subscribetion infos error", node->node); + } else { + utarray_foreach(sub_infos, neu_persist_subscription_info_t *, info) + { + rv = neu_manager_subscribe(manager, info->sub_adapter_name, + info->src_adapter_name, + info->group_config_name); + const char *ok_or_err = (0 == rv) ? "success" : "fail"; + nlog_info("%s load subscription app:%s driver:%s grp:%s", + ok_or_err, info->sub_adapter_name, + info->src_adapter_name, info->group_config_name); + } + + neu_persist_subscription_infos_free(sub_infos); + } + } + + utarray_free(nodes); + return 0; +} \ No newline at end of file diff --git a/plugins/dummy/dummy.h b/src/core/storage.h similarity index 61% rename from plugins/dummy/dummy.h rename to src/core/storage.h index d26d7eb7..db8ecd65 100644 --- a/plugins/dummy/dummy.h +++ b/src/core/storage.h @@ -17,11 +17,20 @@ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ -#ifndef _NEU_PLUGIN_DUMMY_H_ -#define _NEU_PLUGIN_DUMMY_H_ +#ifndef _NEU_MANAGER_STORAGE_H_ +#define _NEU_MANAGER_STORAGE_H_ -#include "plugin.h" +#include "persist/persist.h" -extern const neu_plugin_module_t default_dummy_plugin_module; +#include "manager_internal.h" + +void manager_strorage_plugin(neu_manager_t *manager); +void manager_storage_del_node(neu_manager_t *manager, const char *node); +void manager_storage_add_node(neu_manager_t *manager, const char *node); +void manager_storage_subscribe(neu_manager_t *manager, const char *node); + +int manager_load_plugin(neu_manager_t *manager); +int manager_load_node(neu_manager_t *manager); +int manager_load_subscribe(neu_manager_t *manager); #endif