GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-20 15:39:19 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #1438 from eeff/eeff/dev

Browse Source 
feat(restful): add group subscribe update api
This commit is contained in:
fengzeroz
2023-08-02 16:36:07 +08:00
committed by GitHub
parent e7f81c18d2 1b3d383d64
commit 3031a31010
14 changed files with 166 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+8 -6
View File
@@ -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",
include/neuron/persist/persist.h
+13
View File
@@ -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
plugins/restful/group_config_handle.c
+38 -15
View File
@@ -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); });
}
})
}
plugins/restful/group_config_handle.h
+1
View File
@@ -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,
plugins/restful/handle.c
+6
View File
@@ -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,
src/adapter/adapter.c
+2
View File
@@ -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:
src/core/manager.c
+20
View File
@@ -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 };
src/core/manager_internal.c
+8
View File
@@ -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)
src/core/manager_internal.h
+3
View File
@@ -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);
src/core/storage.c
+12
View File
@@ -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)
{
src/core/storage.h
+3
View File
@@ -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,
src/core/subscribe.c
+38
View File
@@ -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)
{
src/core/subscribe.h
+3
View File
@@ -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
src/persist/persist.c
+11
View File
@@ -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,