From 23be6a9e1f3feee45f72894d60f7980ce9fd364a Mon Sep 17 00:00:00 2001 From: fengzero Date: Wed, 24 Aug 2022 05:53:22 +0000 Subject: [PATCH] chore(manager): remove notify app sub/unsub --- include/neuron/adapter.h | 19 ------- plugins/ekuiper/plugin_ekuiper.c | 7 --- plugins/mqtt/mqtt_plugin.c | 17 ------ src/adapter/adapter.c | 24 --------- src/core/manager.c | 9 ---- src/core/manager_internal.c | 88 -------------------------------- src/core/manager_internal.h | 12 ----- src/core/storage.c | 4 -- 8 files changed, 180 deletions(-) diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index eb22ab99..e3c4a9cc 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -50,10 +50,6 @@ typedef enum neu_reqresp_type { NEU_REQ_GET_SUB_DRIVER_TAGS, NEU_RESP_GET_SUB_DRIVER_TAGS, - NEU_REQ_APP_SUBSCRIBE_GROUP, - NEU_RESP_APP_SUBSCRIBE_GROUP, - NEU_REQ_APP_UNSUBSCRIBE_GROUP, - NEU_REQ_NODE_INIT, NEU_REQ_NODE_UNINIT, NEU_RESP_NODE_UNINIT, @@ -113,10 +109,6 @@ static const char *neu_reqresp_type_string_t[] = { [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", - [NEU_REQ_APP_UNSUBSCRIBE_GROUP] = "NEU_REQ_APP_UNSUBSCRIBE_GROUP", - [NEU_REQ_NODE_INIT] = "NEU_REQ_NODE_INIT", [NEU_REQ_NODE_UNINIT] = "NEU_REQ_NODE_UNINIT", [NEU_RESP_NODE_UNINIT] = "NEU_RESP_NODE_UNINIT", @@ -281,17 +273,6 @@ typedef struct neu_resp_get_tag { UT_array *tags; // array neu_datatag_t } neu_resp_get_tag_t; -typedef struct { - char driver[NEU_NODE_NAME_LEN]; - char group[NEU_GROUP_NAME_LEN]; - UT_array *tags; // array neu_datatag_t -} neu_resp_app_subscribe_group_t; - -typedef struct { - char driver[NEU_NODE_NAME_LEN]; - char group[NEU_GROUP_NAME_LEN]; -} neu_req_app_subscribe_group_t, neu_req_app_unsubscribe_group_t; - typedef struct { char app[NEU_NODE_NAME_LEN]; char driver[NEU_NODE_NAME_LEN]; diff --git a/plugins/ekuiper/plugin_ekuiper.c b/plugins/ekuiper/plugin_ekuiper.c index 9daf8de0..c2e29304 100644 --- a/plugins/ekuiper/plugin_ekuiper.c +++ b/plugins/ekuiper/plugin_ekuiper.c @@ -172,13 +172,6 @@ static int ekuiper_plugin_request(neu_plugin_t * plugin, send_data(plugin, trans_data); break; } - case NEU_RESP_APP_SUBSCRIBE_GROUP: { - neu_resp_app_subscribe_group_t *sub = - (neu_resp_app_subscribe_group_t *) data; - - utarray_free(sub->tags); - break; - } default: plog_notice(plugin, "unsupported request type: %d", header->type); break; diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index 60494e87..68b8eef6 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -516,19 +516,6 @@ static neu_err_code_e trans_data(neu_plugin_t *plugin, void *data) return NEU_ERR_SUCCESS; } -static neu_err_code_e subscribe_handle(neu_plugin_t * plugin, - neu_reqresp_head_t *head, void *data) -{ - UNUSED(plugin); - UNUSED(head); - - neu_resp_app_subscribe_group_t *sub = - (neu_resp_app_subscribe_group_t *) data; - utarray_free(sub->tags); - - return NEU_ERR_SUCCESS; -} - static neu_err_code_e node_state_send(neu_plugin_t * plugin, neu_reqresp_head_t *head, void *data) { @@ -585,10 +572,6 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, case NEU_REQRESP_TRANS_DATA: error = trans_data(plugin, data); break; - case NEU_RESP_APP_SUBSCRIBE_GROUP: { - error = subscribe_handle(plugin, head, data); - break; - } case NEU_REQRESP_NODES_STATE: { error = node_state_send(plugin, head, data); break; diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 4ba7dce9..b12e906c 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -277,7 +277,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) 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: case NEU_RESP_GET_NODES_STATE: case NEU_RESP_GET_NODE_SETTING: @@ -291,7 +290,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_RESP_GET_GROUP: case NEU_RESP_ERROR: case NEU_REQRESP_TRANS_DATA: - case NEU_RESP_APP_SUBSCRIBE_GROUP: case NEU_REQRESP_NODES_STATE: adapter->module->intf_funs->request( adapter->plugin, (neu_reqresp_head_t *) header, &header[1]); @@ -387,21 +385,6 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) } break; } - case NEU_REQ_APP_SUBSCRIBE_GROUP: { - neu_req_app_subscribe_group_t *cmd = - (neu_req_app_subscribe_group_t *) &header[1]; - neu_resp_app_subscribe_group_t resp = { 0 }; - - neu_adapter_driver_get_tag((neu_adapter_driver_t *) adapter, cmd->group, - &resp.tags); - neu_msg_exchange(header); - strcpy(resp.driver, cmd->driver); - strcpy(resp.group, cmd->group); - - header->type = NEU_RESP_APP_SUBSCRIBE_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 }; @@ -975,13 +958,6 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data) case NEU_REQ_UPDATE_LICENSE: data_size = sizeof(neu_req_update_license_t); break; - case NEU_RESP_APP_SUBSCRIBE_GROUP: - data_size = sizeof(neu_resp_app_subscribe_group_t); - break; - case NEU_REQ_APP_SUBSCRIBE_GROUP: - 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; diff --git a/src/core/manager.c b/src/core/manager.c index 8440d26b..2b95b841 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -352,8 +352,6 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) if (error.error == NEU_ERR_SUCCESS) { manager_storage_subscribe(manager, cmd->app, cmd->driver, cmd->group); - neu_manager_notify_app_sub(manager, cmd->app, cmd->driver, - cmd->group); } header->type = NEU_RESP_ERROR; @@ -371,8 +369,6 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) if (error.error == NEU_ERR_SUCCESS) { manager_storage_unsubscribe(manager, cmd->app, cmd->driver, cmd->group); - neu_manager_notify_app_unsub(manager, cmd->app, cmd->driver, - cmd->group); } header->type = NEU_RESP_ERROR; @@ -467,8 +463,6 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) reply(manager, header, &e); } else { forward_msg(manager, msg, header->receiver); - neu_manager_notify_app_unsub_update(manager, cmd->driver, - cmd->group); neu_subscribe_manager_remove(manager->subscribe_manager, cmd->driver, cmd->group); } @@ -489,7 +483,6 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) free(cmd->tags); } else { forward_msg(manager, msg, header->receiver); - neu_manager_notify_app_sub_update(manager, cmd->driver, cmd->group); } break; @@ -512,7 +505,6 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) free(cmd->tags); } else { forward_msg(manager, msg, header->receiver); - neu_manager_notify_app_sub_update(manager, cmd->driver, cmd->group); } break; @@ -541,7 +533,6 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_RESP_GET_NODE_STATE: case NEU_RESP_ERROR: case NEU_RESP_READ_GROUP: - case NEU_RESP_APP_SUBSCRIBE_GROUP: forward_msg(manager, msg, header->receiver); break; default: diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c index 50af0ed4..93f8258e 100644 --- a/src/core/manager_internal.c +++ b/src/core/manager_internal.c @@ -198,91 +198,3 @@ int neu_manager_get_node_info(neu_manager_t *manager, const char *name, return -1; } - -void neu_manager_notify_app_sub(neu_manager_t *manager, const char *app, - const char *driver, const char *group) -{ - nng_msg * msg; - neu_reqresp_head_t header = { 0 }; - neu_req_app_subscribe_group_t sub = { 0 }; - nng_pipe pipe = neu_node_manager_get_pipe(manager->node_manager, driver); - - header.type = NEU_REQ_APP_SUBSCRIBE_GROUP; - strcpy(header.sender, app); - strcpy(header.receiver, driver); - - strcpy(sub.driver, driver); - strcpy(sub.group, group); - - msg = neu_msg_gen(&header, (void *) &sub); - nng_msg_set_pipe(msg, pipe); - if (nng_sendmsg(manager->socket, msg, 0) != 0) { - nng_msg_free(msg); - nlog_warn("%s -> %s req app subscribe group send fail", app, driver); - } -} - -void neu_manager_notify_app_unsub(neu_manager_t *manager, const char *app, - const char *driver, const char *group) -{ - nng_msg * msg; - neu_reqresp_head_t header = { 0 }; - neu_req_app_unsubscribe_group_t unsub = { 0 }; - nng_pipe pipe = neu_node_manager_get_pipe(manager->node_manager, app); - - header.type = NEU_REQ_APP_UNSUBSCRIBE_GROUP; - strcpy(header.sender, "manager"); - strcpy(header.receiver, app); - - strcpy(unsub.driver, driver); - strcpy(unsub.group, group); - - msg = neu_msg_gen(&header, (void *) &unsub); - nng_msg_set_pipe(msg, pipe); - if (nng_sendmsg(manager->socket, msg, 0) != 0) { - nng_msg_free(msg); - nlog_warn("manager -> %s req app unsubscribe group send fail", app); - } -} - -void neu_manager_notify_app_unsub_update(neu_manager_t *manager, - const char *driver, const char *group) -{ - UT_array *apps = - neu_subscribe_manager_find(manager->subscribe_manager, driver, group); - if (apps != NULL) { - utarray_foreach(apps, neu_app_subscribe_t *, app) - { - neu_manager_notify_app_unsub(manager, app->app_name, driver, group); - } - utarray_free(apps); - } -} - -void neu_manager_notify_app_unsub_driver_update(neu_manager_t *manager, - const char * driver) -{ - UT_array *apps = neu_subscribe_manager_find_by_driver( - manager->subscribe_manager, driver); - if (apps != NULL) { - utarray_foreach(apps, neu_app_subscribe_t *, app) - { - neu_manager_notify_app_unsub(manager, app->app_name, driver, ""); - } - utarray_free(apps); - } -} - -void neu_manager_notify_app_sub_update(neu_manager_t *manager, - const char *driver, const char *group) -{ - UT_array *apps = - neu_subscribe_manager_find(manager->subscribe_manager, driver, group); - if (apps != NULL) { - utarray_foreach(apps, neu_app_subscribe_t *, app) - { - neu_manager_notify_app_sub(manager, app->app_name, driver, group); - } - utarray_free(apps); - } -} \ No newline at end of file diff --git a/src/core/manager_internal.h b/src/core/manager_internal.h index 2d60ab0b..a2692e40 100644 --- a/src/core/manager_internal.h +++ b/src/core/manager_internal.h @@ -64,16 +64,4 @@ UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app); int neu_manager_get_node_info(neu_manager_t *manager, const char *name, neu_persist_node_info_t *info); - -void neu_manager_notify_app_sub(neu_manager_t *manager, const char *app, - const char *driver, const char *group); -void neu_manager_notify_app_unsub(neu_manager_t *manager, const char *app, - const char *driver, const char *group); -void neu_manager_notify_app_unsub_update(neu_manager_t *manager, - const char *driver, const char *group); -void neu_manager_notify_app_unsub_driver_update(neu_manager_t *manager, - const char * driver); -void neu_manager_notify_app_sub_update(neu_manager_t *manager, - const char *driver, const char *group); - #endif diff --git a/src/core/storage.c b/src/core/storage.c index 9fd860d9..a4021a08 100644 --- a/src/core/storage.c +++ b/src/core/storage.c @@ -152,10 +152,6 @@ int manager_load_subscribe(neu_manager_t *manager) nlog_info("%s load subscription app:%s driver:%s grp:%s", ok_or_err, node->node, info->driver_name, info->group_name); - if (rv == 0) { - neu_manager_notify_app_sub_update( - manager, info->driver_name, info->group_name); - } } utarray_free(sub_infos);