diff --git a/include/neuron/json/json.h b/include/neuron/json/json.h index 4115d9a9..7df908e6 100644 --- a/include/neuron/json/json.h +++ b/include/neuron/json/json.h @@ -26,6 +26,7 @@ extern "C" { #include #include +#include typedef enum neu_json_type { NEU_JSON_UNDEFINE = 0, @@ -69,6 +70,7 @@ int neu_json_decode_array_size(char *buf, char *child); int neu_json_decode_array(char *buf, char *name, int index, int size, neu_json_elem_t *ele); void *neu_json_decode_new(char *buf); +void *neu_json_decode_newb(char *buf, size_t len); void neu_json_decode_free(void *ob); int neu_json_decode_value(void *object, neu_json_elem_t *ele); diff --git a/plugins/ekuiper/json_rw.c b/plugins/ekuiper/json_rw.c index c05211d3..2f1d298e 100644 --- a/plugins/ekuiper/json_rw.c +++ b/plugins/ekuiper/json_rw.c @@ -19,6 +19,8 @@ #include +#include "neuron/tag.h" + #include "json_rw.h" int wrap_tag_data(neu_json_read_resp_tag_t *json_tag, neu_int_val_t *int_val, @@ -281,3 +283,217 @@ int json_encode_read_resp(void *json_object, void *param) return ret; } + +int json_decode_write_req(char *buf, size_t len, json_write_req_t **result) +{ + int ret = 0; + void * json_obj = NULL; + json_write_req_t *req = calloc(1, sizeof(json_write_req_t)); + if (req == NULL) { + return -1; + } + + json_obj = neu_json_decode_newb(buf, len); + if (NULL == json_obj) { + goto decode_fail; + } + + neu_json_elem_t req_elems[] = { + { + .name = "node_name", + .t = NEU_JSON_STR, + }, + { + .name = "group_name", + .t = NEU_JSON_STR, + }, + { + .name = "tag_name", + .t = NEU_JSON_STR, + }, + { + .name = "value", + .t = NEU_JSON_VALUE, + }, + }; + ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems), + req_elems); + if (ret != 0) { + goto decode_fail; + } + + req->node_name = req_elems[0].v.val_str; + req->group_name = req_elems[1].v.val_str; + req->tag_name = req_elems[2].v.val_str; + req->t = req_elems[3].t; + req->value = req_elems[3].v; + + *result = req; + goto decode_exit; + +decode_fail: + if (req != NULL) { + free(req); + } + ret = -1; + +decode_exit: + if (json_obj != NULL) { + neu_json_decode_free(json_obj); + } + return ret; +} + +void json_decode_write_req_free(json_write_req_t *req) +{ + if (NULL == req) { + return; + } + + free(req->group_name); + free(req->node_name); + free(req->tag_name); + if (req->t == NEU_JSON_STR) { + free(req->value.val_str); + } + + free(req); +} + +int unwrap_write_req_val(json_write_req_t * req, + neu_datatag_table_t *datatag_table, + neu_data_val_t ** data_val) +{ + neu_data_val_t * val = NULL; + union neu_json_value value = req->value; + neu_datatag_t * tag = + neu_datatag_tbl_get_by_name(datatag_table, req->tag_name); + + if (NULL == data_val || tag == NULL) { + return NEU_ERR_EINVAL; + } + + switch (tag->type) { + case NEU_DTYPE_UINT8: + val = neu_dvalue_new(NEU_DTYPE_UINT8); + neu_dvalue_set_uint8(val, (uint8_t) value.val_int); + break; + case NEU_DTYPE_INT8: + val = neu_dvalue_new(NEU_DTYPE_INT8); + neu_dvalue_set_int8(val, (int8_t) value.val_int); + break; + case NEU_DTYPE_UINT16: + val = neu_dvalue_new(NEU_DTYPE_UINT16); + neu_dvalue_set_uint16(val, (uint16_t) value.val_int); + break; + case NEU_DTYPE_INT16: + val = neu_dvalue_new(NEU_DTYPE_INT16); + neu_dvalue_set_int16(val, (int16_t) value.val_int); + break; + case NEU_DTYPE_INT32: + val = neu_dvalue_new(NEU_DTYPE_INT32); + neu_dvalue_set_int32(val, (int32_t) value.val_int); + break; + case NEU_DTYPE_UINT32: + val = neu_dvalue_new(NEU_DTYPE_UINT32); + neu_dvalue_set_uint32(val, (uint32_t) value.val_int); + break; + case NEU_DTYPE_INT64: + val = neu_dvalue_new(NEU_DTYPE_INT64); + neu_dvalue_set_int64(val, (int64_t) value.val_int); + break; + case NEU_DTYPE_UINT64: + val = neu_dvalue_new(NEU_DTYPE_UINT64); + neu_dvalue_set_uint64(val, (uint64_t) value.val_int); + break; + case NEU_DTYPE_FLOAT: + val = neu_dvalue_new(NEU_DTYPE_FLOAT); + neu_dvalue_set_float(val, (float) value.val_double); + break; + case NEU_DTYPE_DOUBLE: + val = neu_dvalue_new(NEU_DTYPE_DOUBLE); + neu_dvalue_set_double(val, value.val_double); + break; + case NEU_DTYPE_BIT: + val = neu_dvalue_new(NEU_DTYPE_BIT); + neu_dvalue_set_bit(val, (uint8_t) value.val_int); + break; + case NEU_DTYPE_BOOL: + val = neu_dvalue_new(NEU_DTYPE_BOOL); + neu_dvalue_set_bool(val, value.val_bool); + break; + case NEU_DTYPE_CSTR: + val = neu_dvalue_new(NEU_DTYPE_CSTR); + neu_dvalue_set_cstr(val, value.val_str); + break; + default: + break; + } + + *data_val = val; + + return 0; +} + +int plugin_send_write_cmd_from_write_req(neu_plugin_t *plugin, uint32_t req_id, + json_write_req_t *write_req) +{ + int rv = 0; + + neu_node_id_t node_id = + neu_plugin_get_node_id_by_node_name(plugin, write_req->node_name); + if (node_id == 0) { + log_error("node %s does not exist", write_req->node_name); + return NEU_ERR_NODE_NOT_EXIST; + } + + neu_datatag_table_t *table = neu_system_get_datatags_table(plugin, node_id); + neu_datatag_t * tag = + neu_datatag_tbl_get_by_name(table, write_req->tag_name); + if (tag == NULL) { + return NEU_ERR_TAG_NOT_EXIST; + } + + neu_taggrp_config_t *config = + neu_system_find_group_config(plugin, node_id, write_req->group_name); + if (NULL == config) { + return NEU_ERR_GRP_CONFIG_NOT_EXIST; + } + + neu_data_val_t *write_val = neu_dvalue_unit_new(); + if (write_val == NULL) { + log_error("Failed to allocate data value for write data"); + neu_taggrp_cfg_free(config); + return NEU_ERR_ENOMEM; + } + + neu_fixed_array_t *array = neu_fixed_array_new(1, sizeof(neu_int_val_t)); + if (array == NULL) { + log_error("Failed to allocate array for write data"); + neu_dvalue_free(write_val); + neu_taggrp_cfg_free(config); + return NEU_ERR_ENOMEM; + } + + neu_data_val_t *val = NULL; + rv = unwrap_write_req_val(write_req, table, &val); + if (rv < 0) { + neu_fixed_array_free(array); + neu_dvalue_free(write_val); + neu_taggrp_cfg_free(config); + return rv; + } + + neu_datatag_id_t id = tag->id; + + neu_int_val_t int_val; + neu_int_val_init(&int_val, id, val); + neu_fixed_array_set(array, 0, (void *) &int_val); + + neu_dvalue_init_move_array(write_val, NEU_DTYPE_INT_VAL, array); + + neu_plugin_send_write_cmd(plugin, req_id, node_id, config, write_val); + + neu_taggrp_cfg_free(config); + return rv; +} diff --git a/plugins/ekuiper/json_rw.h b/plugins/ekuiper/json_rw.h index 99385375..7bd7e550 100644 --- a/plugins/ekuiper/json_rw.h +++ b/plugins/ekuiper/json_rw.h @@ -23,7 +23,9 @@ #include #include "neuron.h" +#include "neuron/data_expr.h" #include "neuron/datatag_table.h" +#include "neuron/errcodes.h" #include "neuron/tag_group_config.h" #include "json/neu_json_rw.h" @@ -81,6 +83,23 @@ int json_encode_read_resp_tags(void *json_object, void *param); // } int json_encode_read_resp(void *json_object, void *param); +typedef struct { + char * node_name; + char * group_name; + char * tag_name; + enum neu_json_type t; + union neu_json_value value; +} json_write_req_t; + +int unwrap_write_req_val(json_write_req_t * req, + neu_datatag_table_t *datatag_table, + neu_data_val_t ** data_val); +int json_decode_write_req(char *buf, size_t len, json_write_req_t **result); +void json_decode_write_req_free(json_write_req_t *req); + +int plugin_send_write_cmd_from_write_req(neu_plugin_t *plugin, uint32_t req_id, + json_write_req_t *write_req); + #ifdef __cplusplus } #endif diff --git a/plugins/ekuiper/plugin_ekuiper.c b/plugins/ekuiper/plugin_ekuiper.c index e6a3c82c..e7a57f60 100644 --- a/plugins/ekuiper/plugin_ekuiper.c +++ b/plugins/ekuiper/plugin_ekuiper.c @@ -64,11 +64,18 @@ static int ekuiper_plugin_close(neu_plugin_t *plugin) static int ekuiper_plugin_init(neu_plugin_t *plugin) { - int rv = 0; + int rv = 0; + nng_aio *recv_aio = NULL; - (void) plugin; + rv = nng_aio_alloc(&recv_aio, recv_data_callback, plugin); + if (rv < 0) { + log_error("cannot allocate recv_aio: %s", nng_strerror(rv)); + return rv; + } - log_info("Initialize plugin: %s", neu_plugin_module.module_name); + plugin->recv_aio = recv_aio; + + log_info("Initialized plugin: %s", neu_plugin_module.module_name); return rv; } @@ -76,7 +83,7 @@ static int ekuiper_plugin_uninit(neu_plugin_t *plugin) { int rv = 0; - (void) plugin; + nng_aio_free(plugin->recv_aio); log_info("Uninitialize plugin: %s", neu_plugin_module.module_name); return rv; @@ -97,6 +104,8 @@ static int ekuiper_plugin_start(neu_plugin_t *plugin) return NEU_ERR_FAILURE; } + nng_recv_aio(plugin->sock, plugin->recv_aio); + plugin->common.link_state = NEU_PLUGIN_LINK_STATE_CONNECTED; return NEU_ERR_SUCCESS; } diff --git a/plugins/ekuiper/plugin_ekuiper.h b/plugins/ekuiper/plugin_ekuiper.h index 8f28ce44..35e0393f 100644 --- a/plugins/ekuiper/plugin_ekuiper.h +++ b/plugins/ekuiper/plugin_ekuiper.h @@ -32,6 +32,7 @@ extern "C" { struct neu_plugin { neu_plugin_common_t common; nng_socket sock; + nng_aio * recv_aio; }; #ifdef __cplusplus diff --git a/plugins/ekuiper/read_write.c b/plugins/ekuiper/read_write.c index 28368bee..0efb378b 100644 --- a/plugins/ekuiper/read_write.c +++ b/plugins/ekuiper/read_write.c @@ -17,6 +17,7 @@ * Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. **/ +#include "json/neu_json_rw.h" #include #include "json/neu_json_fn.h" @@ -46,7 +47,7 @@ void send_data(neu_plugin_t *plugin, neu_request_t *req) } nng_msg *msg = NULL; - size_t json_len = strlen(json_str) + 1; // for terminating null byte + size_t json_len = strlen(json_str); rv = nng_msg_alloc(&msg, json_len); if (0 != rv) { log_error("nng cannot allocate msg"); @@ -54,7 +55,7 @@ void send_data(neu_plugin_t *plugin, neu_request_t *req) return; } - memcpy(nng_msg_body(msg), json_str, json_len); + memcpy(nng_msg_body(msg), json_str, json_len); // no null byte rv = nng_sendmsg(plugin->sock, msg, 0); // TODO: use aio to send message if (0 != rv) { log_error("nng cannot send msg"); @@ -63,3 +64,43 @@ void send_data(neu_plugin_t *plugin, neu_request_t *req) free(json_str); } + +void recv_data_callback(void *arg) +{ + int rv = 0; + neu_plugin_t * plugin = arg; + nng_msg * msg = NULL; + char * json_str = NULL; + json_write_req_t *req = NULL; + + rv = nng_aio_result(plugin->recv_aio); + if (0 != rv) { + log_error("receive error: %s", nng_strerror(rv)); + return; + } + + msg = nng_aio_get_msg(plugin->recv_aio); + json_str = nng_msg_body(msg); + // log_info("receive cmd: %s\n", json_str); + if (json_decode_write_req(json_str, nng_msg_len(msg), &req) < 0) { + log_error("cannot decode write request"); + goto recv_data_callback_end; + } + + rv = write_data(plugin, req); + if (0 != rv) { + log_error("failed to write data"); + goto recv_data_callback_end; + } + +recv_data_callback_end: + nng_msg_free(msg); + json_decode_write_req_free(req); + nng_recv_aio(plugin->sock, plugin->recv_aio); +} + +int write_data(neu_plugin_t *plugin, json_write_req_t *write_req) +{ + uint32_t req_id = neu_plugin_get_event_id(plugin); + return plugin_send_write_cmd_from_write_req(plugin, req_id, write_req); +} diff --git a/plugins/ekuiper/read_write.h b/plugins/ekuiper/read_write.h index 855a1f89..7180987c 100644 --- a/plugins/ekuiper/read_write.h +++ b/plugins/ekuiper/read_write.h @@ -22,6 +22,7 @@ #include "neuron.h" +#include "json_rw.h" #include "plugin_ekuiper.h" #ifdef __cplusplus @@ -30,6 +31,9 @@ extern "C" { void send_data(neu_plugin_t *plugin, neu_request_t *req); +void recv_data_callback(void *arg); +int write_data(neu_plugin_t *plugin, json_write_req_t *write_req); + #ifdef __cplusplus } #endif diff --git a/src/utils/json.c b/src/utils/json.c index 69bc4e7c..6d7c5445 100644 --- a/src/utils/json.c +++ b/src/utils/json.c @@ -364,6 +364,21 @@ void *neu_json_decode_new(char *buf) return root; } +void *neu_json_decode_newb(char *buf, size_t len) +{ + json_error_t error; + json_t * root = json_loadb(buf, len, 0, &error); + + if (root == NULL) { + log_error( + "json load error, line: %d, column: %d, position: %d, info: %s", + error.line, error.column, error.position, error.text); + return NULL; + } + + return root; +} + void neu_json_decode_free(void *ob) { json_decref(ob);