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

Merge pull request #1468 from fengzeroz/main

Browse Source 
core: neu_type_ptr
This commit is contained in:
fengzeroz
2023-08-24 14:27:27 +08:00
committed by GitHub
parent 769c7b9997 5c0c6f6e06
commit 3d874cd3af
10 changed files with 190 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/type.h
+23 -13
View File
@@ -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)
plugins/ekuiper/json_rw.c
+4
View File
@@ -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;
plugins/ekuiper/plugin_ekuiper.c
+9 -3
View File
@@ -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:
plugins/mqtt/mqtt_handle.c
+4
View File
@@ -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;
}
plugins/mqtt/mqtt_plugin.c
+10 -1
View File
@@ -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;
plugins/restful/rw_handle.c
+8
View File
@@ -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);
}
src/adapter/driver/cache.c
+69 -2
View File
@@ -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);
}
src/adapter/driver/driver.c
+18 -2
View File
@@ -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) {
src/connection/mqtt_client.c
+2
View File
@@ -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);
src/core/manager.c
+43 -7
View File
@@ -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)
{