Merge pull request #427 from eeff/eeff/ekuiper

Update ekuiper plugin to handle write request

include/neuron/json/json.h

@@ -26,6 +26,7 @@ extern "C" {
 
 #include <stdbool.h>
 #include <stdint.h>
+#include <stdlib.h>
 
 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);
 

plugins/ekuiper/json_rw.c

@@ -19,6 +19,8 @@
 
 #include <jansson.h>
 
+#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;
+}

plugins/ekuiper/json_rw.h

@@ -23,7 +23,9 @@
 #include <sys/time.h>
 
 #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

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;
 }

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

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 <nng/nng.h>
 
 #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);
+}

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

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);