diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 883b92cd..0172a4a1 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -98,6 +98,9 @@ typedef enum neu_reqresp_type { NEU_REQ_DEL_NODE_EVENT, NEU_REQ_NODE_CTL_EVENT, NEU_REQ_NODE_SETTING_EVENT, + NEU_REQ_ADD_GROUP_EVENT, + NEU_REQ_DEL_GROUP_EVENT, + NEU_REQ_UPDATE_GROUP_EVENT, } neu_reqresp_type_e; static const char *neu_reqresp_type_string_t[] = { @@ -162,6 +165,9 @@ static const char *neu_reqresp_type_string_t[] = { [NEU_REQ_DEL_NODE_EVENT] = "NEU_REQ_DEL_NODE_EVENT", [NEU_REQ_NODE_CTL_EVENT] = "NEU_REQ_NODE_CTL_EVENT", [NEU_REQ_NODE_SETTING_EVENT] = "NEU_REQ_NODE_SETTING_EVENT", + [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", }; 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 fd82d88b..4c158eeb 100644 --- a/plugins/monitor/monitor.c +++ b/plugins/monitor/monitor.c @@ -397,7 +397,10 @@ static int monitor_plugin_request(neu_plugin_t * plugin, case NEU_REQ_ADD_NODE_EVENT: case NEU_REQ_DEL_NODE_EVENT: case NEU_REQ_NODE_CTL_EVENT: - case NEU_REQ_NODE_SETTING_EVENT: { + case NEU_REQ_NODE_SETTING_EVENT: + case NEU_REQ_ADD_GROUP_EVENT: + case NEU_REQ_DEL_GROUP_EVENT: + case NEU_REQ_UPDATE_GROUP_EVENT: { handle_events(plugin, header->type, data); break; } diff --git a/plugins/monitor/monitor_config.c b/plugins/monitor/monitor_config.c index cc2ea715..14e0b7f6 100644 --- a/plugins/monitor/monitor_config.c +++ b/plugins/monitor/monitor_config.c @@ -125,6 +125,9 @@ static int make_event_topics(monitor_config_t *config, const char *prefix) char *node_del_topic = NULL; char *node_ctl_topic = NULL; char *node_setting_topic = NULL; + char *group_add_topic = NULL; + char *group_del_topic = NULL; + char *group_update_topic = NULL; neu_asprintf(&node_add_topic, "%s/node/add", prefix); if (NULL == node_add_topic) { @@ -146,14 +149,35 @@ static int make_event_topics(monitor_config_t *config, const char *prefix) goto error; } + neu_asprintf(&group_add_topic, "%s/group/add", prefix); + if (NULL == group_add_topic) { + goto error; + } + + neu_asprintf(&group_del_topic, "%s/group/delete", prefix); + if (NULL == group_del_topic) { + goto error; + } + + neu_asprintf(&group_update_topic, "%s/group/update", prefix); + if (NULL == group_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; config->node_setting_topic = node_setting_topic; + config->group_add_topic = group_add_topic; + config->group_del_topic = group_del_topic; + config->group_update_topic = group_update_topic; return 0; error: + free(group_update_topic); + free(group_del_topic); + free(group_add_topic); free(node_setting_topic); free(node_ctl_topic); free(node_del_topic); @@ -299,6 +323,10 @@ int monitor_config_parse(neu_plugin_t *plugin, const char *setting, plog_info(plugin, "node ctl event topic : %s", config->node_ctl_topic); plog_info(plugin, "node setting event topic : %s", config->node_setting_topic); + plog_info(plugin, "group add topic : %s", config->group_add_topic); + plog_info(plugin, "group del topic : %s", config->group_del_topic); + plog_info(plugin, "group update topic : %s", + config->group_update_topic); return 0; @@ -337,6 +365,9 @@ void monitor_config_fini(monitor_config_t *config) free(config->node_del_topic); free(config->node_ctl_topic); free(config->node_setting_topic); + free(config->group_add_topic); + free(config->group_del_topic); + free(config->group_update_topic); memset(config, 0, sizeof(*config)); } diff --git a/plugins/monitor/monitor_config.h b/plugins/monitor/monitor_config.h index 46675f33..8ad5a245 100644 --- a/plugins/monitor/monitor_config.h +++ b/plugins/monitor/monitor_config.h @@ -48,6 +48,9 @@ typedef struct { char *node_del_topic; char *node_ctl_topic; char *node_setting_topic; + char *group_add_topic; + char *group_del_topic; + char *group_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 4ae0d00e..bd29cb5c 100644 --- a/plugins/monitor/mqtt_handle.c +++ b/plugins/monitor/mqtt_handle.c @@ -23,6 +23,7 @@ #include "json/neu_json_mqtt.h" #include "json/neu_json_rw.h" +#include "parser/neu_json_group_config.h" #include "parser/neu_json_node.h" #include "monitor.h" @@ -64,10 +65,13 @@ static char *generate_event_json(neu_plugin_t *plugin, neu_reqresp_type_e event, char *json_str = NULL; int (*encode_fn)(void *json_object, void *param) = NULL; union { - neu_json_add_node_req_t add_node; - neu_json_del_node_req_t del_node; - neu_json_node_ctl_req_t node_ctl; - neu_json_node_setting_req_t node_setting; + neu_json_add_node_req_t add_node; + neu_json_del_node_req_t del_node; + neu_json_node_ctl_req_t node_ctl; + neu_json_node_setting_req_t node_setting; + 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; switch (event) { @@ -102,6 +106,32 @@ static char *generate_event_json(neu_plugin_t *plugin, neu_reqresp_type_e event, *topic_p = plugin->config->node_setting_topic; break; } + case NEU_REQ_ADD_GROUP_EVENT: { + neu_req_add_group_t *add_grp = data; + 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; + 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; + break; + } + case NEU_REQ_UPDATE_GROUP_EVENT: { + neu_req_update_group_t *update_grp = data; + 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; + break; + } default: { plog_error(plugin, "unsupported event:%s", neu_reqresp_type_string(event)); diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 8001563c..ed465fbd 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -404,6 +404,9 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_REQ_DEL_NODE_EVENT: case NEU_REQ_NODE_CTL_EVENT: case NEU_REQ_NODE_SETTING_EVENT: + case NEU_REQ_ADD_GROUP_EVENT: + case NEU_REQ_DEL_GROUP_EVENT: + case NEU_REQ_UPDATE_GROUP_EVENT: adapter->module->intf_funs->request( adapter->plugin, (neu_reqresp_head_t *) header, &header[1]); break; @@ -549,6 +552,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) if (error.error == NEU_ERR_SUCCESS) { adapter_storage_add_group(adapter->name, cmd->group, cmd->interval); + notify_monitor(adapter, NEU_REQ_ADD_GROUP_EVENT, cmd); } neu_msg_exchange(header); @@ -575,6 +579,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) if (error.error == NEU_ERR_SUCCESS) { adapter_storage_update_group(adapter->name, cmd->group, cmd->interval); + notify_monitor(adapter, NEU_REQ_UPDATE_GROUP_EVENT, cmd); } neu_msg_exchange(header); @@ -595,6 +600,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) if (error.error == NEU_ERR_SUCCESS) { adapter_storage_del_group(cmd->driver, cmd->group); + notify_monitor(adapter, NEU_REQ_DEL_GROUP_EVENT, cmd); } neu_msg_exchange(header); @@ -1123,10 +1129,15 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data) data_size = sizeof(neu_resp_get_node_t); break; case NEU_REQ_ADD_GROUP: - case NEU_REQ_UPDATE_GROUP: + case NEU_REQ_ADD_GROUP_EVENT: data_size = sizeof(neu_req_add_group_t); break; + case NEU_REQ_UPDATE_GROUP: + case NEU_REQ_UPDATE_GROUP_EVENT: + data_size = sizeof(neu_req_update_group_t); + break; case NEU_REQ_DEL_GROUP: + case NEU_REQ_DEL_GROUP_EVENT: data_size = sizeof(neu_req_del_group_t); break; case NEU_REQ_GET_DRIVER_GROUP: diff --git a/src/core/manager.c b/src/core/manager.c index 5c2c8a7c..6afead63 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -527,7 +527,10 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_REQ_ADD_NODE_EVENT: case NEU_REQ_DEL_NODE_EVENT: - case NEU_REQ_NODE_CTL_EVENT: { + case NEU_REQ_NODE_CTL_EVENT: + case NEU_REQ_ADD_GROUP_EVENT: + case NEU_REQ_DEL_GROUP_EVENT: + case NEU_REQ_UPDATE_GROUP_EVENT: { if (neu_node_manager_find(manager->node_manager, header->receiver)) { forward_msg(manager, msg, header->receiver); } diff --git a/src/parser/neu_json_group_config.c b/src/parser/neu_json_group_config.c index 084aeec2..817cf25f 100644 --- a/src/parser/neu_json_group_config.c +++ b/src/parser/neu_json_group_config.c @@ -31,6 +31,32 @@ #include "neu_json_group_config.h" +int neu_json_encode_add_group_config_req(void *json_object, void *param) +{ + int ret = 0; + neu_json_add_group_config_req_t *req = param; + + 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 = "interval", + .t = NEU_JSON_INT, + .v.val_int = req->interval, + } }; + ret = neu_json_encode_field(json_object, req_elems, + NEU_JSON_ELEM_SIZE(req_elems)); + + return ret; +} + int neu_json_decode_add_group_config_req( char *buf, neu_json_add_group_config_req_t **result) { @@ -89,6 +115,27 @@ void neu_json_decode_add_group_config_req_free( free(req); } +int neu_json_encode_del_group_config_req(void *json_object, void *param) +{ + int ret = 0; + neu_json_del_group_config_req_t *req = param; + + 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, + } }; + ret = neu_json_encode_field(json_object, req_elems, + NEU_JSON_ELEM_SIZE(req_elems)); + + return ret; +} + int neu_json_decode_del_group_config_req( char *buf, neu_json_del_group_config_req_t **result) { diff --git a/src/parser/neu_json_group_config.h b/src/parser/neu_json_group_config.h index 1d683988..55975a5f 100644 --- a/src/parser/neu_json_group_config.h +++ b/src/parser/neu_json_group_config.h @@ -37,6 +37,7 @@ typedef struct { int64_t interval; } neu_json_add_group_config_req_t, neu_json_update_group_req_t; +int neu_json_encode_add_group_config_req(void *json_object, void *param); int neu_json_decode_add_group_config_req( char *buf, neu_json_add_group_config_req_t **result); void neu_json_decode_add_group_config_req_free( @@ -47,6 +48,7 @@ typedef struct { char *node; } neu_json_del_group_config_req_t; +int neu_json_encode_del_group_config_req(void *json_object, void *param); int neu_json_decode_del_group_config_req( char *buf, neu_json_del_group_config_req_t **result); void neu_json_decode_del_group_config_req_free(