From de26f6518cd8f1572ae097870f9c00c6662913c5 Mon Sep 17 00:00:00 2001 From: fengzero Date: Wed, 17 Aug 2022 09:36:53 +0000 Subject: [PATCH 1/2] chore(adapter): add node_deleted and get_all_tags msg --- include/neuron/adapter.h | 32 ++++++++++++++++-- plugins/mqtt/mqtt_plugin.c | 2 ++ src/adapter/adapter.c | 28 ++++++++++++++++ src/adapter/adapter_internal.h | 2 ++ src/adapter/driver/driver.c | 25 ++++++++++++++ src/adapter/driver/driver_internal.h | 18 +++++----- src/core/manager.c | 50 ++++++++++++++++++++++++++-- src/core/plugin_manager.c | 1 + 8 files changed, 145 insertions(+), 13 deletions(-) diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 10e58264..525b4c14 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -47,6 +47,8 @@ typedef enum neu_reqresp_type { NEU_REQ_UNSUBSCRIBE_GROUP, NEU_REQ_GET_SUBSCRIBE_GROUP, NEU_RESP_GET_SUBSCRIBE_GROUP, + NEU_REQ_GET_SUB_DRIVER_TAGS, + NEU_RESP_GET_SUB_DRIVER_TAGS, NEU_REQ_APP_SUBSCRIBE_GROUP, NEU_RESP_APP_SUBSCRIBE_GROUP, @@ -89,6 +91,7 @@ typedef enum neu_reqresp_type { NEU_REQRESP_TRANS_DATA, NEU_REQRESP_NODES_STATE, + NEU_REQRESP_NODE_DELETED, } neu_reqresp_type_e; typedef enum { @@ -107,6 +110,8 @@ static const char *neu_reqresp_type_string_t[] = { [NEU_REQ_UNSUBSCRIBE_GROUP] = "NEU_REQ_UNSUBSCRIBE_GROUP", [NEU_REQ_GET_SUBSCRIBE_GROUP] = "NEU_REQ_GET_SUBSCRIBE_GROUP", [NEU_RESP_GET_SUBSCRIBE_GROUP] = "NEU_RESP_GET_SUBSCRIBE_GROUP", + [NEU_REQ_GET_SUB_DRIVER_TAGS] = "NEU_REQ_GET_SUB_DRIVER_TAGS", + [NEU_RESP_GET_SUB_DRIVER_TAGS] = "NEU_RESP_GET_SUB_DRIVER_TAGS", [NEU_REQ_APP_SUBSCRIBE_GROUP] = "NEU_REQ_APP_SUBSCRIBE_GROUP", [NEU_RESP_APP_SUBSCRIBE_GROUP] = "NEU_RESP_APP_SUBSCRIBE_GROUP", @@ -147,8 +152,9 @@ static const char *neu_reqresp_type_string_t[] = { [NEU_REQ_GET_PLUGIN] = "NEU_REQ_GET_PLUGIN", [NEU_RESP_GET_PLUGIN] = "NEU_RESP_GET_PLUGIN", - [NEU_REQRESP_TRANS_DATA] = "NEU_REQRESP_TRANS_DATA", - [NEU_REQRESP_NODES_STATE] = "NEU_REQRESP_NODE_STATE", + [NEU_REQRESP_TRANS_DATA] = "NEU_REQRESP_TRANS_DATA", + [NEU_REQRESP_NODES_STATE] = "NEU_REQRESP_NODES_STATE", + [NEU_REQRESP_NODE_DELETED] = "NEU_REQRESP_NODE_DELETED", }; inline static const char *neu_reqresp_type_string(neu_reqresp_type_e type) @@ -289,7 +295,7 @@ typedef struct { typedef struct neu_req_get_subscribe_group { char app[NEU_NODE_NAME_LEN]; -} neu_req_get_subscribe_group_t; +} neu_req_get_subscribe_group_t, neu_req_get_sub_driver_tags_t; typedef struct neu_resp_subscribe_info { char app[NEU_NODE_NAME_LEN]; @@ -297,6 +303,22 @@ typedef struct neu_resp_subscribe_info { char group[NEU_GROUP_NAME_LEN]; } neu_resp_subscribe_info_t; +typedef struct { + char driver[NEU_NODE_NAME_LEN]; + char group[NEU_GROUP_NAME_LEN]; + UT_array *tags; +} neu_resp_get_sub_driver_tags_info_t; +inline static UT_icd *neu_resp_get_sub_driver_tags_info_icd() +{ + static UT_icd icd = { sizeof(neu_resp_get_sub_driver_tags_info_t), NULL, + NULL, NULL }; + return &icd; +} + +typedef struct { + UT_array *infos; // array neu_resp_get_sub_driver_tags_info_t +} neu_resp_get_sub_driver_tags_t; + typedef struct neu_resp_get_subscribe_group { UT_array *groups; // array neu_resp_subscribe_info_t } neu_resp_get_subscribe_group_t; @@ -384,6 +406,10 @@ typedef struct { neu_resp_tag_value_t tags[]; } neu_reqresp_trans_data_t; +typedef struct { + char node[NEU_NODE_NAME_LEN]; +} neu_reqresp_node_deleted_t; + void *neu_msg_gen(neu_reqresp_head_t *header, void *data); inline static void neu_msg_exchange(neu_reqresp_head_t *header) diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index 37aa40a9..4c1ed15b 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -712,6 +712,8 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, utarray_free(states->states); break; } + case NEU_REQRESP_NODE_DELETED: + break; default: error = NEU_ERR_MQTT_FAILURE; break; diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 2c69a62d..62233ca2 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -157,6 +157,11 @@ void neu_adapter_init(neu_adapter_t *adapter) nng_sendmsg(adapter->sock, init_msg, 0); } +neu_node_type_e neu_adapter_get_type(neu_adapter_t *adapter) +{ + return adapter->module->type; +} + static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header, void *data) { @@ -171,6 +176,11 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header, strcpy(header.receiver, cmd->driver); break; } + case NEU_REQ_DEL_NODE: { + neu_req_del_node_t *cmd = (neu_req_del_node_t *) data; + strcpy(header.receiver, cmd->node); + break; + } case NEU_REQ_UPDATE_GROUP: case NEU_REQ_GET_GROUP: case NEU_REQ_DEL_GROUP: @@ -195,6 +205,15 @@ static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header, strcpy(header.receiver, cmd->node); break; } + case NEU_REQ_GET_SUB_DRIVER_TAGS: { + strcpy(header.receiver, "manager"); + break; + } + case NEU_REQRESP_NODE_DELETED: { + neu_reqresp_node_deleted_t *cmd = (neu_reqresp_node_deleted_t *) data; + strcpy(header.receiver, cmd->node); + break; + } default: break; } @@ -248,6 +267,8 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) header->sender, neu_reqresp_type_string(header->type)); switch (header->type) { + case NEU_REQRESP_NODE_DELETED: + case NEU_RESP_GET_SUB_DRIVER_TAGS: case NEU_REQ_UPDATE_LICENSE: case NEU_REQ_APP_UNSUBSCRIBE_GROUP: case NEU_RESP_GET_NODE_STATE: @@ -886,8 +907,12 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data) data_size = sizeof(neu_req_unsubscribe_t); break; case NEU_REQ_GET_SUBSCRIBE_GROUP: + case NEU_REQ_GET_SUB_DRIVER_TAGS: data_size = sizeof(neu_req_get_subscribe_group_t); break; + case NEU_RESP_GET_SUB_DRIVER_TAGS: + data_size = sizeof(neu_resp_get_sub_driver_tags_t); + break; case NEU_RESP_GET_SUBSCRIBE_GROUP: data_size = sizeof(neu_resp_get_subscribe_group_t); break; @@ -941,6 +966,9 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data) case NEU_REQ_APP_UNSUBSCRIBE_GROUP: data_size = sizeof(neu_req_app_unsubscribe_group_t); break; + case NEU_REQRESP_NODE_DELETED: + data_size = sizeof(neu_reqresp_node_deleted_t); + break; default: assert(false); break; diff --git a/src/adapter/adapter_internal.h b/src/adapter/adapter_internal.h index 8e75e5d5..efb1ef7b 100644 --- a/src/adapter/adapter_internal.h +++ b/src/adapter/adapter_internal.h @@ -67,6 +67,8 @@ void neu_adapter_init(neu_adapter_t *adapter); int neu_adapter_start(neu_adapter_t *adapter); int neu_adapter_stop(neu_adapter_t *adapter); +neu_node_type_e neu_adapter_get_type(neu_adapter_t *adapter); + int neu_adapter_uninit(neu_adapter_t *adapter); void neu_adapter_destroy(neu_adapter_t *adapter); diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 9052d779..4b24126d 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -553,6 +553,31 @@ int neu_adapter_driver_get_tag(neu_adapter_driver_t *driver, const char *group, return ret; } +void neu_adapter_driver_get_value_tag(neu_adapter_driver_t *driver, + const char *group, UT_array **tags) +{ + int ret = neu_adapter_driver_get_tag(driver, group, tags); + if (ret == NEU_ERR_SUCCESS) { + utarray_foreach(*tags, neu_datatag_t *, tag) + { + if (tag->decimal != 0) { + if (tag->type == NEU_TYPE_UINT8 || tag->type == NEU_TYPE_INT8 || + tag->type == NEU_TYPE_INT16 || + tag->type == NEU_TYPE_UINT16 || + tag->type == NEU_TYPE_INT32 || + tag->type == NEU_TYPE_UINT32 || + tag->type == NEU_TYPE_INT64 || + tag->type == NEU_TYPE_UINT64 || + tag->type == NEU_TYPE_FLOAT) { + tag->type = NEU_TYPE_DOUBLE; + } + } + } + } else { + utarray_new(*tags, neu_tag_get_icd()); + } +} + UT_array *neu_adapter_driver_get_read_tag(neu_adapter_driver_t *driver, const char * group) { diff --git a/src/adapter/driver/driver_internal.h b/src/adapter/driver/driver_internal.h index 46a10181..acd3affa 100644 --- a/src/adapter/driver/driver_internal.h +++ b/src/adapter/driver/driver_internal.h @@ -46,14 +46,16 @@ int neu_adapter_driver_group_exist(neu_adapter_driver_t *driver, const char * name); UT_array *neu_adapter_driver_get_group(neu_adapter_driver_t *driver); -int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group, - neu_datatag_t *tag); -int neu_adapter_driver_del_tag(neu_adapter_driver_t *driver, const char *group, - const char *tag); -int neu_adapter_driver_update_tag(neu_adapter_driver_t *driver, - const char *group, neu_datatag_t *tag); -int neu_adapter_driver_get_tag(neu_adapter_driver_t *driver, const char *group, - UT_array **tags); +int neu_adapter_driver_add_tag(neu_adapter_driver_t *driver, const char *group, + neu_datatag_t *tag); +int neu_adapter_driver_del_tag(neu_adapter_driver_t *driver, const char *group, + const char *tag); +int neu_adapter_driver_update_tag(neu_adapter_driver_t *driver, + const char *group, neu_datatag_t *tag); +int neu_adapter_driver_get_tag(neu_adapter_driver_t *driver, const char *group, + UT_array **tags); +void neu_adapter_driver_get_value_tag(neu_adapter_driver_t *driver, + const char *group, UT_array **tags); UT_array *neu_adapter_driver_get_read_tag(neu_adapter_driver_t *driver, const char * group); #endif diff --git a/src/core/manager.c b/src/core/manager.c index dfadb655..a10d9fd2 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -273,13 +273,30 @@ 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]; neu_resp_error_t error = { 0 }; + neu_adapter_t * adapter = + neu_node_manager_find(manager->node_manager, cmd->node); strcpy(header->receiver, cmd->node); - if (neu_node_manager_find(manager->node_manager, header->receiver) != - NULL) { + if (adapter != NULL) { manager_storage_del_node(manager, cmd->node); header->type = NEU_REQ_NODE_UNINIT; forward_msg(manager, msg, header->receiver); + if (neu_adapter_get_type(adapter) == NEU_NA_TYPE_DRIVER) { + UT_array *apps = neu_subscribe_manager_find_by_driver( + manager->subscribe_manager, cmd->node); + + utarray_foreach(apps, neu_app_subscribe_t *, app) + { + neu_reqresp_node_deleted_t resp = { 0 }; + header->type = NEU_REQRESP_NODE_DELETED; + + strcpy(resp.node, header->receiver); + strcpy(header->receiver, app->app_name); + strcpy(header->sender, "manager"); + reply(manager, header, &resp); + } + utarray_free(apps); + } } else { error.error = NEU_ERR_NODE_NOT_EXIST; header->type = NEU_RESP_ERROR; @@ -359,6 +376,35 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) reply(manager, header, &resp); break; } + case NEU_REQ_GET_SUB_DRIVER_TAGS: { + neu_req_get_sub_driver_tags_t *cmd = + (neu_req_get_sub_driver_tags_t *) &header[1]; + neu_resp_get_sub_driver_tags_t resp = { 0 }; + UT_array *groups = neu_manager_get_sub_group(manager, cmd->app); + + utarray_new(resp.infos, neu_resp_get_sub_driver_tags_info_icd()); + utarray_foreach(groups, neu_resp_subscribe_info_t *, info) + { + neu_resp_get_sub_driver_tags_info_t in = { 0 }; + neu_adapter_t * driver = + neu_node_manager_find(manager->node_manager, info->driver); + assert(driver != NULL); + + strcpy(in.driver, info->driver); + strcpy(in.group, info->group); + neu_adapter_driver_get_value_tag((neu_adapter_driver_t *) driver, + info->group, &in.tags); + + utarray_push_back(resp.infos, &in); + } + utarray_free(groups); + + strcpy(header->receiver, header->sender); + header->type = NEU_RESP_GET_SUB_DRIVER_TAGS; + reply(manager, header, &resp); + + break; + } case NEU_REQ_GET_NODES_STATE: { neu_resp_get_nodes_state_t resp = { 0 }; UT_array *states = neu_node_manager_get_state(manager->node_manager); diff --git a/src/core/plugin_manager.c b/src/core/plugin_manager.c index 0d567cc6..af60edf3 100644 --- a/src/core/plugin_manager.c +++ b/src/core/plugin_manager.c @@ -77,6 +77,7 @@ int neu_plugin_manager_add(neu_plugin_manager_t *mgr, neu_plugin_module_t *pm = NULL; plugin_entity_t * plugin = NULL; + assert(strlen(plugin_lib_name) <= NEU_PLUGIN_LIBRARY_LEN); snprintf(lib_path, sizeof(lib_path) - 1, "./plugins/%s", plugin_lib_name); handle = dlopen(lib_path, RTLD_NOW | RTLD_NODELETE); From 3ae3fb221e8af60eea896f9ef592d824ed7177b9 Mon Sep 17 00:00:00 2001 From: fengzero Date: Wed, 17 Aug 2022 10:20:44 +0000 Subject: [PATCH 2/2] fix(log): log size limit --- zlog.conf | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/zlog.conf b/zlog.conf index 591733ca..8af6bab4 100644 --- a/zlog.conf +++ b/zlog.conf @@ -9,4 +9,4 @@ simple = "%d [%V] %f:%L %m%n" [rules] -*.INFO "./logs/%c.log", 10MB * 3 ~ "./logs/%c-%d(%Y%m%d).#2s.log"; simple +*.INFO "./logs/%c.log", 10MB * 3 ~ "./logs/%c.#2s.log"; simple