From f4c4586696a454234ef870f1171456cc46ae1614 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Wed, 2 Aug 2023 16:17:26 +0800 Subject: [PATCH 1/3] feat(core): add message type NEU_REQ_UPDATE_SUBSCRIBE_GROUP --- include/neuron/adapter.h | 14 ++++++++------ src/adapter/adapter.c | 2 ++ 2 files changed, 10 insertions(+), 6 deletions(-) diff --git a/include/neuron/adapter.h b/include/neuron/adapter.h index 2ed0a497..00397384 100644 --- a/include/neuron/adapter.h +++ b/include/neuron/adapter.h @@ -47,6 +47,7 @@ typedef enum neu_reqresp_type { NEU_REQ_SUBSCRIBE_GROUP, NEU_REQ_UNSUBSCRIBE_GROUP, + NEU_REQ_UPDATE_SUBSCRIBE_GROUP, NEU_REQ_GET_SUBSCRIBE_GROUP, NEU_RESP_GET_SUBSCRIBE_GROUP, NEU_REQ_GET_SUB_DRIVER_TAGS, @@ -147,12 +148,13 @@ static const char *neu_reqresp_type_string_t[] = { [NEU_REQ_WRITE_TAG] = "NEU_REQ_WRITE_TAG", [NEU_REQ_WRITE_TAGS] = "NEU_REQ_WRITE_TAGS", - [NEU_REQ_SUBSCRIBE_GROUP] = "NEU_REQ_SUBSCRIBE_GROUP", - [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_SUBSCRIBE_GROUP] = "NEU_REQ_SUBSCRIBE_GROUP", + [NEU_REQ_UNSUBSCRIBE_GROUP] = "NEU_REQ_UNSUBSCRIBE_GROUP", + [NEU_REQ_UPDATE_SUBSCRIBE_GROUP] = "NEU_REQ_UPDATE_SUBSCRIBE_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_NODE_INIT] = "NEU_REQ_NODE_INIT", [NEU_REQ_NODE_UNINIT] = "NEU_REQ_NODE_UNINIT", diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index ba9ac7bb..137e2b16 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -465,6 +465,7 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data) case NEU_RESP_GET_NODES_STATE: case NEU_RESP_GET_NODE_SETTING: case NEU_REQ_SUBSCRIBE_GROUP: + case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: case NEU_REQ_UNSUBSCRIBE_GROUP: case NEU_RESP_READ_GROUP: case NEU_RESP_GET_SUBSCRIBE_GROUP: @@ -1422,6 +1423,7 @@ void *neu_msg_gen(neu_reqresp_head_t *header, void *data) data_size = sizeof(neu_resp_get_tag_t); break; case NEU_REQ_SUBSCRIBE_GROUP: + case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: data_size = sizeof(neu_req_subscribe_t); break; case NEU_REQ_UNSUBSCRIBE_GROUP: From 5f0e6987f7e718ffb352687f3dfd467b55ea1dff Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Wed, 2 Aug 2023 16:19:10 +0800 Subject: [PATCH 2/3] feat(restful): add group subscribe update api --- plugins/restful/group_config_handle.c | 53 +++++++++++++++++++-------- plugins/restful/group_config_handle.h | 1 + plugins/restful/handle.c | 6 +++ 3 files changed, 45 insertions(+), 15 deletions(-) diff --git a/plugins/restful/group_config_handle.c b/plugins/restful/group_config_handle.c index 7d010934..2b8a7ed7 100644 --- a/plugins/restful/group_config_handle.c +++ b/plugins/restful/group_config_handle.c @@ -194,27 +194,50 @@ void handle_get_driver_group_resp(nng_aio * aio, utarray_free(groups->groups); } -void handle_grp_subscribe(nng_aio *aio) +static inline int send_subscribe(nng_aio *aio, neu_reqresp_type_e type, + neu_json_subscribe_req_t *req) { neu_plugin_t *plugin = neu_rest_get_plugin(); + neu_reqresp_head_t header = { + .ctx = aio, + .type = type, + }; + + neu_req_subscribe_t cmd = { 0 }; + strcpy(cmd.app, req->app); + strcpy(cmd.driver, req->driver); + strcpy(cmd.group, req->group); + cmd.params = req->params; // ownership moved + req->params = NULL; + + if (0 != neu_plugin_op(plugin, header, &cmd)) { + return NEU_ERR_IS_BUSY; + } + + return 0; +} + +void handle_grp_subscribe(nng_aio *aio) +{ NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT( aio, neu_json_subscribe_req_t, neu_json_decode_subscribe_req, { - int ret = 0; - neu_req_subscribe_t cmd = { 0 }; - neu_reqresp_head_t header = { 0 }; - header.ctx = aio; - header.type = NEU_REQ_SUBSCRIBE_GROUP; - strcpy(cmd.app, req->app); - strcpy(cmd.driver, req->driver); - strcpy(cmd.group, req->group); - cmd.params = req->params; // ownership moved - req->params = NULL; - ret = neu_plugin_op(plugin, header, &cmd); + int ret = send_subscribe(aio, NEU_REQ_SUBSCRIBE_GROUP, req); if (ret != 0) { - NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, { - neu_http_response(aio, NEU_ERR_IS_BUSY, result_error); - }); + NEU_JSON_RESPONSE_ERROR( + ret, { neu_http_response(aio, ret, result_error); }); + } + }) +} + +void handle_grp_update_subscribe(nng_aio *aio) +{ + NEU_PROCESS_HTTP_REQUEST_VALIDATE_JWT( + aio, neu_json_subscribe_req_t, neu_json_decode_subscribe_req, { + int ret = send_subscribe(aio, NEU_REQ_UPDATE_SUBSCRIBE_GROUP, req); + if (ret != 0) { + NEU_JSON_RESPONSE_ERROR( + ret, { neu_http_response(aio, ret, result_error); }); } }) } diff --git a/plugins/restful/group_config_handle.h b/plugins/restful/group_config_handle.h index 5826079a..b9a62c68 100644 --- a/plugins/restful/group_config_handle.h +++ b/plugins/restful/group_config_handle.h @@ -32,6 +32,7 @@ void handle_get_driver_group_resp(nng_aio * aio, neu_resp_get_driver_group_t *groups); void handle_grp_subscribe(nng_aio *aio); +void handle_grp_update_subscribe(nng_aio *aio); void handle_grp_unsubscribe(nng_aio *aio); void handle_grp_get_subscribe(nng_aio *aio); void handle_grp_get_subscribe_resp(nng_aio * aio, diff --git a/plugins/restful/handle.c b/plugins/restful/handle.c index 29b1e137..0c8a20a9 100644 --- a/plugins/restful/handle.c +++ b/plugins/restful/handle.c @@ -286,6 +286,12 @@ static struct neu_http_handler rest_handlers[] = { .url = "/api/v2/subscribe", .value.handler = handle_grp_subscribe, }, + { + .method = NEU_HTTP_METHOD_PUT, + .type = NEU_HTTP_HANDLER_FUNCTION, + .url = "/api/v2/subscribe", + .value.handler = handle_grp_update_subscribe, + }, { .method = NEU_HTTP_METHOD_DELETE, .type = NEU_HTTP_HANDLER_FUNCTION, From 1b3d383d6415f6dc5248d473ba42315f2bfc7078 Mon Sep 17 00:00:00 2001 From: eeff <33566201+eeff@users.noreply.github.com> Date: Wed, 2 Aug 2023 16:21:47 +0800 Subject: [PATCH 3/3] feat(core): fulfill update subscription params requests --- include/neuron/persist/persist.h | 13 +++++++++++ src/core/manager.c | 20 +++++++++++++++++ src/core/manager_internal.c | 8 +++++++ src/core/manager_internal.h | 3 +++ src/core/storage.c | 12 ++++++++++ src/core/storage.h | 3 +++ src/core/subscribe.c | 38 ++++++++++++++++++++++++++++++++ src/core/subscribe.h | 3 +++ src/persist/persist.c | 11 +++++++++ 9 files changed, 111 insertions(+) diff --git a/include/neuron/persist/persist.h b/include/neuron/persist/persist.h index 49d44939..2c439325 100644 --- a/include/neuron/persist/persist.h +++ b/include/neuron/persist/persist.h @@ -221,6 +221,19 @@ int neu_persister_store_subscription(const char *app_name, const char *group_name, const char *params); +/** + * Update subscriptions. + * @param app_name name of the app node + * @param driver_name name of the driver node + * @param group_name name of the group + * @param params subscription params + * @return 0 on success, non-zero otherwise + */ +int neu_persister_update_subscription(const char *app_name, + const char *driver_name, + const char *group_name, + const char *params); + /** * Load adapter subscriptions. * @param app_name name of the app node diff --git a/src/core/manager.c b/src/core/manager.c index ba0e681b..15bc8f1a 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -701,6 +701,26 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) reply(manager, header, &error); break; } + case NEU_REQ_UPDATE_SUBSCRIBE_GROUP: { + neu_req_subscribe_t *cmd = (neu_req_subscribe_t *) &header[1]; + neu_resp_error_t error = { 0 }; + + error.error = neu_manager_update_subscribe( + manager, cmd->app, cmd->driver, cmd->group, cmd->params); + + if (error.error == NEU_ERR_SUCCESS) { + forward_msg(manager, msg, cmd->app); + manager_storage_update_subscribe(manager, cmd->app, cmd->driver, + cmd->group, cmd->params); + } else { + free(cmd->params); + } + + header->type = NEU_RESP_ERROR; + strcpy(header->receiver, header->sender); + reply(manager, header, &error); + break; + } case NEU_REQ_UNSUBSCRIBE_GROUP: { neu_req_unsubscribe_t *cmd = (neu_req_unsubscribe_t *) &header[1]; neu_resp_error_t error = { 0 }; diff --git a/src/core/manager_internal.c b/src/core/manager_internal.c index 10ea07ee..63237768 100644 --- a/src/core/manager_internal.c +++ b/src/core/manager_internal.c @@ -624,6 +624,14 @@ int neu_manager_subscribe(neu_manager_t *manager, const char *app, return manager_subscribe(manager, app, driver, group, params); } +int neu_manager_update_subscribe(neu_manager_t *manager, const char *app, + const char *driver, const char *group, + const char *params) +{ + return neu_subscribe_manager_update_params(manager->subscribe_manager, app, + driver, group, params); +} + int neu_manager_send_subscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group, const char *params) diff --git a/src/core/manager_internal.h b/src/core/manager_internal.h index 1c63a96c..ea107a2a 100644 --- a/src/core/manager_internal.h +++ b/src/core/manager_internal.h @@ -98,6 +98,9 @@ UT_array *neu_manager_get_driver_group(neu_manager_t *manager); int neu_manager_subscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group, const char *params); +int neu_manager_update_subscribe(neu_manager_t *manager, const char *app, + const char *driver, const char *group, + const char *params); int neu_manager_send_subscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group, const char *params); diff --git a/src/core/storage.c b/src/core/storage.c index 01b6527a..06e608af 100644 --- a/src/core/storage.c +++ b/src/core/storage.c @@ -81,6 +81,18 @@ void manager_storage_subscribe(neu_manager_t *manager, const char *app, } } +void manager_storage_update_subscribe(neu_manager_t *manager, const char *app, + const char *driver, const char *group, + const char *params) +{ + (void) manager; + int rv = neu_persister_update_subscription(app, driver, group, params); + if (0 != rv) { + nlog_error("fail update subscription app:%s driver:%s group:%s", app, + driver, group); + } +} + void manager_storage_unsubscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group) { diff --git a/src/core/storage.h b/src/core/storage.h index e3669b58..0ce6e798 100644 --- a/src/core/storage.h +++ b/src/core/storage.h @@ -32,6 +32,9 @@ void manager_storage_update_node(neu_manager_t *manager, const char *node, void manager_storage_subscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group, const char *params); +void manager_storage_update_subscribe(neu_manager_t *manager, const char *app, + const char *driver, const char *group, + const char *params); void manager_storage_unsubscribe(neu_manager_t *manager, const char *app, const char *driver, const char *group); void manager_storage_add_ndriver_map(neu_manager_t *manager, diff --git a/src/core/subscribe.c b/src/core/subscribe.c index a8e439fd..1b38c4c1 100644 --- a/src/core/subscribe.c +++ b/src/core/subscribe.c @@ -211,6 +211,44 @@ int neu_subscribe_manager_sub(neu_subscribe_mgr_t *mgr, const char *driver, return NEU_ERR_SUCCESS; } +int neu_subscribe_manager_update_params(neu_subscribe_mgr_t *mgr, + const char *app, const char *driver, + const char *group, const char *params) +{ + sub_elem_key_t key = { 0 }; + strncpy(key.driver, driver, sizeof(key.driver)); + strncpy(key.group, group, sizeof(key.group)); + + sub_elem_t *find = NULL; + HASH_FIND(hh, mgr->ss, &key, sizeof(sub_elem_key_t), find); + + if (NULL == find) { + return NEU_ERR_GROUP_NOT_SUBSCRIBE; + } + + neu_app_subscribe_t *app_sub = NULL; + utarray_foreach(find->apps, neu_app_subscribe_t *, sub) + { + if (strcmp(sub->app_name, app) == 0) { + app_sub = sub; + break; + } + } + + if (NULL == app_sub) { + return NEU_ERR_GROUP_NOT_SUBSCRIBE; + } + + char *p = strdup(params); + if (NULL == p) { + return NEU_ERR_EINTERNAL; + } + + free(app_sub->params); + app_sub->params = p; + return NEU_ERR_SUCCESS; +} + int neu_subscribe_manager_unsub(neu_subscribe_mgr_t *mgr, const char *driver, const char *app, const char *group) { diff --git a/src/core/subscribe.h b/src/core/subscribe.h index cdd0535e..63e85fda 100644 --- a/src/core/subscribe.h +++ b/src/core/subscribe.h @@ -64,5 +64,8 @@ int neu_subscribe_manager_update_app_name(neu_subscribe_mgr_t *mgr, int neu_subscribe_manager_update_driver_name(neu_subscribe_mgr_t *mgr, const char * driver, const char * new_name); +int neu_subscribe_manager_update_params(neu_subscribe_mgr_t *mgr, + const char *app, const char *driver, + const char *group, const char *params); #endif diff --git a/src/persist/persist.c b/src/persist/persist.c index bb38a09f..ab7ab8e5 100644 --- a/src/persist/persist.c +++ b/src/persist/persist.c @@ -1066,6 +1066,17 @@ int neu_persister_store_subscription(const char *app_name, app_name, driver_name, group_name, params); } +int neu_persister_update_subscription(const char *app_name, + const char *driver_name, + const char *group_name, + const char *params) +{ + return execute_sql(global_db, + "UPDATE subscriptions SET params=%Q " + "WHERE app_name=%Q AND driver_name=%Q AND group_name=%Q", + params, app_name, driver_name, group_name); +} + static UT_icd subscription_info_icd = { sizeof(neu_persist_subscription_info_t), NULL,