GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-27 12:28:44 +08:00
Code Issues Packages Projects Releases Wiki Activity

api: read/write subscribe/unsubscribe

Browse Source
This commit is contained in:
fengzero
2021-10-15 07:20:23 +00:00
parent d24623b401
commit 6eaa2908bf
11 changed files with 386 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/modbus/modbus_tcp.c
+2 -2
View File
@@ -564,7 +564,7 @@ static int modbus_tcp_request(neu_plugin_t *plugin, neu_request_t *req)
data_resp.data_val = array_resp;
}
data_resp.grp_config = read_req->grp_config;
resp.req_id = 1;
resp.req_id = req->req_id;
resp.resp_type = NEU_REQRESP_READ_RESP;
resp.buf = &data_resp;
resp.buf_len = sizeof(neu_reqresp_read_resp_t);
@@ -586,7 +586,7 @@ static int modbus_tcp_request(neu_plugin_t *plugin, neu_request_t *req)
setup_write_resp_data_value(write_req->data_val, plugin);
memset(&resp, 0, sizeof(resp));
resp.req_id = 1;
resp.req_id = req->req_id;
resp.resp_type = NEU_REQRESP_WRITE_RESP;
resp.buf = &data_resp;
resp.buf_len = sizeof(neu_reqresp_data_t);
plugins/restful/group_config_handle.c
+42 -1
View File
@@ -158,10 +158,51 @@ void handle_get_group_config(nng_aio *aio)
void handle_grp_subscribe(nng_aio *aio)
{
(void) aio;
neu_plugin_t *plugin = neu_rest_get_plugin();
REST_PROCESS_HTTP_REQUEST(
aio, neu_parse_subscribe_req_t, neu_parse_decode_subscribe, {
neu_taggrp_config_t *config = neu_system_find_group_config(
plugin, req->src_node_id, req->name);
if (config == NULL) {
http_not_found(aio, "{\"error\" : 1}");
} else {
intptr_t err = neu_plugin_send_subscribe_cmd(
plugin, req->dst_node_id, config);
if (err != 0) {
http_bad_request(aio, "{\"error\": 1}");
} else {
http_ok(aio, "{\"error\": 0}");
}
}
})
}
void handle_grp_unsubscribe(nng_aio *aio)
{
neu_plugin_t *plugin = neu_rest_get_plugin();
REST_PROCESS_HTTP_REQUEST(
aio, neu_parse_unsubscribe_req_t, neu_parse_decode_unsubscribe, {
neu_taggrp_config_t *config = neu_system_find_group_config(
plugin, req->src_node_id, req->name);
if (config == NULL) {
http_not_found(aio, "{\"error\" : 1}");
} else {
intptr_t err = neu_plugin_send_unsubscribe_cmd(
plugin, req->dst_node_id, config);
if (err != 0) {
http_bad_request(aio, "{\"error\": 1}");
} else {
http_ok(aio, "{\"error\": 0}");
}
}
})
}
void handle_grp_get_subscribe(nng_aio *aio)
{
(void) aio;
}
plugins/restful/group_config_handle.h
+1
View File
@@ -27,5 +27,6 @@ void handle_update_group_config(nng_aio *aio);
void handle_get_group_config(nng_aio *aio);
void handle_grp_subscribe(nng_aio *aio);
void handle_grp_unsubscribe(nng_aio *aio);
void handle_grp_get_subscribe(nng_aio *aio);
#endif
plugins/restful/handle.c
+22 -2
View File
@@ -86,6 +86,9 @@ struct neu_rest_handler cors_handler[] = {
{
.url = "/api/v2/write",
},
{
.url = "/api/v2/subscribe",
},
};
struct neu_rest_handler api_handlers[] = {
@@ -223,6 +226,24 @@ struct neu_rest_handler api_handlers[] = {
.url = "/api/v2/write",
.value.handler = handle_write,
},
{
.method = NEU_REST_METHOD_POST,
.type = NEU_REST_HANDLER_FUNCTION,
.url = "/api/v2/subscribe",
.value.handler = handle_grp_subscribe,
},
{
.method = NEU_REST_METHOD_DELETE,
.type = NEU_REST_HANDLER_FUNCTION,
.url = "/api/v2/subscribe",
.value.handler = handle_grp_unsubscribe,
},
{
.method = NEU_REST_METHOD_GET,
.type = NEU_REST_HANDLER_FUNCTION,
.url = "/api/v2/subscribe",
.value.handler = handle_grp_get_subscribe,
},
};
void neu_rest_web_handler(const struct neu_rest_handler **handlers,
@@ -279,8 +300,7 @@ static void cors(nng_aio *aio)
nng_http_res_set_header(res, "Access-Control-Allow-Origin", "*");
nng_http_res_set_header(res, "Access-Control-Allow-Methods",
"POST,GET,PUT,DELETE,OPTIONS");
nng_http_res_set_header(res, "Access-Control-Allow-Headers",
"Content-Type");
nng_http_res_set_header(res, "Access-Control-Allow-Headers", "*");
nng_http_res_copy_data(res, " ", strlen(" "));
nng_http_res_set_status(res, NNG_HTTP_STATUS_OK);
plugins/restful/rest.c
+8 -5
View File
@@ -29,6 +29,7 @@
#include "http.h"
#include "neu_log.h"
#include "rest.h"
#include "rw_handle.h"
#define neu_plugin_module default_dashboard_plugin_module
@@ -181,6 +182,7 @@ static neu_plugin_t *dashb_plugin_open(neu_adapter_t * adapter,
return NULL;
}
handle_rw_init();
log_info("Success to create plugin: %s", neu_plugin_module.module_name);
return plugin;
}
@@ -189,6 +191,7 @@ static int dashb_plugin_close(neu_plugin_t *plugin)
{
int rv = 0;
handle_rw_uninit();
nng_http_server_stop(plugin->api_server);
nng_http_server_release(plugin->api_server);
nng_http_server_stop(plugin->web_server);
@@ -247,12 +250,12 @@ static int dashb_plugin_request(neu_plugin_t *plugin, neu_request_t *req)
(void) adapter_callbacks;
switch (req->req_type) {
case NEU_REQRESP_READ_RESP: {
neu_reqresp_read_resp_t *read_resp =
(neu_reqresp_read_resp_t *) req->buf;
// http_not_found(read_resp->context, "{\"error\" : 1}");
case NEU_REQRESP_READ_RESP:
handle_read_resp(req);
break;
case NEU_REQRESP_WRITE_RESP:
handle_write_resp(req);
break;
}
default:
break;
}
plugins/restful/rw_handle.c
+157 -2
View File
@@ -1,4 +1,6 @@
#include <pthread.h>
#include <stdlib.h>
#include <sys/queue.h>
#include "neu_log.h"
#include "neu_plugin.h"
@@ -10,6 +12,129 @@
#include "rw_handle.h"
struct cmd_ctx {
uint32_t event_id;
nng_aio *aio;
TAILQ_ENTRY(cmd_ctx) node;
};
TAILQ_HEAD(, cmd_ctx) read_cmd_ctxs;
TAILQ_HEAD(, cmd_ctx) write_cmd_ctxs;
static pthread_mutex_t read_ctx_mtx;
static pthread_mutex_t write_ctx_mtx;
static void read_add_ctx(uint32_t event_id, nng_aio *aio);
static void write_add_ctx(uint32_t event_id, nng_aio *aio);
static struct cmd_ctx *read_find_ctx(uint32_t event_id);
static struct cmd_ctx *write_find_ctx(uint32_t event_id);
void handle_rw_init()
{
TAILQ_INIT(&read_cmd_ctxs);
TAILQ_INIT(&write_cmd_ctxs);
pthread_mutex_init(&read_ctx_mtx, NULL);
pthread_mutex_init(&write_ctx_mtx, NULL);
}
void handle_rw_uninit()
{
struct cmd_ctx *ctx = NULL;
pthread_mutex_lock(&read_ctx_mtx);
ctx = TAILQ_FIRST(&read_cmd_ctxs);
while (ctx != NULL) {
TAILQ_REMOVE(&read_cmd_ctxs, ctx, node);
nng_aio_finish(ctx->aio, 0);
free(ctx);
ctx = TAILQ_FIRST(&read_cmd_ctxs);
}
pthread_mutex_unlock(&read_ctx_mtx);
pthread_mutex_lock(&write_ctx_mtx);
ctx = TAILQ_FIRST(&write_cmd_ctxs);
while (ctx != NULL) {
TAILQ_REMOVE(&write_cmd_ctxs, ctx, node);
nng_aio_finish(ctx->aio, 0);
free(ctx);
ctx = TAILQ_FIRST(&write_cmd_ctxs);
}
pthread_mutex_unlock(&write_ctx_mtx);
pthread_mutex_destroy(&read_ctx_mtx);
pthread_mutex_destroy(&write_ctx_mtx);
}
static void read_add_ctx(uint32_t event_id, nng_aio *aio)
{
struct cmd_ctx *ctx = calloc(1, sizeof(struct cmd_ctx));
pthread_mutex_lock(&read_ctx_mtx);
ctx->aio = aio;
ctx->event_id = event_id;
TAILQ_INSERT_TAIL(&read_cmd_ctxs, ctx, node);
pthread_mutex_unlock(&read_ctx_mtx);
}
static void write_add_ctx(uint32_t event_id, nng_aio *aio)
{
struct cmd_ctx *ctx = calloc(1, sizeof(struct cmd_ctx));
pthread_mutex_lock(&write_ctx_mtx);
ctx->aio = aio;
ctx->event_id = event_id;
TAILQ_INSERT_TAIL(&write_cmd_ctxs, ctx, node);
pthread_mutex_unlock(&write_ctx_mtx);
}
static struct cmd_ctx *read_find_ctx(uint32_t event_id)
{
struct cmd_ctx *ctx = NULL;
struct cmd_ctx *ret = NULL;
pthread_mutex_lock(&read_ctx_mtx);
TAILQ_FOREACH(ctx, &read_cmd_ctxs, node)
{
if (ctx->event_id == event_id) {
TAILQ_REMOVE(&read_cmd_ctxs, ctx, node);
ret = ctx;
break;
}
}
pthread_mutex_unlock(&read_ctx_mtx);
return ret;
}
static struct cmd_ctx *write_find_ctx(uint32_t event_id)
{
struct cmd_ctx *ctx = NULL;
struct cmd_ctx *ret = NULL;
pthread_mutex_lock(&write_ctx_mtx);
TAILQ_FOREACH(ctx, &write_cmd_ctxs, node)
{
if (ctx->event_id == event_id) {
TAILQ_REMOVE(&write_cmd_ctxs, ctx, node);
ret = ctx;
break;
}
}
pthread_mutex_unlock(&write_ctx_mtx);
return ret;
}
void handle_read(nng_aio *aio)
{
neu_plugin_t *plugin = neu_rest_get_plugin();
@@ -18,11 +143,41 @@ void handle_read(nng_aio *aio)
aio, neu_parse_read_req_t, neu_parse_decode_read, {
neu_taggrp_config_t *config = neu_system_find_group_config(
plugin, req->node_id, req->group_config_name);
neu_plugin_send_read_cmd(plugin, req->node_id, config);
uint32_t event_id =
neu_plugin_send_read_cmd(plugin, req->node_id, config);
read_add_ctx(event_id, aio);
})
}
void handle_write(nng_aio *aio)
{
(void) aio;
neu_plugin_t *plugin = neu_rest_get_plugin();
REST_PROCESS_HTTP_REQUEST(
aio, neu_parse_write_req_t, neu_parse_decode_write, {
neu_taggrp_config_t *config = neu_system_find_group_config(
plugin, req->node_id, req->group_config_name);
neu_data_val_t *data = neu_parse_write_req_to_val(req);
uint32_t event_id =
neu_plugin_send_write_cmd(plugin, req->node_id, config, data);
write_add_ctx(event_id, aio);
})
}
void handle_read_resp(void *cmd_resp)
{
neu_request_t * req = (neu_request_t *) cmd_resp;
struct cmd_ctx *ctx = read_find_ctx(req->req_id);
http_ok(ctx->aio, "{\"status\": \"OK\"}");
free(ctx);
}
void handle_write_resp(void *cmd_resp)
{
neu_request_t * req = (neu_request_t *) cmd_resp;
struct cmd_ctx *ctx = write_find_ctx(req->req_id);
http_ok(ctx->aio, "{\"status\": \"OK\"}");
free(ctx);
}
plugins/restful/rw_handle.h
+4
View File
@@ -22,7 +22,11 @@
#include <nng/nng.h>
void handle_rw_init();
void handle_rw_uninit();
void handle_read(nng_aio *aio);
void handle_write(nng_aio *aio);
void handle_read_resp(void *cmd_resp);
void handle_write_resp(void *cmd_resp);
#endif
src/adapter/adapter.c
+7 -4
View File
@@ -58,6 +58,7 @@
cmd_ptr->sender_id = to_node_id((adapter), (adapter)->id); \
cmd_ptr->dst_node_id = reqresp_cmd->dst_node_id; \
cmd_ptr->grp_config = reqresp_cmd->grp_config; \
cmd_ptr->req_id = cmd->req_id; \
{ func }; \
nng_sendmsg((adapter)->sock, msg, 0); \
} \
@@ -313,7 +314,7 @@ static void adapter_loop(void *arg)
read_req.dst_node_id = cmd_ptr->dst_node_id;
intf_funs = adapter->plugin_module->intf_funs;
req.req_id = adapter_get_req_id(adapter);
req.req_id = cmd_ptr->req_id;
req.req_type = NEU_REQRESP_READ_DATA;
req.sender_id = cmd_ptr->sender_id;
req.buf_len = sizeof(neu_reqresp_read_t);
@@ -339,7 +340,7 @@ static void adapter_loop(void *arg)
neu_dvalue_deserialize(buf, buf_len, &read_resp.data_val);
intf_funs = adapter->plugin_module->intf_funs;
req.req_id = adapter_get_req_id(adapter);
req.req_id = cmd_ptr->req_id;
req.req_type = NEU_REQRESP_READ_RESP;
req.sender_id = 0;
req.buf_len = sizeof(neu_reqresp_read_resp_t);
@@ -366,7 +367,7 @@ static void adapter_loop(void *arg)
neu_dvalue_deserialize(buf, buf_len, &write_req.data_val);
intf_funs = adapter->plugin_module->intf_funs;
req.req_id = adapter_get_req_id(adapter);
req.req_id = cmd_ptr->req_id;
req.req_type = NEU_REQRESP_WRITE_DATA;
req.sender_id = cmd_ptr->sender_id;
req.buf_len = sizeof(neu_reqresp_write_t);
@@ -394,7 +395,7 @@ static void adapter_loop(void *arg)
neu_dvalue_deserialize(buf, buf_len, &write_resp.data_val);
intf_funs = adapter->plugin_module->intf_funs;
req.req_id = adapter_get_req_id(adapter);
req.req_id = cmd_ptr->req_id;
req.req_type = NEU_REQRESP_WRITE_RESP;
req.sender_id = 0;
req.buf_len = sizeof(neu_reqresp_write_resp_t);
@@ -798,6 +799,7 @@ static int adapter_response(neu_adapter_t *adapter, neu_response_t *resp)
read_data_resp->grp_config = read_resp->grp_config;
read_data_resp->databuf = databuf;
read_data_resp->recver_id = resp->recver_id;
read_data_resp->req_id = resp->req_id;
nng_sendmsg(adapter->sock, read_resp_msg, 0);
}
neu_dvalue_free(read_resp->data_val);
@@ -830,6 +832,7 @@ static int adapter_response(neu_adapter_t *adapter, neu_response_t *resp)
write_data_resp->grp_config = write_resp->grp_config;
write_data_resp->databuf = databuf;
write_data_resp->recver_id = resp->recver_id;
write_data_resp->req_id = resp->req_id;
nng_sendmsg(adapter->sock, write_resp_msg, 0);
}
neu_dvalue_free(write_resp->data_val);
src/core/message.h
+6
View File
@@ -85,6 +85,7 @@ typedef struct read_data_cmd {
neu_taggrp_config_t *grp_config;
adapter_id_t sender_id; // adapter_id of sender
neu_node_id_t dst_node_id;
uint32_t req_id;
} read_data_cmd_t;
/* MSG_CMD_WRITE_DATA */
@@ -93,6 +94,7 @@ typedef struct write_data_cmd {
adapter_id_t sender_id; // adapter_id of sender
neu_node_id_t dst_node_id;
core_databuf_t * databuf;
uint32_t req_id;
} write_data_cmd_t;
/* MSG_CMD_START_PERIODIC_READ */
@@ -114,6 +116,7 @@ typedef struct subscribe_driver_cmd {
neu_taggrp_config_t *grp_config;
adapter_id_t sender_id; // adapter_id of sender
neu_node_id_t dst_node_id;
uint32_t req_id;
} subscribe_node_cmd_t;
/* MSG_CMD_UNSUBSCRIBE_DRIVER */
@@ -121,6 +124,7 @@ typedef struct unsubscribe_driver_cmd {
neu_taggrp_config_t *grp_config;
adapter_id_t sender_id; // adapter_id of sender
neu_node_id_t dst_node_id;
uint32_t req_id;
} unsubscribe_node_cmd_t;
/* MSG_CMD_GROUP_CONFIG_CHANGED */
@@ -148,6 +152,7 @@ typedef struct read_data_resp {
neu_taggrp_config_t *grp_config;
core_databuf_t * databuf;
adapter_id_t recver_id;
uint32_t req_id;
} read_data_resp_t;
/* MSG_DATA_WRITE_RESP */
@@ -155,6 +160,7 @@ typedef struct write_data_resp {
neu_taggrp_config_t *grp_config;
core_databuf_t * databuf;
adapter_id_t recver_id;
uint32_t req_id;
} write_data_resp_t;
typedef struct message message_t;
src/parser/neu_json_group_config.c
+112
View File
@@ -218,3 +218,115 @@ int neu_parse_encode_get_group_config(void *json_object, void *param)
return neu_json_encode_field(json_object, elems, 1);
}
int neu_parse_decode_subscribe(char *buf, neu_parse_subscribe_req_t **result)
{
neu_parse_subscribe_req_t *req = malloc(sizeof(neu_parse_subscribe_req_t));
memset(req, 0, sizeof(neu_parse_subscribe_req_t));
neu_json_elem_t elem[] = {
{
.name = "name",
.t = NEU_JSON_STR,
},
{
.name = "dst_node_id",
.t = NEU_JSON_INT,
},
{
.name = "src_node_id",
.t = NEU_JSON_INT,
},
};
int ret = neu_json_decode(buf, NEU_JSON_ELEM_SIZE(elem), elem);
if (ret != 0) {
free(req);
return -1;
}
req->name = elem[0].v.val_str;
req->dst_node_id = elem[1].v.val_int;
req->src_node_id = elem[2].v.val_int;
*result = req;
return 0;
}
void neu_parse_decode_subscribe_free(neu_parse_subscribe_req_t *req)
{
free(req->name);
free(req);
}
int neu_parse_decode_unsubscribe(char * buf,
neu_parse_unsubscribe_req_t **result)
{
neu_parse_unsubscribe_req_t *req =
malloc(sizeof(neu_parse_unsubscribe_req_t));
memset(req, 0, sizeof(neu_parse_unsubscribe_req_t));
neu_json_elem_t elem[] = {
{
.name = "name",
.t = NEU_JSON_STR,
},
{
.name = "dst_node_id",
.t = NEU_JSON_INT,
},
{
.name = "src_node_id",
.t = NEU_JSON_INT,
},
};
int ret = neu_json_decode(buf, NEU_JSON_ELEM_SIZE(elem), elem);
if (ret != 0) {
free(req);
return -1;
}
req->name = elem[0].v.val_str;
req->dst_node_id = elem[1].v.val_int;
req->src_node_id = elem[2].v.val_int;
*result = req;
return 0;
}
void neu_parse_decode_unsubscribe_free(neu_parse_unsubscribe_req_t *req)
{
free(req->name);
free(req);
}
int neu_parse_encode_get_subscribe(void *json_object, void *param)
{
neu_parse_get_subscribe_res_t *res =
(neu_parse_get_subscribe_res_t *) param;
void *array = NULL;
for (int i = 0; i < res->n_config; i++) {
neu_json_elem_t grp[] = {
{
.name = "group_config_name",
.t = NEU_JSON_STR,
.v.val_str = res->group_configs[i].group_config_name,
},
{
.name = "node_id",
.t = NEU_JSON_INT,
.v.val_int = res->group_configs[i].node_id,
},
};
array = neu_json_encode_array(array, grp, 2);
}
neu_json_elem_t elems[] = { {
.name = "groups",
.t = NEU_JSON_OBJECT,
.v.object = array,
} };
return neu_json_encode_field(json_object, elems, 1);
}
src/parser/neu_json_group_config.h
+25
View File
@@ -80,6 +80,31 @@ int neu_parse_decode_del_group_config(
void neu_parse_decode_del_group_config_free(
neu_parse_del_group_config_req_t *req);
typedef struct {
uint32_t dst_node_id;
uint32_t src_node_id;
char * name;
} neu_parse_subscribe_req_t, neu_parse_unsubscribe_req_t;
int neu_parse_decode_subscribe(char *buf, neu_parse_subscribe_req_t **result);
void neu_parse_decode_subscribe_free(neu_parse_subscribe_req_t *req);
int neu_parse_decode_unsubscribe(char * buf,
neu_parse_unsubscribe_req_t **result);
void neu_parse_decode_unsubscribe_free(neu_parse_unsubscribe_req_t *req);
typedef struct {
uint32_t node_id;
char * group_config_name;
} neu_parse_subscribe_res_grp_t;
typedef struct {
int n_config;
neu_parse_subscribe_res_grp_t *group_configs;
} neu_parse_get_subscribe_res_t;
int neu_parse_encode_get_subscribe(void *json_object, void *param);
#ifdef __cplusplus
}
#endif