diff --git a/include/neuron/type.h b/include/neuron/type.h index bdffa1c1..1769e97e 100644 --- a/include/neuron/type.h +++ b/include/neuron/type.h @@ -49,6 +49,7 @@ typedef enum { NEU_TYPE_WORD = 16, NEU_TYPE_DWORD = 17, NEU_TYPE_LWORD = 18, + NEU_TYPE_PTR = 19, } neu_type_e; inline static const char *neu_type_string(neu_type_e type) @@ -90,26 +91,35 @@ inline static const char *neu_type_string(neu_type_e type) return "NEU_TYPE_DWORD"; case NEU_TYPE_LWORD: return "NEU_TYPE_LWORD"; + case NEU_TYPE_PTR: + return "NEU_TYPE_PTR"; } return ""; } +typedef struct { + neu_type_e type; // string or bytes + uint16_t length; + uint8_t * ptr; +} neu_value_ptr_t; + #define NEU_VALUE_SIZE 128 typedef union { - bool boolean; - int8_t i8; - uint8_t u8; - int16_t i16; - uint16_t u16; - int32_t i32; - uint32_t u32; - int64_t i64; - uint64_t u64; - float f32; - double d64; - char str[NEU_VALUE_SIZE]; - uint8_t bytes[NEU_VALUE_SIZE]; + bool boolean; + int8_t i8; + uint8_t u8; + int16_t i16; + uint16_t u16; + int32_t i32; + uint32_t u32; + int64_t i64; + uint64_t u64; + float f32; + double d64; + char str[NEU_VALUE_SIZE]; + uint8_t bytes[NEU_VALUE_SIZE]; + neu_value_ptr_t ptr; } neu_value_u; static inline char *neu_value_str(neu_type_e type, neu_value_u value) diff --git a/plugins/ekuiper/json_rw.c b/plugins/ekuiper/json_rw.c index 82e2329a..5c8592d1 100644 --- a/plugins/ekuiper/json_rw.c +++ b/plugins/ekuiper/json_rw.c @@ -109,6 +109,10 @@ int wrap_tag_data(neu_json_read_resp_tag_t * json_tag, json_tag->t = NEU_JSON_STR; json_tag->value.val_str = tag_value->value.value.str; break; + case NEU_TYPE_PTR: + json_tag->t = NEU_JSON_STR; + json_tag->value.val_str = (char *) tag_value->value.value.ptr.ptr; + break; case NEU_TYPE_ERROR: json_tag->t = NEU_JSON_INT; json_tag->value.val_int = tag_value->value.value.i32; diff --git a/plugins/ekuiper/plugin_ekuiper.c b/plugins/ekuiper/plugin_ekuiper.c index 3d854500..a3f20ce6 100644 --- a/plugins/ekuiper/plugin_ekuiper.c +++ b/plugins/ekuiper/plugin_ekuiper.c @@ -301,12 +301,18 @@ static int ekuiper_plugin_request(neu_plugin_t * plugin, break; } case NEU_REQRESP_TRANS_DATA: { + neu_reqresp_trans_data_t *trans_data = data; + if (disconnected) { plog_debug(plugin, "not connected"); - return -1; + } else { + send_data(plugin, trans_data); + } + for (int i = 0; i < trans_data->n_tag; i++) { + if (trans_data->tags[i].value.type == NEU_TYPE_PTR) { + free(trans_data->tags[i].value.value.ptr.ptr); + } } - neu_reqresp_trans_data_t *trans_data = data; - send_data(plugin, trans_data); break; } case NEU_REQ_SUBSCRIBE_GROUP: diff --git a/plugins/mqtt/mqtt_handle.c b/plugins/mqtt/mqtt_handle.c index 4d502764..5aa24a0e 100644 --- a/plugins/mqtt/mqtt_handle.c +++ b/plugins/mqtt/mqtt_handle.c @@ -125,6 +125,10 @@ static int tag_values_to_json(neu_resp_tag_value_meta_t *tags, uint16_t len, json->tags[i].t = NEU_JSON_STR; json->tags[i].value.val_str = tag->value.value.str; break; + case NEU_TYPE_PTR: + json->tags[i].t = NEU_JSON_STR; + json->tags[i].value.val_str = (char *) tag->value.value.ptr.ptr; + break; default: break; } diff --git a/plugins/mqtt/mqtt_plugin.c b/plugins/mqtt/mqtt_plugin.c index d8e2ee3b..93db252b 100644 --- a/plugins/mqtt/mqtt_plugin.c +++ b/plugins/mqtt/mqtt_plugin.c @@ -352,9 +352,18 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head, case NEU_RESP_READ_GROUP: error = handle_read_response(plugin, head->ctx, data); break; - case NEU_REQRESP_TRANS_DATA: + case NEU_REQRESP_TRANS_DATA: { error = handle_trans_data(plugin, data); + + neu_reqresp_trans_data_t *t_data = (neu_reqresp_trans_data_t *) data; + for (int i = 0; i < t_data->n_tag; i++) { + if (t_data->tags[i].value.type == NEU_TYPE_PTR) { + free(t_data->tags[i].value.value.ptr.ptr); + } + } + break; + } case NEU_REQ_SUBSCRIBE_GROUP: error = handle_subscribe_group(plugin, data); break; diff --git a/plugins/restful/rw_handle.c b/plugins/restful/rw_handle.c index 283f4f3f..ceb64ef5 100644 --- a/plugins/restful/rw_handle.c +++ b/plugins/restful/rw_handle.c @@ -265,6 +265,11 @@ void handle_read_resp(nng_aio *aio, neu_resp_read_group_t *resp) api_res.tags[i].value.val_int = resp->tags[i].value.value.i32; api_res.tags[i].error = resp->tags[i].value.value.i32; break; + case NEU_TYPE_PTR: + api_res.tags[i].t = NEU_JSON_STR; + api_res.tags[i].value.val_str = + (char *) resp->tags[i].value.value.ptr.ptr; + break; default: break; } @@ -272,6 +277,9 @@ void handle_read_resp(nng_aio *aio, neu_resp_read_group_t *resp) neu_json_encode_by_fn(&api_res, neu_json_encode_read_resp, &result); for (int i = 0; i < resp->n_tag; i++) { + if (resp->tags[i].value.type == NEU_TYPE_PTR) { + free(resp->tags[i].value.value.ptr.ptr); + } if (api_res.tags[i].n_meta > 0) { free(api_res.tags[i].metas); } diff --git a/src/adapter/driver/cache.c b/src/adapter/driver/cache.c index 253940f1..a21573a1 100644 --- a/src/adapter/driver/cache.c +++ b/src/adapter/driver/cache.c @@ -85,6 +85,12 @@ void neu_driver_cache_destroy(neu_driver_cache_t *cache) HASH_ITER(hh, cache->table, elem, tmp) { HASH_DEL(cache->table, elem); + if (elem->value.type == NEU_TYPE_PTR) { + if (elem->value.value.ptr.ptr != NULL) { + free(elem->value.value.ptr.ptr); + elem->value.value.ptr.ptr = NULL; + } + } free(elem); } nng_mtx_unlock(cache->mtx); @@ -162,6 +168,18 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, elem->changed = true; } break; + case NEU_TYPE_PTR: { + if (elem->value.value.ptr.length != value.value.ptr.length) { + elem->changed = true; + } else { + if (memcmp(elem->value.value.ptr.ptr, value.value.ptr.ptr, + value.value.ptr.length) != 0) { + elem->changed = true; + } + } + + break; + } case NEU_TYPE_FLOAT: if (elem->value.precision == 0) { elem->changed = elem->value.value.f32 != value.value.f32; @@ -193,8 +211,20 @@ void neu_driver_cache_update_change(neu_driver_cache_t *cache, elem->changed = true; } - elem->value.type = value.type; - elem->value.value = value.value; + elem->value.type = value.type; + if (elem->value.type == NEU_TYPE_PTR) { + elem->value.value.ptr.length = value.value.ptr.length; + elem->value.value.ptr.type = value.value.ptr.type; + if (elem->value.value.ptr.ptr != NULL) { + free(elem->value.value.ptr.ptr); + } + elem->value.value.ptr.ptr = calloc(1, value.value.ptr.length); + memcpy(elem->value.value.ptr.ptr, value.value.ptr.ptr, + value.value.ptr.length); + } else { + elem->value.value = value.value; + } + memset(elem->metas, 0, sizeof(neu_tag_meta_t) * NEU_TAG_META_SIZE); for (int i = 0; i < n_meta; i++) { memcpy(&elem->metas[i], &metas[i], sizeof(neu_tag_meta_t)); @@ -264,6 +294,14 @@ int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group, memcpy(value->value.value.str, elem->value.value.str, sizeof(elem->value.value.str)); break; + case NEU_TYPE_PTR: + value->value.value.ptr.length = elem->value.value.ptr.length; + value->value.value.ptr.type = elem->value.value.ptr.type; + value->value.value.ptr.ptr = + calloc(1, elem->value.value.ptr.length); + memcpy(value->value.value.ptr.ptr, elem->value.value.ptr.ptr, + elem->value.value.ptr.length); + break; } for (int i = 0; i < NEU_TAG_META_SIZE; i++) { @@ -334,6 +372,14 @@ int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache, memcpy(value->value.value.str, elem->value.value.str, sizeof(elem->value.value.str)); break; + case NEU_TYPE_PTR: + value->value.value.ptr.length = elem->value.value.ptr.length; + value->value.value.ptr.type = elem->value.value.ptr.type; + value->value.value.ptr.ptr = + calloc(1, elem->value.value.ptr.length); + memcpy(value->value.value.ptr.ptr, elem->value.value.ptr.ptr, + elem->value.value.ptr.length); + break; } for (int i = 0; i < NEU_TAG_META_SIZE; i++) { @@ -402,6 +448,14 @@ int neu_driver_cache_get(neu_driver_cache_t *cache, const char *group, memcpy(value->value.value.str, elem->value.value.str, sizeof(elem->value.value.str)); break; + case NEU_TYPE_PTR: + value->value.value.ptr.length = elem->value.value.ptr.length; + value->value.value.ptr.type = elem->value.value.ptr.type; + value->value.value.ptr.ptr = + calloc(1, elem->value.value.ptr.length); + memcpy(value->value.value.ptr.ptr, elem->value.value.ptr.ptr, + elem->value.value.ptr.length); + break; } for (int i = 0; i < NEU_TAG_META_SIZE; i++) { @@ -468,6 +522,13 @@ int neu_driver_cache_get_changed(neu_driver_cache_t *cache, const char *group, memcpy(value->value.value.str, elem->value.value.str, sizeof(elem->value.value.str)); break; + case NEU_TYPE_PTR: + value->value.value.ptr.length = elem->value.value.ptr.length; + value->value.value.ptr.type = elem->value.value.ptr.type; + value->value.value.ptr.ptr = + calloc(1, elem->value.value.ptr.length); + memcpy(value->value.value.ptr.ptr, elem->value.value.ptr.ptr, + elem->value.value.ptr.length); } for (int i = 0; i < NEU_TAG_META_SIZE; i++) { @@ -500,6 +561,12 @@ void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group, if (elem != NULL) { HASH_DEL(cache->table, elem); + if (elem->value.type == NEU_TYPE_PTR) { + if (elem->value.value.ptr.ptr != NULL) { + free(elem->value.value.ptr.ptr); + elem->value.value.ptr.ptr = NULL; + } + } free(elem); } diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 33246b79..c5e89b75 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -1408,7 +1408,15 @@ static int read_report_group(int64_t timestamp, int64_t timeout, datas[index].value.type = NEU_TYPE_ERROR; datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED; } else { - datas[index].value = value.value; + if (value.value.type == NEU_TYPE_PTR) { + datas[index].value.type = NEU_TYPE_PTR; + datas[index].value.value.ptr.length = + value.value.value.ptr.length; + datas[index].value.value.ptr.type = value.value.value.ptr.type; + datas[index].value.value.ptr.ptr = value.value.value.ptr.ptr; + } else { + datas[index].value = value.value; + } if (tag->decimal != 0) { datas[index].value.type = NEU_TYPE_DOUBLE; switch (tag->type) { @@ -1549,7 +1557,15 @@ static void read_group(int64_t timestamp, int64_t timeout, datas[index].value.type = NEU_TYPE_ERROR; datas[index].value.value.i32 = NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED; } else { - datas[index].value = value.value; + if (value.value.type == NEU_TYPE_PTR) { + datas[index].value.type = NEU_TYPE_PTR; + datas[index].value.value.ptr.length = + value.value.value.ptr.length; + datas[index].value.value.ptr.type = value.value.value.ptr.type; + datas[index].value.value.ptr.ptr = value.value.value.ptr.ptr; + } else { + datas[index].value = value.value; + } if (tag->decimal != 0) { datas[index].value.type = NEU_TYPE_DOUBLE; switch (tag->type) { diff --git a/src/connection/mqtt_client.c b/src/connection/mqtt_client.c index 889b835b..bd488506 100644 --- a/src/connection/mqtt_client.c +++ b/src/connection/mqtt_client.c @@ -1141,6 +1141,8 @@ int neu_mqtt_client_set_cache_sync_interval(neu_mqtt_client_t *client, client->retry = interval; } else { rv = -1; + // client->retry = 100; + // rv = 0; } nng_mtx_unlock(client->mtx); diff --git a/src/core/manager.c b/src/core/manager.c index ec88a3d6..8174664c 100644 --- a/src/core/manager.c +++ b/src/core/manager.c @@ -49,10 +49,13 @@ static const char *const url = "inproc://neu_manager"; static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data); static int manager_level_check(void *usr_data); -inline static void reply(neu_manager_t *manager, neu_reqresp_head_t *header, - void *data); +inline static void reply(neu_manager_t *manager, neu_reqresp_head_t *header, + void *data); +inline static nng_msg *trans_data_dup(nng_msg *msg); +inline static void trans_data_free(nng_msg *msg); +typedef nng_msg *(*msg_dup)(nng_msg *msg); inline static void forward_msg_dup(neu_manager_t *manager, nng_msg *msg, - nng_pipe pipe); + nng_pipe pipe, msg_dup dup); inline static void forward_msg(neu_manager_t *manager, nng_msg *msg, const char *node); inline static void notify_monitor(neu_manager_t * manager, @@ -244,12 +247,13 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) if (apps != NULL) { utarray_foreach(apps, neu_app_subscribe_t *, app) { - forward_msg_dup(manager, msg, app->pipe); + forward_msg_dup(manager, msg, app->pipe, trans_data_dup); nlog_debug("forward trans data to pipe: %d", app->pipe.id); } utarray_free(apps); - break; } + + trans_data_free(msg); break; } case NEU_REQ_UPDATE_LICENSE: { @@ -257,7 +261,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) utarray_foreach(pipes, nng_pipe *, pipe) { - forward_msg_dup(manager, msg, *pipe); + forward_msg_dup(manager, msg, *pipe, NULL); nlog_notice("forward license update to pipe: %d", pipe->id); } utarray_free(pipes); @@ -1091,11 +1095,14 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data) } inline static void forward_msg_dup(neu_manager_t *manager, nng_msg *msg, - nng_pipe pipe) + nng_pipe pipe, msg_dup dup) { nng_msg *out_msg; nng_msg_dup(&out_msg, msg); + if (dup != NULL) { + out_msg = dup(out_msg); + } nng_msg_set_pipe(out_msg, pipe); if (nng_sendmsg(manager->socket, out_msg, 0) == 0) { nlog_info("forward msg to pipe %d", pipe.id); @@ -1105,6 +1112,35 @@ inline static void forward_msg_dup(neu_manager_t *manager, nng_msg *msg, } } +inline static nng_msg *trans_data_dup(nng_msg *msg) +{ + neu_reqresp_head_t *header = (neu_reqresp_head_t *) nng_msg_body(msg); + assert(header->type == NEU_REQRESP_TRANS_DATA); + neu_reqresp_trans_data_t *cmd = (neu_reqresp_trans_data_t *) &header[1]; + + for (int i = 0; i < cmd->n_tag; i++) { + if (cmd->tags[i].value.type == NEU_TYPE_PTR) { + uint8_t *ptr = calloc(cmd->tags[i].value.value.ptr.length, 1); + cmd->tags[i].value.value.ptr.ptr = ptr; + } + } + + return msg; +} + +inline static void trans_data_free(nng_msg *msg) +{ + neu_reqresp_head_t *header = (neu_reqresp_head_t *) nng_msg_body(msg); + assert(header->type == NEU_REQRESP_TRANS_DATA); + neu_reqresp_trans_data_t *cmd = (neu_reqresp_trans_data_t *) &header[1]; + + for (int i = 0; i < cmd->n_tag; i++) { + if (cmd->tags[i].value.type == NEU_TYPE_PTR) { + free(cmd->tags[i].value.value.ptr.ptr); + } + } +} + inline static void forward_msg(neu_manager_t *manager, nng_msg *msg, const char *node) {