diff --git a/plugins/modbus/modbus_tcp.c b/plugins/modbus/modbus_tcp.c index 5a36db44..347fdba4 100644 --- a/plugins/modbus/modbus_tcp.c +++ b/plugins/modbus/modbus_tcp.c @@ -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); diff --git a/plugins/restful/group_config_handle.c b/plugins/restful/group_config_handle.c index b79a0e60..6fcb93ae 100644 --- a/plugins/restful/group_config_handle.c +++ b/plugins/restful/group_config_handle.c @@ -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; } \ No newline at end of file diff --git a/plugins/restful/group_config_handle.h b/plugins/restful/group_config_handle.h index 1c6a58b1..e94019ad 100644 --- a/plugins/restful/group_config_handle.h +++ b/plugins/restful/group_config_handle.h @@ -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 \ No newline at end of file diff --git a/plugins/restful/handle.c b/plugins/restful/handle.c index fba78a1a..52646ad7 100644 --- a/plugins/restful/handle.c +++ b/plugins/restful/handle.c @@ -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); diff --git a/plugins/restful/rest.c b/plugins/restful/rest.c index 77875291..1704c0f5 100644 --- a/plugins/restful/rest.c +++ b/plugins/restful/rest.c @@ -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; } diff --git a/plugins/restful/rw_handle.c b/plugins/restful/rw_handle.c index cf1629a3..e2d84c77 100644 --- a/plugins/restful/rw_handle.c +++ b/plugins/restful/rw_handle.c @@ -1,4 +1,6 @@ +#include #include +#include #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); } \ No newline at end of file diff --git a/plugins/restful/rw_handle.h b/plugins/restful/rw_handle.h index c7bfdebd..0859851b 100644 --- a/plugins/restful/rw_handle.h +++ b/plugins/restful/rw_handle.h @@ -22,7 +22,11 @@ #include +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 \ No newline at end of file diff --git a/src/adapter/adapter.c b/src/adapter/adapter.c index 953183b3..be49fb29 100644 --- a/src/adapter/adapter.c +++ b/src/adapter/adapter.c @@ -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); diff --git a/src/core/message.h b/src/core/message.h index fd2afe4e..43e87700 100644 --- a/src/core/message.h +++ b/src/core/message.h @@ -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; diff --git a/src/parser/neu_json_group_config.c b/src/parser/neu_json_group_config.c index eb7b7c8e..136bc60d 100644 --- a/src/parser/neu_json_group_config.c +++ b/src/parser/neu_json_group_config.c @@ -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); +} \ No newline at end of file diff --git a/src/parser/neu_json_group_config.h b/src/parser/neu_json_group_config.h index a1c45c3e..5912ebca 100644 --- a/src/parser/neu_json_group_config.h +++ b/src/parser/neu_json_group_config.h @@ -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