diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 0172a4a1..1ae30c8f 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -101,6 +101,9 @@ typedef enum neu_reqresp_type { NEU_REQ_ADD_GROUP_EVENT, NEU_REQ_DEL_GROUP_EVENT, NEU_REQ_UPDATE_GROUP_EVENT, + NEU_REQ_ADD_TAG_EVENT, + NEU_REQ_DEL_TAG_EVENT, + NEU_REQ_UPDATE_TAG_EVENT, } neu_reqresp_type_e; static const char *neu_reqresp_type_string_t[] = { @@ -168,6 +171,9 @@ static const char *neu_reqresp_type_string_t[] = { [NEU_REQ_ADD_GROUP_EVENT] = "NEU_REQ_ADD_GROUP_EVENT", [NEU_REQ_DEL_GROUP_EVENT] = "NEU_REQ_DEL_GROUP_EVENT", [NEU_REQ_UPDATE_GROUP_EVENT] = "NEU_REQ_UPDATE_GROUP_EVENT", + [NEU_REQ_ADD_TAG_EVENT] = "NEU_REQ_ADD_TAG_EVENT", + [NEU_REQ_DEL_TAG_EVENT] = "NEU_REQ_DEL_TAG_EVENT", + [NEU_REQ_UPDATE_TAG_EVENT] = "NEU_REQ_UPDATE_TAG_EVENT", }; inline static const char *neu_reqresp_type_string(neu_reqresp_type_e type) diff --git a/plugins/monitor/monitor.c b/plugins/monitor/monitor.c index 4c158eeb..3a3c3ace 100644 --- a/plugins/monitor/monitor.c +++ b/plugins/monitor/monitor.c @@ -400,7 +400,10 @@ static int monitor_plugin_request(neu_plugin_t * plugin, case NEU_REQ_NODE_SETTING_EVENT: case NEU_REQ_ADD_GROUP_EVENT: case NEU_REQ_DEL_GROUP_EVENT: - case NEU_REQ_UPDATE_GROUP_EVENT: { + case NEU_REQ_UPDATE_GROUP_EVENT: + case NEU_REQ_ADD_TAG_EVENT: + case NEU_REQ_DEL_TAG_EVENT: + case NEU_REQ_UPDATE_TAG_EVENT: { handle_events(plugin, header->type, data); break; } diff --git a/plugins/monitor/monitor_config.c b/plugins/monitor/monitor_config.c index 14e0b7f6..700703d1 100644 --- a/plugins/monitor/monitor_config.c +++ b/plugins/monitor/monitor_config.c @@ -128,6 +128,9 @@ static int make_event_topics(monitor_config_t *config, const char *prefix) char *group_add_topic = NULL; char *group_del_topic = NULL; char *group_update_topic = NULL; + char *tag_add_topic = NULL; + char *tag_del_topic = NULL; + char *tag_update_topic = NULL; neu_asprintf(&node_add_topic, "%s/node/add", prefix); if (NULL == node_add_topic) { @@ -164,6 +167,21 @@ static int make_event_topics(monitor_config_t *config, const char *prefix) goto error; } + neu_asprintf(&tag_add_topic, "%s/tag/add", prefix); + if (NULL == tag_add_topic) { + goto error; + } + + neu_asprintf(&tag_del_topic, "%s/tag/delete", prefix); + if (NULL == tag_del_topic) { + goto error; + } + + neu_asprintf(&tag_update_topic, "%s/tag/update", prefix); + if (NULL == tag_update_topic) { + goto error; + } + config->node_add_topic = node_add_topic; config->node_del_topic = node_del_topic; config->node_ctl_topic = node_ctl_topic; @@ -171,10 +189,16 @@ static int make_event_topics(monitor_config_t *config, const char *prefix) config->group_add_topic = group_add_topic; config->group_del_topic = group_del_topic; config->group_update_topic = group_update_topic; + config->tag_add_topic = tag_add_topic; + config->tag_del_topic = tag_del_topic; + config->tag_update_topic = tag_update_topic; return 0; error: + free(tag_update_topic); + free(tag_del_topic); + free(tag_add_topic); free(group_update_topic); free(group_del_topic); free(group_add_topic); @@ -327,6 +351,10 @@ int monitor_config_parse(neu_plugin_t *plugin, const char *setting, plog_info(plugin, "group del topic : %s", config->group_del_topic); plog_info(plugin, "group update topic : %s", config->group_update_topic); + plog_info(plugin, "tag add topic : %s", config->tag_add_topic); + plog_info(plugin, "tag del topic : %s", config->tag_del_topic); + plog_info(plugin, "tag update topic : %s", + config->tag_update_topic); return 0; @@ -368,6 +396,9 @@ void monitor_config_fini(monitor_config_t *config) free(config->group_add_topic); free(config->group_del_topic); free(config->group_update_topic); + free(config->tag_add_topic); + free(config->tag_del_topic); + free(config->tag_update_topic); memset(config, 0, sizeof(*config)); } diff --git a/plugins/monitor/monitor_config.h b/plugins/monitor/monitor_config.h index 8ad5a245..7ed933af 100644 --- a/plugins/monitor/monitor_config.h +++ b/plugins/monitor/monitor_config.h @@ -51,6 +51,9 @@ typedef struct { char *group_add_topic; char *group_del_topic; char *group_update_topic; + char *tag_add_topic; + char *tag_del_topic; + char *tag_update_topic; } monitor_config_t; int monitor_config_parse(neu_plugin_t *plugin, const char *setting, diff --git a/plugins/monitor/mqtt_handle.c b/plugins/monitor/mqtt_handle.c index bd29cb5c..69c6e061 100644 --- a/plugins/monitor/mqtt_handle.c +++ b/plugins/monitor/mqtt_handle.c @@ -25,6 +25,7 @@ #include "parser/neu_json_group_config.h" #include "parser/neu_json_node.h" +#include "parser/neu_json_tag.h" #include "monitor.h" #include "mqtt_handle.h" @@ -62,8 +63,7 @@ static char *generate_heartbeat_json(neu_plugin_t *plugin, UT_array *states) static char *generate_event_json(neu_plugin_t *plugin, neu_reqresp_type_e event, void *data, char **topic_p) { - char *json_str = NULL; - int (*encode_fn)(void *json_object, void *param) = NULL; + char *json_str = NULL; union { neu_json_add_node_req_t add_node; neu_json_del_node_req_t del_node; @@ -72,38 +72,44 @@ static char *generate_event_json(neu_plugin_t *plugin, neu_reqresp_type_e event, neu_json_add_group_config_req_t add_grp; neu_json_del_group_config_req_t del_grp; neu_json_update_group_req_t update_grp; - } json_req; + neu_json_add_tags_req_t add_tags; + neu_json_del_tags_req_t del_tags; + } json_req = {}; switch (event) { case NEU_REQ_ADD_NODE_EVENT: { neu_req_add_node_t *add_node = data; json_req.add_node.name = add_node->node; json_req.add_node.plugin = add_node->plugin; - encode_fn = neu_json_encode_add_node_req; *topic_p = plugin->config->node_add_topic; + neu_json_encode_by_fn(&json_req, neu_json_encode_add_node_req, + &json_str); break; } case NEU_REQ_DEL_NODE_EVENT: { neu_req_del_node_t *del_node = data; json_req.del_node.name = del_node->node; - encode_fn = neu_json_encode_del_node_req; *topic_p = plugin->config->node_del_topic; + neu_json_encode_by_fn(&json_req, neu_json_encode_del_node_req, + &json_str); break; } case NEU_REQ_NODE_CTL_EVENT: { neu_req_node_ctl_t *node_ctl = data; json_req.node_ctl.node = node_ctl->node; json_req.node_ctl.cmd = node_ctl->ctl; - encode_fn = neu_json_encode_node_ctl_req; *topic_p = plugin->config->node_ctl_topic; + neu_json_encode_by_fn(&json_req, neu_json_encode_node_ctl_req, + &json_str); break; } case NEU_REQ_NODE_SETTING_EVENT: { neu_req_node_setting_t *node_setting = data; json_req.node_setting.node = node_setting->node; json_req.node_setting.setting = node_setting->setting; - encode_fn = neu_json_encode_node_setting_req; *topic_p = plugin->config->node_setting_topic; + neu_json_encode_by_fn(&json_req, neu_json_encode_node_setting_req, + &json_str); break; } case NEU_REQ_ADD_GROUP_EVENT: { @@ -111,16 +117,18 @@ static char *generate_event_json(neu_plugin_t *plugin, neu_reqresp_type_e event, json_req.add_grp.node = add_grp->driver; json_req.add_grp.group = add_grp->group; json_req.add_grp.interval = add_grp->interval; - encode_fn = neu_json_encode_add_group_config_req; *topic_p = plugin->config->group_add_topic; + neu_json_encode_by_fn(&json_req, neu_json_encode_add_group_config_req, + &json_str); break; } case NEU_REQ_DEL_GROUP_EVENT: { neu_req_del_group_t *del_grp = data; json_req.del_grp.node = del_grp->driver; json_req.del_grp.group = del_grp->group; - encode_fn = neu_json_encode_del_group_config_req; *topic_p = plugin->config->group_del_topic; + neu_json_encode_by_fn(&json_req, neu_json_encode_del_group_config_req, + &json_str); break; } case NEU_REQ_UPDATE_GROUP_EVENT: { @@ -128,8 +136,61 @@ static char *generate_event_json(neu_plugin_t *plugin, neu_reqresp_type_e event, json_req.update_grp.node = update_grp->driver; json_req.update_grp.group = update_grp->group; json_req.update_grp.interval = update_grp->interval; - encode_fn = neu_json_encode_add_group_config_req; - *topic_p = plugin->config->group_update_topic; + *topic_p = plugin->config->group_update_topic; + neu_json_encode_by_fn(&json_req, neu_json_encode_add_group_config_req, + &json_str); + break; + } + case NEU_REQ_ADD_TAG_EVENT: + case NEU_REQ_UPDATE_TAG_EVENT: { + neu_req_add_tag_t *add_tag = data; + json_req.add_tags.node = add_tag->driver; + json_req.add_tags.group = add_tag->group; + json_req.add_tags.n_tag = add_tag->n_tag; + json_req.add_tags.tags = + calloc(add_tag->n_tag, sizeof(*json_req.add_tags.tags)); + if (NULL == json_req.add_tags.tags) { + plog_error(plugin, "calloc fail"); + break; + } + for (uint16_t i = 0; i < add_tag->n_tag; ++i) { + json_req.add_tags.tags[i].type = add_tag->tags[i].type; + json_req.add_tags.tags[i].name = add_tag->tags[i].name; + json_req.add_tags.tags[i].attribute = add_tag->tags[i].attribute; + json_req.add_tags.tags[i].address = add_tag->tags[i].address; + json_req.add_tags.tags[i].decimal = add_tag->tags[i].decimal; + json_req.add_tags.tags[i].precision = add_tag->tags[i].precision; + json_req.add_tags.tags[i].description = + add_tag->tags[i].description; + } + *topic_p = NEU_REQ_ADD_TAG_EVENT == event + ? plugin->config->tag_add_topic + : plugin->config->tag_update_topic; + neu_json_encode_by_fn(&json_req, neu_json_encode_add_tags_req, + &json_str); + + free(json_req.add_tags.tags); + break; + } + case NEU_REQ_DEL_TAG_EVENT: { + neu_req_del_tag_t *del_tag = data; + json_req.del_tags.node = del_tag->driver; + json_req.del_tags.group = del_tag->group; + json_req.del_tags.n_tags = del_tag->n_tag; + json_req.del_tags.tags = + calloc(del_tag->n_tag, sizeof(*json_req.del_tags.tags)); + if (NULL == json_req.del_tags.tags) { + plog_error(plugin, "calloc fail"); + break; + } + for (uint16_t i = 0; i < del_tag->n_tag; ++i) { + json_req.del_tags.tags[i] = del_tag->tags[i]; + } + *topic_p = plugin->config->tag_del_topic; + neu_json_encode_by_fn(&json_req, neu_json_encode_del_tags_req, + &json_str); + + free(json_req.del_tags.tags); break; } default: { @@ -139,7 +200,6 @@ static char *generate_event_json(neu_plugin_t *plugin, neu_reqresp_type_e event, } } - neu_json_encode_by_fn(&json_req, encode_fn, &json_str); return json_str; } @@ -252,6 +312,21 @@ int handle_events(neu_plugin_t *plugin, neu_reqresp_type_e event, void *data) end: if (NEU_REQ_NODE_SETTING_EVENT == event) { free(((neu_req_node_setting_t *) data)->setting); + } else if (NEU_REQ_ADD_TAG_EVENT == event || + NEU_REQ_UPDATE_TAG_EVENT == event) { + neu_req_add_tag_t *add_tag = data; + for (uint16_t i = 0; i < add_tag->n_tag; i++) { + free(add_tag->tags[i].address); + free(add_tag->tags[i].name); + free(add_tag->tags[i].description); + } + free(add_tag->tags); + } else if (NEU_REQ_DEL_TAG_EVENT == event) { + neu_req_del_tag_t *del_tag = data; + for (uint16_t i = 0; i < del_tag->n_tag; i++) { + free(del_tag->tags[i]); + } + free(del_tag->tags); } return rv; } diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index ed465fbd..4bd97951 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -407,6 +407,9 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_REQ_ADD_GROUP_EVENT: case NEU_REQ_DEL_GROUP_EVENT: case NEU_REQ_UPDATE_GROUP_EVENT: + case NEU_REQ_ADD_TAG_EVENT: + case NEU_REQ_DEL_TAG_EVENT: + case NEU_REQ_UPDATE_TAG_EVENT: adapter->module->intf_funs->request( adapter->plugin, (neu_reqresp_head_t *) header, &header[1]); break; @@ -647,10 +650,14 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) } } - for (int i = 0; i < cmd->n_tag; i++) { - free(cmd->tags[i]); + if (0 == error.error) { + notify_monitor(adapter, NEU_REQ_DEL_TAG_EVENT, cmd); + } else { + for (uint16_t i = 0; i < cmd->n_tag; i++) { + free(cmd->tags[i]); + } + free(cmd->tags); } - free(cmd->tags); neu_msg_exchange(header); header->type = NEU_RESP_ERROR; @@ -677,18 +684,21 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) } } - if (resp.index) { - // we have added some tags, try to persist - adapter_storage_add_tags(cmd->driver, cmd->group, cmd->tags, - resp.index); - } - - for (int i = 0; i < cmd->n_tag; i++) { + for (uint16_t i = resp.index; 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) { + // we have added some tags, try to persist + adapter_storage_add_tags(cmd->driver, cmd->group, cmd->tags, + resp.index); + cmd->n_tag = resp.index; + notify_monitor(adapter, NEU_REQ_ADD_TAG_EVENT, cmd); + } else { + free(cmd->tags); + } neu_msg_exchange(header); header->type = NEU_RESP_ADD_TAG; @@ -717,12 +727,18 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) } } } - for (int i = 0; i < cmd->n_tag; i++) { + + for (uint16_t i = resp.index; 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) { + cmd->n_tag = resp.index; + notify_monitor(adapter, NEU_REQ_UPDATE_TAG_EVENT, cmd); + } else { + free(cmd->tags); + } neu_msg_exchange(header); header->type = NEU_RESP_UPDATE_TAG; @@ -1148,6 +1164,7 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data) data_size = sizeof(neu_resp_get_group_t); break; case NEU_REQ_ADD_TAG: + case NEU_REQ_ADD_TAG_EVENT: data_size = sizeof(neu_req_add_tag_t); break; case NEU_RESP_ADD_TAG: @@ -1157,9 +1174,11 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data) data_size = sizeof(neu_resp_update_tag_t); break; case NEU_REQ_UPDATE_TAG: + case NEU_REQ_UPDATE_TAG_EVENT: data_size = sizeof(neu_req_update_tag_t); break; case NEU_REQ_DEL_TAG: + case NEU_REQ_DEL_TAG_EVENT: data_size = sizeof(neu_req_del_tag_t); break; case NEU_REQ_GET_TAG: diff --git a/src/core/manager.c b/src/core/manager.c index 6afead63..e61f3c8c 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -530,7 +530,10 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_REQ_NODE_CTL_EVENT: case NEU_REQ_ADD_GROUP_EVENT: case NEU_REQ_DEL_GROUP_EVENT: - case NEU_REQ_UPDATE_GROUP_EVENT: { + case NEU_REQ_UPDATE_GROUP_EVENT: + case NEU_REQ_ADD_TAG_EVENT: + case NEU_REQ_DEL_TAG_EVENT: + case NEU_REQ_UPDATE_TAG_EVENT: { if (neu_node_manager_find(manager->node_manager, header->receiver)) { forward_msg(manager, msg, header->receiver); } diff --git a/src/parser/neu_json_tag.c b/src/parser/neu_json_tag.c index b099739a..4bebae4e 100644 --- a/src/parser/neu_json_tag.c +++ b/src/parser/neu_json_tag.c @@ -29,6 +29,78 @@ #include "neu_json_tag.h" +int neu_json_encode_add_tags_req(void *json_object, void *param) +{ + int ret = 0; + neu_json_add_tags_req_t *req = param; + + void * tag_array = neu_json_array(); + neu_json_add_tags_req_tag_t *p_tag = req->tags; + for (int i = 0; i < req->n_tag; i++) { + neu_json_elem_t tag_elems[] = { + { + .name = "type", + .t = NEU_JSON_INT, + .v.val_int = p_tag->type, + }, + { + .name = "name", + .t = NEU_JSON_STR, + .v.val_str = p_tag->name, + }, + { + .name = "attribute", + .t = NEU_JSON_INT, + .v.val_int = p_tag->attribute, + }, + { + .name = "precision", + .t = NEU_JSON_INT, + .v.val_int = p_tag->precision, + }, + { + .name = "decimal", + .t = NEU_JSON_DOUBLE, + .v.val_double = p_tag->decimal, + }, + { + .name = "address", + .t = NEU_JSON_STR, + .v.val_str = p_tag->address, + }, + { + .name = "description", + .t = NEU_JSON_STR, + .v.val_str = p_tag->description, + }, + }; + tag_array = neu_json_encode_array(tag_array, tag_elems, + NEU_JSON_ELEM_SIZE(tag_elems)); + p_tag++; + } + + neu_json_elem_t req_elems[] = { { + .name = "node", + .t = NEU_JSON_STR, + .v.val_str = req->node, + }, + { + .name = "group", + .t = NEU_JSON_STR, + .v.val_str = req->group, + }, + { + .name = "tags", + .t = NEU_JSON_OBJECT, + .v.val_object = tag_array, + } }; + + ret = neu_json_encode_field(json_object, req_elems, + NEU_JSON_ELEM_SIZE(req_elems)); + + return ret; +} + int neu_json_decode_add_tags_req(char *buf, neu_json_add_tags_req_t **result) { int ret = 0; @@ -176,6 +248,48 @@ int neu_json_encode_au_tags_resp(void *json_object, void *param) return ret; } +int neu_json_encode_del_tags_req(void *json_object, void *param) +{ + int ret = 0; + neu_json_del_tags_req_t *req = param; + + void * tag_array = neu_json_array(); + neu_json_del_tags_req_name_t *p_name = req->tags; + for (int i = 0; i < req->n_tags; i++) { + neu_json_elem_t tag_elems[] = { + { + .name = NULL, + .t = NEU_JSON_STR, + .v.val_str = *p_name, + }, + }; + tag_array = neu_json_encode_array_value(tag_array, tag_elems, + NEU_JSON_ELEM_SIZE(tag_elems)); + p_name++; + } + + neu_json_elem_t req_elems[] = { { + .name = "node", + .t = NEU_JSON_STR, + .v.val_str = req->node, + }, + { + .name = "group", + .t = NEU_JSON_STR, + .v.val_str = req->group, + }, + { + .name = "tags", + .t = NEU_JSON_OBJECT, + .v.val_object = tag_array, + } }; + + ret = neu_json_encode_field(json_object, req_elems, + NEU_JSON_ELEM_SIZE(req_elems)); + + return ret; +} + int neu_json_decode_del_tags_req(char *buf, neu_json_del_tags_req_t **result) { int ret = 0; diff --git a/src/parser/neu_json_tag.h b/src/parser/neu_json_tag.h index 6ffa0a99..75dd2947 100644 --- a/src/parser/neu_json_tag.h +++ b/src/parser/neu_json_tag.h @@ -48,6 +48,7 @@ typedef struct { neu_json_add_tags_req_tag_t *tags; } neu_json_add_tags_req_t; +int neu_json_encode_add_tags_req(void *json_object, void *param); int neu_json_decode_add_tags_req(char *buf, neu_json_add_tags_req_t **result); void neu_json_decode_add_tags_req_free(neu_json_add_tags_req_t *req); @@ -67,6 +68,7 @@ typedef struct { neu_json_del_tags_req_name_t *tags; } neu_json_del_tags_req_t; +int neu_json_encode_del_tags_req(void *json_object, void *param); int neu_json_decode_del_tags_req(char *buf, neu_json_del_tags_req_t **result); void neu_json_decode_del_tags_req_free(neu_json_del_tags_req_t *req);