From 8e8fa443c204315fbd5d86ca2aa7ecfe09064f01 Mon Sep 17 00:00:00 2001 From: fengzero Date: Thu, 2 Jun 2022 07:45:33 +0000 Subject: [PATCH] subscribe/unsubscribe/get subscribe --- include/neuron/adapter_msg.h | 6 +- include/neuron/plugin.h | 12 +- plugins/restful/group_config_handle.c | 78 ++++---- src/adapter/adapter.c | 261 +++++++++----------------- src/adapter/driver/driver.c | 16 ++ src/adapter/driver/driver_internal.h | 2 + src/base/neu_plugin_common.c | 30 ++- src/core/manager.c | 42 +++-- src/core/subscribe.c | 42 ++++- src/core/subscribex.h | 2 + src/persist/json/persist_json_subs.c | 12 +- src/persist/json/persist_json_subs.h | 14 +- src/persist/persist.c | 45 +++-- src/persist/persist.h | 18 +- 14 files changed, 277 insertions(+), 303 deletions(-) diff --git a/include/neuron/adapter_msg.h b/include/neuron/adapter_msg.h index 4725a21d..b679d198 100644 --- a/include/neuron/adapter_msg.h +++ b/include/neuron/adapter_msg.h @@ -110,8 +110,9 @@ typedef struct neu_group_info { } neu_group_info_t; typedef struct neu_subscribe_info { - char node_name[NEU_NODE_NAME_LEN]; - char group_name[NEU_GROUP_NAME_LEN]; + char app[NEU_NODE_NAME_LEN]; + char driver[NEU_NODE_NAME_LEN]; + char group[NEU_GROUP_NAME_LEN]; } neu_subscribe_info_t; /* NEU_REQRESP_READ_DATA */ @@ -306,7 +307,6 @@ typedef struct neu_cmd_get_sub_grp_configs { /* NEU_REQRESP_SUB_GRP_CONFIGS_RESP */ typedef struct neu_reqresp_sub_grp_configs { - int32_t result; UT_array *groups; } neu_reqresp_sub_grp_configs_t; diff --git a/include/neuron/plugin.h b/include/neuron/plugin.h index 292178c0..2b4cf20f 100644 --- a/include/neuron/plugin.h +++ b/include/neuron/plugin.h @@ -116,14 +116,10 @@ intptr_t neu_system_update_group_config(neu_plugin_t *plugin, int neu_system_get_group_configs(neu_plugin_t *plugin, const char *node_name, UT_array **groups); -uint32_t neu_plugin_send_unsubscribe_cmd(neu_plugin_t *plugin, - const char * app_name, - const char * driver_name, - const char * group); -uint32_t neu_plugin_send_subscribe_cmd(neu_plugin_t *plugin, - const char * app_name, - const char * driver_name, - const char * group); +int neu_plugin_send_unsubscribe_cmd(neu_plugin_t *plugin, const char *app_name, + const char *driver_name, const char *group); +int neu_plugin_send_subscribe_cmd(neu_plugin_t *plugin, const char *app_name, + const char *driver_name, const char *group); void neu_plugin_send_read_cmd(neu_plugin_t *plugin, uint32_t event_id, const char *node_name, const char *group_name); diff --git a/plugins/restful/group_config_handle.c b/plugins/restful/group_config_handle.c index d6444c2e..d3ac67f0 100644 --- a/plugins/restful/group_config_handle.c +++ b/plugins/restful/group_config_handle.c @@ -126,13 +126,11 @@ void handle_grp_subscribe(nng_aio *aio) neu_plugin_t *plugin = neu_rest_get_plugin(); REST_PROCESS_HTTP_REQUEST_VALIDATE_JWT( - aio, neu_json_subscribe_req_t, neu_json_decode_subscribe_req, { + aio, neu_json_subscribe_req_t, neu_json_decode_subscribe_req, + { NEU_JSON_RESPONSE_ERROR( neu_plugin_send_subscribe_cmd(plugin, req->app_name, - req->driver_name, req->name); - NEU_JSON_RESPONSE_ERROR(NEU_ERR_SUCCESS, { - http_response(aio, error_code.error, result_error); - }) - }) + req->driver_name, req->name), + { http_response(aio, error_code.error, result_error); }) }) } void handle_grp_unsubscribe(nng_aio *aio) @@ -140,52 +138,50 @@ void handle_grp_unsubscribe(nng_aio *aio) neu_plugin_t *plugin = neu_rest_get_plugin(); REST_PROCESS_HTTP_REQUEST_VALIDATE_JWT( - aio, neu_json_unsubscribe_req_t, neu_json_decode_unsubscribe_req, { + aio, neu_json_unsubscribe_req_t, neu_json_decode_unsubscribe_req, + { NEU_JSON_RESPONSE_ERROR( neu_plugin_send_unsubscribe_cmd(plugin, req->app_name, - req->driver_name, req->name); - NEU_JSON_RESPONSE_ERROR(NEU_ERR_SUCCESS, { - http_response(aio, error_code.error, result_error); - }) - }) + req->driver_name, req->name), + { http_response(aio, error_code.error, result_error); }) }) } void handle_grp_get_subscribe(nng_aio *aio) { - (void) aio; - // neu_plugin_t * plugin = neu_rest_get_plugin(); - // char * result = NULL; - // neu_json_get_subscribe_resp_t sub_grp_configs = { 0 }; - // int index = 0; - // char node_name[NEU_NODE_NAME_LEN] = { 0 }; + neu_plugin_t * plugin = neu_rest_get_plugin(); + char * result = NULL; + neu_json_get_subscribe_resp_t sub_grp_configs = { 0 }; + int index = 0; + char node_name[NEU_NODE_NAME_LEN] = { 0 }; - // VALIDATE_JWT(aio); + VALIDATE_JWT(aio); - // if (http_get_param_str(aio, "node_name", sizeof(node_name)) != 0) { - // NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, { - // http_response(aio, error_code.error, result_error); - //}) - // return; - //} + if (http_get_param_str(aio, "node_name", node_name, sizeof(node_name)) <= + 0) { + NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, { + http_response(aio, error_code.error, result_error); + }) + return; + } - // UT_array *groups = neu_system_get_sub_group_configs(plugin, node_name); + UT_array *groups = neu_system_get_sub_group_configs(plugin, node_name); - // sub_grp_configs.n_group = utarray_len(groups); - // sub_grp_configs.groups = calloc( - // sub_grp_configs.n_group, sizeof(neu_json_get_subscribe_resp_group_t)); + sub_grp_configs.n_group = utarray_len(groups); + sub_grp_configs.groups = calloc( + sub_grp_configs.n_group, sizeof(neu_json_get_subscribe_resp_group_t)); - // utarray_foreach(groups, neu_subscribe_info_t *, group) - //{ - // sub_grp_configs.groups[index].node_name = group->node_name; - // sub_grp_configs.groups[index].group_name = group->group_name; + utarray_foreach(groups, neu_subscribe_info_t *, group) + { + sub_grp_configs.groups[index].node_name = group->driver; + sub_grp_configs.groups[index].group_name = group->group; - // index += 1; - //} + index += 1; + } - // neu_json_encode_by_fn(&sub_grp_configs, - // neu_json_encode_get_subscribe_resp, &result); + neu_json_encode_by_fn(&sub_grp_configs, neu_json_encode_get_subscribe_resp, + &result); - // http_ok(aio, result); - // free(result); - // free(sub_grp_configs.groups); - // utarray_free(groups); + http_ok(aio, result); + free(result); + free(sub_grp_configs.groups); + utarray_free(groups); } \ No newline at end of file diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 4c722a3b..370f3ef2 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -326,18 +326,7 @@ static int persister_singleton_load_subscriptions(neu_adapter_t *adapter, { int rv = 0; neu_persister_t *persister = persister_singleton_get(); - vector_t * sub_infos = NULL; - - neu_node_id_t dst_node_id = 0; - (void) dst_node_id; - - // rv = neu_manager_get_node_id_by_name(adapter->manager, adapter_name, - //&dst_node_id); - if (0 != rv) { - nlog_error("%s fail get node id by name: %s", adapter->name, - adapter_name); - return rv; - } + UT_array * sub_infos = NULL; rv = neu_persister_load_subscriptions(persister, adapter_name, &sub_infos); if (0 != rv) { @@ -352,42 +341,15 @@ static int persister_singleton_load_subscriptions(neu_adapter_t *adapter, return rv; } - VECTOR_FOR_EACH(sub_infos, iter) + utarray_foreach(sub_infos, neu_persist_subscription_info_t *, info) { - neu_persist_subscription_info_t *p = iterator_get(&iter); - - neu_node_id_t src_node_id = 0; - (void) src_node_id; - // rv = neu_manager_get_node_id_by_name(adapter->manager, - // p->src_adapter_name, &src_node_id); - if (0 != rv) { - nlog_error("%s fail get node id by name: %s", adapter->name, - p->src_adapter_name); - break; - } - - neu_taggrp_config_t *grp_config = NULL; - (void) grp_config; - // rv = neu_manager_adapter_get_grp_config_ref_by_name( - // adapter->manager, src_node_id, p->group_config_name, &grp_config); - // if (0 != rv) { - // break; - //} - - // subscribe_node_cmd_t cmd = { - //.grp_config = grp_config, - //.dst_node_id = dst_node_id, - //.src_node_id = src_node_id, - //.sender_id = adapter->id, - //// .req_id = adapter_get_req_id(adapter), - //}; - // rv = neu_manager_subscribe_node(adapter->manager, &cmd); + 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 dst:%s src:%s grp:%s", adapter->name, - ok_or_err, adapter_name, p->src_adapter_name, - p->group_config_name); - - neu_taggrp_cfg_free((neu_taggrp_config_t *) grp_config); + 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); @@ -442,17 +404,14 @@ static int persister_singleton_load_data(neu_adapter_t *adapter) } rv = persister_singleton_load_grp_and_tags(adapter, adapter_info->name); - - // rv = persister_singleton_load_setting(adapter, adapter_info->name, - // node_id); if (0 != rv) { continue; } if (ADAPTER_STATE_RUNNING == adapter_info->state) { // rv = neu_manager_start_adapter_with_id(adapter->manager, - // node_id); if (0 != rv) { nlog_error("%s fail start adapter %s", - // adapter->name, adapter_info->name); + // node_id); if (0 != rv) { nlog_error("%s fail start adapter + // %s", adapter->name, adapter_info->name); //} } } @@ -631,36 +590,18 @@ static int persister_singleton_handle_datatags(neu_adapter_t *adapter, } static int persister_singleton_handle_subscriptions(neu_adapter_t *adapter, - neu_node_id_t node_id) + const char * app) { - neu_persister_t *persister = persister_singleton_get(); - char * adapter_name = NULL; - int rv = 0; - (void) node_id; - (void) adapter; - (void) persister; - // int rv = neu_manager_get_node_name_by_id(adapter->manager, node_id, - //&adapter_name); + 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) { - return rv; + nlog_error("%s fail store adapter:%s subscription infos", app, app); } - // vector_t *sub_infos = NULL; - // rv = neu_manager_get_persist_subscription_infos(adapter->manager, - // node_id, &sub_infos); - // if (0 != rv) { - // nlog_error("%s fail get persist subscription infos", adapter->name); - // free(adapter_name); - // return rv; - //} - - // rv = neu_persister_store_subscriptions(persister, adapter_name, - // sub_infos); if (0 != rv) { nlog_error("%s fail store adapter:%s - // subscription infos", adapter_name, adapter->name); - //} - - // neu_persist_subscription_infos_free(sub_infos); - free(adapter_name); + utarray_free(sub_infos); return rv; } @@ -721,24 +662,6 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd, nlog_debug("Get command from plugin %d, %s", cmd->req_type, adapter->name); switch (cmd->req_type) { - case NEU_REQRESP_SUBSCRIBE_NODE: { - ADAPTER_RESP_CODE(adapter, cmd, intptr_t, subscribe_node_cmd_t, rv, - NEU_REQRESP_ERR_CODE, p_result, { - ret = neu_manager_subscribe( - adapter->manager, req_cmd->app, - req_cmd->driver, req_cmd->group); - }); - break; - } - case NEU_REQRESP_UNSUBSCRIBE_NODE: { - ADAPTER_RESP_CODE(adapter, cmd, intptr_t, unsubscribe_node_cmd_t, rv, - NEU_REQRESP_ERR_CODE, p_result, { - ret = neu_manager_unsubscribe( - adapter->manager, req_cmd->app, - req_cmd->driver, req_cmd->group); - }); - break; - } // case NEU_REQRESP_READ_DATA: { // ADAPTER_SEND_MSG(adapter, cmd, rv, MSG_CMD_READ_DATA, // read_data_cmd_t, neu_reqresp_read_t, {}); @@ -747,7 +670,8 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd, // case NEU_REQRESP_WRITE_DATA: { // ADAPTER_SEND_MSG(adapter, cmd, rv, MSG_CMD_WRITE_DATA, - // write_data_cmd_t, neu_reqresp_write_t, { neu_trans_buf_t *trans_buf; + // write_data_cmd_t, neu_reqresp_write_t, { neu_trans_buf_t + // *trans_buf; // trans_buf = &cmd_ptr->trans_buf; // neu_trans_buf_init(trans_buf, adapter->trans_kind, @@ -755,6 +679,40 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd, //}) // break; //} + case NEU_REQRESP_SUBSCRIBE_NODE: { + ADAPTER_RESP_CODE( + adapter, cmd, intptr_t, subscribe_node_cmd_t, rv, + NEU_REQRESP_ERR_CODE, p_result, { + ret = neu_manager_subscribe(adapter->manager, req_cmd->app, + req_cmd->driver, req_cmd->group); + if (ret == 0) { + subscribe_node_cmd_t event = { 0 }; + event.app = strdup(req_cmd->app); + event.driver = strdup(req_cmd->driver); + event.group = strdup(req_cmd->group); + ADAPTER_SEND_BUF(adapter, rv, MSG_EVENT_SUBSCRIBE_NODE, + &event, sizeof(event)); + } + }); + break; + } + case NEU_REQRESP_UNSUBSCRIBE_NODE: { + ADAPTER_RESP_CODE( + adapter, cmd, intptr_t, unsubscribe_node_cmd_t, rv, + NEU_REQRESP_ERR_CODE, p_result, { + ret = neu_manager_unsubscribe(adapter->manager, req_cmd->app, + req_cmd->driver, req_cmd->group); + if (ret == 0) { + unsubscribe_node_cmd_t event = { 0 }; + event.app = strdup(req_cmd->app); + event.driver = strdup(req_cmd->driver); + event.group = strdup(req_cmd->group); + ADAPTER_SEND_BUF(adapter, rv, MSG_EVENT_UNSUBSCRIBE_NODE, + &event, sizeof(event)); + } + }); + break; + } case NEU_REQRESP_ADD_NODE: { ADAPTER_RESP_CODE(adapter, cmd, intptr_t, neu_cmd_add_node_t, rv, @@ -933,7 +891,8 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd, ret = calloc(1, sizeof(neu_reqresp_node_setting_t)); - // ret->result = neu_manager_adapter_get_setting( + // ret->result = + // neu_manager_adapter_get_setting( // adapter->manager, req_cmd->node_id, // &ret->setting); }); @@ -958,7 +917,8 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd, ADAPTER_RESP_CODE(adapter, cmd, intptr_t, neu_cmd_node_ctl_t, rv, NEU_REQRESP_ERR_CODE, p_result, { ret = 0; - // ret = neu_manager_adapter_ctl(adapter->manager, + // ret = + // neu_manager_adapter_ctl(adapter->manager, // req_cmd->node_id, req_cmd->ctl); // if (0 == ret) { // char *node_name = NULL; @@ -976,17 +936,14 @@ static int adapter_command(neu_adapter_t *adapter, neu_request_t *cmd, } case NEU_REQRESP_GET_SUB_GRP_CONFIGS: { - ADAPTER_RESP_CMD(adapter, cmd, neu_reqresp_sub_grp_configs_t, - neu_cmd_get_sub_grp_configs_t, rv, - NEU_REQRESP_SUB_GRP_CONFIGS_RESP, p_result, { - ret = calloc( - 1, sizeof(neu_reqresp_sub_grp_configs_t)); - - // ret->result = - // neu_manager_adapter_get_sub_grp_configs( - // adapter->manager, req_cmd->node_id, - // &ret->sub_grp_configs); - }); + ADAPTER_RESP_CMD( + adapter, cmd, neu_reqresp_sub_grp_configs_t, + neu_cmd_get_sub_grp_configs_t, rv, NEU_REQRESP_SUB_GRP_CONFIGS_RESP, + p_result, { + ret = calloc(1, sizeof(neu_reqresp_sub_grp_configs_t)); + ret->groups = neu_manager_get_sub_group(adapter->manager, + req_cmd->node_name); + }); break; } @@ -1312,6 +1269,16 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) persister_singleton_load_data(adapter); break; } + case MSG_EVENT_SUBSCRIBE_NODE: + case MSG_EVENT_UNSUBSCRIBE_NODE: { + subscribe_node_cmd_t *cmd = msg_get_buf_ptr(pay_msg); + + persister_singleton_handle_subscriptions(adapter, cmd->app); + free(cmd->app); + free(cmd->driver); + free(cmd->group); + break; + } case MSG_EVENT_ADD_NODE: // fall through @@ -1377,15 +1344,6 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) break; } - case MSG_EVENT_SUBSCRIBE_NODE: - // fall through - - case MSG_EVENT_UNSUBSCRIBE_NODE: { - const neu_node_id_t *node_id_p = msg_get_buf_ptr(pay_msg); - persister_singleton_handle_subscriptions(adapter, *node_id_p); - break; - } - case MSG_EVENT_ADD_TAGS: // fall through @@ -1426,60 +1384,6 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) break; } - // case MSG_CMD_SUBSCRIBE_NODE: { - // subscribe_node_cmd_t *cmd_ptr; - // cmd_ptr = (subscribe_node_cmd_t *) msg_get_buf_ptr(pay_msg); - - // const neu_plugin_intf_funs_t *intf_funs; - // neu_request_t req; - // neu_reqresp_subscribe_node_t sub_node_req; - - // sub_node_req.grp_config = cmd_ptr->grp_config; - // sub_node_req.dst_node_id = cmd_ptr->dst_node_id; - // sub_node_req.src_node_id = cmd_ptr->src_node_id; - - // intf_funs = adapter->plugin_info.module->intf_funs; - // req.req_id = adapter_get_req_id(adapter); - // req.req_type = NEU_REQRESP_SUBSCRIBE_NODE; - // req.sender_id = cmd_ptr->sender_id; - // req.buf_len = sizeof(neu_reqresp_subscribe_node_t); - // req.buf = (void *) &sub_node_req; - // if (adapter->plugin_info.module->type == NEU_NA_TYPE_DRIVER) { - // neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, - //&req); - //} else { - // intf_funs->request(adapter->plugin_info.plugin, &req); - //} - // break; - //} - - // case MSG_CMD_UNSUBSCRIBE_NODE: { - // unsubscribe_node_cmd_t *cmd_ptr; - // cmd_ptr = (unsubscribe_node_cmd_t *) msg_get_buf_ptr(pay_msg); - - // const neu_plugin_intf_funs_t * intf_funs; - // neu_request_t req; - // neu_reqresp_unsubscribe_node_t unsub_node_req; - - // unsub_node_req.grp_config = cmd_ptr->grp_config; - // unsub_node_req.dst_node_id = cmd_ptr->dst_node_id; - // unsub_node_req.src_node_id = cmd_ptr->src_node_id; - - // intf_funs = adapter->plugin_info.module->intf_funs; - // req.req_id = adapter_get_req_id(adapter); - // req.req_type = NEU_REQRESP_UNSUBSCRIBE_NODE; - // req.sender_id = cmd_ptr->sender_id; - // req.buf_len = sizeof(neu_reqresp_unsubscribe_node_t); - // req.buf = (void *) &unsub_node_req; - // if (adapter->plugin_info.module->type == NEU_NA_TYPE_DRIVER) { - // neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, - //&req); - //} else { - // intf_funs->request(adapter->plugin_info.plugin, &req); - //} - // break; - //} - case MSG_CMD_READ_DATA: { // read_data_cmd_t *cmd_ptr; // cmd_ptr = (read_data_cmd_t *) msg_get_buf_ptr(pay_msg); @@ -1497,8 +1401,9 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) // req.sender_id = cmd_ptr->sender_id; // req.buf_len = sizeof(neu_reqresp_read_t); // req.buf = (void *) &read_req; - // if (adapter->plugin_info.module->type == NEU_NA_TYPE_DRIVER) { - // neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, + // if (adapter->plugin_info.module->type == NEU_NA_TYPE_DRIVER) + // { neu_adapter_driver_process_msg((neu_adapter_driver_t *) + // adapter, //&req); //} else { // intf_funs->request(adapter->plugin_info.plugin, &req); @@ -1567,8 +1472,9 @@ int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) // req.sender_id = cmd_ptr->sender_id; // req.buf_len = sizeof(neu_reqresp_write_t); // req.buf = (void *) &write_req; - // if (adapter->plugin_info.module->type == NEU_NA_TYPE_DRIVER) { - // neu_adapter_driver_process_msg((neu_adapter_driver_t *) adapter, + // if (adapter->plugin_info.module->type == NEU_NA_TYPE_DRIVER) + // { neu_adapter_driver_process_msg((neu_adapter_driver_t *) + // adapter, //&req); //} else { // intf_funs->request(adapter->plugin_info.plugin, &req); @@ -1796,7 +1702,8 @@ neu_adapter_id_t neu_adapter_get_id(neu_adapter_t *adapter) return adapter->id; } -// NOTE: Do NOT expose this function in header files, use for persistence only! +// NOTE: Do NOT expose this function in header files, use for persistence +// only! neu_adapter_id_t neu_adapter_set_id(neu_adapter_t *adapter, neu_adapter_id_t id) { if (adapter == NULL) { diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index abea5192..1ee99162 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -451,6 +451,22 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name) return ret; } +int neu_adapter_driver_find_group(neu_adapter_driver_t *driver, + const char * name) +{ + group_t *find = NULL; + int ret = NEU_ERR_GROUP_NOT_EXIST; + + nng_mtx_lock(driver->mtx); + HASH_FIND_STR(driver->groups, name, find); + if (find != NULL) { + ret = NEU_ERR_SUCCESS; + } + + nng_mtx_unlock(driver->mtx); + return ret; +} + UT_array *neu_adapter_driver_get_group(neu_adapter_driver_t *driver) { group_t * el = NULL, *tmp = NULL; diff --git a/src/adapter/driver/driver_internal.h b/src/adapter/driver/driver_internal.h index 3a1b129f..92d5d1e7 100644 --- a/src/adapter/driver/driver_internal.h +++ b/src/adapter/driver/driver_internal.h @@ -38,6 +38,8 @@ int neu_adapter_driver_add_group(neu_adapter_driver_t *driver, const char *name, uint32_t interval); int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char * name); +int neu_adapter_driver_find_group(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, diff --git a/src/base/neu_plugin_common.c b/src/base/neu_plugin_common.c index 016edebf..0d9566af 100644 --- a/src/base/neu_plugin_common.c +++ b/src/base/neu_plugin_common.c @@ -218,38 +218,36 @@ int neu_system_get_group_configs(neu_plugin_t *plugin, const char *node_name, return error; } -uint32_t neu_plugin_send_subscribe_cmd(neu_plugin_t *plugin, - const char * app_name, - const char * driver_name, - const char * group) +int neu_plugin_send_subscribe_cmd(neu_plugin_t *plugin, const char *app_name, + const char *driver_name, const char *group) { - uint32_t event_id = 0; - neu_reqresp_subscribe_node_t req = { 0 }; + int error = 0; + neu_reqresp_subscribe_node_t req = { 0 }; req.app_name = (char *) app_name; req.driver_name = (char *) driver_name; req.group = (char *) group; - PLUGIN_SEND_CMD(plugin, NEU_REQRESP_SUBSCRIBE_NODE, req, event_id) + PLUGIN_CALL_CMD(plugin, NEU_REQRESP_SUBSCRIBE_NODE, req, intptr_t, + { error = (intptr_t) resp; }) - return event_id; + return error; } -uint32_t neu_plugin_send_unsubscribe_cmd(neu_plugin_t *plugin, - const char * app_name, - const char * driver_name, - const char * group) +int neu_plugin_send_unsubscribe_cmd(neu_plugin_t *plugin, const char *app_name, + const char *driver_name, const char *group) { - uint32_t event_id = 0; - neu_reqresp_unsubscribe_node_t req = { 0 }; + int error = 0; + neu_reqresp_unsubscribe_node_t req = { 0 }; req.app_name = (char *) app_name; req.driver_name = (char *) driver_name; req.group = (char *) group; - PLUGIN_SEND_CMD(plugin, NEU_REQRESP_UNSUBSCRIBE_NODE, req, event_id) + PLUGIN_CALL_CMD(plugin, NEU_REQRESP_UNSUBSCRIBE_NODE, req, intptr_t, + { error = (intptr_t) resp; }) - return event_id; + return error; } void neu_plugin_send_read_cmd(neu_plugin_t *plugin, uint32_t event_id, diff --git a/src/core/manager.c b/src/core/manager.c index 1d9ebda8..aa6d9f5e 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -173,23 +173,36 @@ UT_array *neu_manager_get_nodes(neu_manager_t *manager, neu_node_type_e type) int neu_manager_subscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group) { - (void) manager; - (void) app; - (void) driver; - (void) group; + int ret = NEU_ERR_SUCCESS; + nng_pipe pipe = { 0 }; + neu_adapter_t *adapter = + neu_node_manager_find(manager->node_manager, driver); - return 0; + if (adapter == NULL) { + return NEU_ERR_NODE_NOT_EXIST; + } + + ret = + neu_adapter_driver_find_group((neu_adapter_driver_t *) adapter, group); + if (ret != NEU_ERR_SUCCESS) { + return ret; + } + + adapter = neu_node_manager_find(manager->node_manager, app); + if (adapter == NULL) { + return NEU_ERR_NODE_NOT_EXIST; + } + + pipe = neu_node_manager_get_pipe(manager->node_manager, app); + return neu_subscribe_manager_sub(manager->subscribe_manager, driver, app, + group, pipe); } int neu_manager_unsubscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group) { - (void) manager; - (void) app; - (void) driver; - (void) group; - - return 0; + return neu_subscribe_manager_unsub(manager->subscribe_manager, driver, app, + group); } int neu_manager_add_group(neu_manager_t *manager, const char *driver, @@ -300,6 +313,11 @@ int neu_manager_stop_node(neu_manager_t *manager, const char *node) return 0; } +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_adapter_info(neu_manager_t *manager, const char *name, neu_persist_adapter_info_t *info) { @@ -430,6 +448,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) case MSG_EVENT_UPDATE_TAGS: case MSG_EVENT_ADD_PLUGIN: case MSG_EVENT_UPDATE_PLUGIN: + case MSG_EVENT_SUBSCRIBE_NODE: + case MSG_EVENT_UNSUBSCRIBE_NODE: case MSG_EVENT_DEL_PLUGIN: { forward_msg(manager, msg, manager->persist_pipe); nlog_info("forward event %d to %s pipe: %d", msg_get_type(pay_msg), diff --git a/src/core/subscribe.c b/src/core/subscribe.c index 0bc20698..f9da742d 100644 --- a/src/core/subscribe.c +++ b/src/core/subscribe.c @@ -18,13 +18,15 @@ **/ #include +#include "adapter.h" #include "errcodes.h" +#include "utils/log.h" #include "subscribex.h" typedef struct sub_elem_key { - char driver[128]; - char group[128]; + char driver[NEU_NODE_NAME_LEN]; + char group[NEU_GROUP_NAME_LEN]; } sub_elem_key_t; typedef struct sub_elem { @@ -35,6 +37,14 @@ typedef struct sub_elem { UT_hash_handle hh; } sub_elem_t; +typedef struct app_elem { + char name[NEU_NODE_NAME_LEN]; + + UT_array *groups; + + UT_hash_handle hh; +} app_elem_t; + static const UT_icd app_sub_icd = { sizeof(neu_app_subscribe_t), NULL, NULL, NULL }; @@ -81,6 +91,33 @@ UT_array *neu_subscribe_manager_find(neu_subscribe_mgr_t *mgr, } } +UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app) +{ + sub_elem_t *el = NULL, *tmp = NULL; + UT_array * groups = NULL; + UT_icd icd = { sizeof(neu_subscribe_info_t), NULL, NULL, NULL }; + + utarray_new(groups, &icd); + + HASH_ITER(hh, mgr->ss, el, tmp) + { + utarray_foreach(el->apps, neu_app_subscribe_t *, sub_app) + { + if (strcmp(sub_app->app_name, app) == 0) { + neu_subscribe_info_t info = { 0 }; + + strncpy(info.driver, el->key.driver, sizeof(info.driver)); + strncpy(info.app, app, sizeof(info.app)); + strncpy(info.group, el->key.group, sizeof(info.group)); + + utarray_push_back(groups, &info); + } + } + } + + return groups; +} + int neu_subscribe_manager_sub(neu_subscribe_mgr_t *mgr, const char *driver, const char *app, const char *group, nng_pipe pipe) { @@ -106,6 +143,7 @@ int neu_subscribe_manager_sub(neu_subscribe_mgr_t *mgr, const char *driver, find = calloc(1, sizeof(sub_elem_t)); utarray_new(find->apps, &app_sub_icd); find->key = key; + HASH_ADD(hh, mgr->ss, key, sizeof(sub_elem_key_t), find); } utarray_push_back(find->apps, &app_sub); diff --git a/src/core/subscribex.h b/src/core/subscribex.h index 29c94d30..acbbfcbe 100644 --- a/src/core/subscribex.h +++ b/src/core/subscribex.h @@ -22,6 +22,7 @@ #include +#include "define.h" #include "utils/utextend.h" typedef struct neu_subscribe_mgr neu_subscribe_mgr_t; @@ -37,6 +38,7 @@ void neu_subscribe_manager_destroy(neu_subscribe_mgr_t *mgr); // neu_app_subscribe_t array UT_array *neu_subscribe_manager_find(neu_subscribe_mgr_t *mgr, const char *driver, const char *group); +UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app); int neu_subscribe_manager_sub(neu_subscribe_mgr_t *mgr, const char *driver, const char *app, const char *group, nng_pipe pipe); diff --git a/src/persist/json/persist_json_subs.c b/src/persist/json/persist_json_subs.c index c2f43a74..b589c47c 100644 --- a/src/persist/json/persist_json_subs.c +++ b/src/persist/json/persist_json_subs.c @@ -61,10 +61,6 @@ int neu_json_decode_subscriptions_req(char * buf, .name = "src_adapter_name", .t = NEU_JSON_STR, }, - { - .name = "read_interval", - .t = NEU_JSON_INT, - }, { .name = "group_config_name", @@ -79,8 +75,7 @@ int neu_json_decode_subscriptions_req(char * buf, p_subscription->sub_adapter_name = subscription_elems[0].v.val_str; p_subscription->src_adapter_name = subscription_elems[1].v.val_str; - p_subscription->read_interval = subscription_elems[2].v.val_int; - p_subscription->group_config_name = subscription_elems[3].v.val_str; + p_subscription->group_config_name = subscription_elems[2].v.val_str; p_subscription++; } @@ -141,11 +136,6 @@ int neu_json_encode_subscriptions_resp(void *json_object, void *param) .t = NEU_JSON_STR, .v.val_str = p_subscription->src_adapter_name, }, - { - .name = "read_interval", - .t = NEU_JSON_INT, - .v.val_int = p_subscription->read_interval, - }, { .name = "group_config_name", .t = NEU_JSON_STR, diff --git a/src/persist/json/persist_json_subs.h b/src/persist/json/persist_json_subs.h index 72cc43d7..d3dbac6a 100644 --- a/src/persist/json/persist_json_subs.h +++ b/src/persist/json/persist_json_subs.h @@ -32,10 +32,9 @@ extern "C" { #endif typedef struct { - char * group_config_name; - char * sub_adapter_name; - int64_t read_interval; - char * src_adapter_name; + char *group_config_name; + char *sub_adapter_name; + char *src_adapter_name; } neu_json_subscriptions_req_subscription_t; typedef struct { @@ -48,10 +47,9 @@ int neu_json_decode_subscriptions_req(char * buf, void neu_json_decode_subscriptions_req_free(neu_json_subscriptions_req_t *req); typedef struct { - char * group_config_name; - char * sub_adapter_name; - int64_t read_interval; - char * src_adapter_name; + char *group_config_name; + char *sub_adapter_name; + char *src_adapter_name; } neu_json_subscriptions_resp_subscription_t; typedef struct { diff --git a/src/persist/persist.c b/src/persist/persist.c index 324ba2ac..370608c2 100644 --- a/src/persist/persist.c +++ b/src/persist/persist.c @@ -390,7 +390,7 @@ static int write_file_string(const char *fn, const char *s) fclose(f); - nlog_debug("pwersister write %s to %s", s, tmp); + nlog_debug("persister write %s to %s", s, tmp); // rename the temporary file to the destination file if (0 != rename(tmp, fn)) { @@ -964,7 +964,7 @@ int neu_persister_load_datatags(neu_persister_t *persister, int neu_persister_store_subscriptions(neu_persister_t *persister, const char * adapter_name, - vector_t * subscription_infos) + UT_array * subscription_infos) { char path[PATH_MAX_SIZE] = { 0 }; @@ -986,10 +986,19 @@ int neu_persister_store_subscriptions(neu_persister_t *persister, return NEU_ERR_FAILURE; } - neu_json_subscriptions_req_t subs_resp = { - .n_subscription = subscription_infos->size, - .subscriptions = subscription_infos->data, - }; + neu_json_subscriptions_req_t subs_resp = { 0 }; + int index = 0; + subs_resp.n_subscription = utarray_len(subscription_infos); + subs_resp.subscriptions = + calloc(subs_resp.n_subscription, + sizeof(neu_json_subscriptions_req_subscription_t)); + utarray_foreach(subscription_infos, neu_subscribe_info_t *, info) + { + subs_resp.subscriptions[index].group_config_name = info->group; + subs_resp.subscriptions[index].src_adapter_name = info->driver; + subs_resp.subscriptions[index].sub_adapter_name = info->app; + index += 1; + } char *result = NULL; rv = neu_json_encode_by_fn(&subs_resp, neu_json_encode_subscriptions_resp, @@ -999,13 +1008,14 @@ int neu_persister_store_subscriptions(neu_persister_t *persister, } free(result); + free(subs_resp.subscriptions); return rv; } int neu_persister_load_subscriptions(neu_persister_t *persister, const char * adapter_name, - vector_t ** subscription_infos) + UT_array ** subscription_infos) { char path[PATH_MAX_SIZE] = { 0 }; @@ -1034,16 +1044,19 @@ int neu_persister_load_subscriptions(neu_persister_t *persister, return rv; } - vector_t *vec = vector_new_move_from_buf( - subs_req->subscriptions, subs_req->n_subscription, - subs_req->n_subscription, sizeof(neu_persist_subscription_info_t)); + UT_icd icd = { sizeof(neu_persist_subscription_info_t), NULL, NULL, NULL }; + utarray_new(*subscription_infos, &icd); - if (NULL != vec) { - subs_req->n_subscription = 0; - subs_req->subscriptions = NULL; - *subscription_infos = vec; - } else { - rv = NEU_ERR_ENOMEM; + for (int i = 0; i < subs_req->n_subscription; i++) { + neu_persist_subscription_info_t info = { 0 }; + info.group_config_name = + strdup(subs_req->subscriptions[i].group_config_name); + info.src_adapter_name = + strdup(subs_req->subscriptions[i].src_adapter_name); + info.sub_adapter_name = + strdup(subs_req->subscriptions[i].sub_adapter_name); + + utarray_push_back(*subscription_infos, &info); } neu_json_decode_subscriptions_req_free(subs_req); diff --git a/src/persist/persist.h b/src/persist/persist.h index efbfdac0..695fe8b2 100644 --- a/src/persist/persist.h +++ b/src/persist/persist.h @@ -90,17 +90,15 @@ static inline void neu_persist_datatag_infos_free(vector_t *datatag_infos) } static inline void -neu_persist_subscription_infos_free(vector_t *subscription_infos) +neu_persist_subscription_infos_free(UT_array *subscription_infos) { - VECTOR_FOR_EACH(subscription_infos, iter) + utarray_foreach(subscription_infos, neu_persist_subscription_info_t *, info) { - neu_persist_subscription_info_t *p = - (neu_persist_subscription_info_t *) iterator_get(&iter); - free(p->src_adapter_name); - free(p->sub_adapter_name); - free(p->group_config_name); + free(info->group_config_name); + free(info->src_adapter_name); + free(info->sub_adapter_name); } - vector_free(subscription_infos); + utarray_free(subscription_infos); } /** @@ -213,7 +211,7 @@ int neu_persister_load_datatags(neu_persister_t *persister, */ int neu_persister_store_subscriptions(neu_persister_t *persister, const char * adapter_name, - vector_t * subscription_infos); + UT_array * subscription_infos); /** * Load adapter subscription infos. * @param persister persiter object. @@ -225,7 +223,7 @@ int neu_persister_store_subscriptions(neu_persister_t *persister, */ int neu_persister_load_subscriptions(neu_persister_t *persister, const char * adapter_name, - vector_t ** subscription_infos); + UT_array ** subscription_infos); /** * Persist group config.